Files
vultr-api/vultr_api/resources/dns.py
HWANG BYUNGHA 184054c6c1 Initial commit: Vultr API v2 Python wrapper with FastAPI server
- vultr_api/: Python library wrapping Vultr API v2
  - 17 resource modules (instances, dns, firewall, vpc, etc.)
  - Pagination support, error handling

- server/: FastAPI REST server
  - All API endpoints exposed via HTTP
  - X-API-Key header authentication
  - Swagger docs at /docs

- Podman quadlet config for systemd deployment

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-22 01:08:17 +09:00

349 lines
8.7 KiB
Python

"""
DNS Resource
DNS domain and record management
"""
from typing import Dict, List, Optional
from .base import BaseResource
class DNSResource(BaseResource):
"""
DNS domain and record management
Usage:
# List domains
domains = client.dns.list_domains()
# Create domain
domain = client.dns.create_domain("example.com")
# List records
records = client.dns.list_records("example.com")
# Create A record
record = client.dns.create_record(
domain="example.com",
name="www",
record_type="A",
data="192.168.1.1",
ttl=300
)
"""
# Domain management
def list_domains(self, per_page: int = 100, cursor: str = None) -> Dict:
"""
List all DNS domains
Args:
per_page: Number of items per page
cursor: Pagination cursor
Returns:
Dict with 'domains' list and 'meta' pagination
"""
params = {"per_page": per_page}
if cursor:
params["cursor"] = cursor
return self.client.get("domains", params=params)
def list_all_domains(self) -> List[Dict]:
"""List all domains (auto-paginate)"""
return self.client.paginate("domains", "domains")
def get_domain(self, domain: str) -> Dict:
"""
Get domain details
Args:
domain: Domain name (e.g., "example.com")
Returns:
Domain details
"""
response = self.client.get(f"domains/{domain}")
return response.get("domain", {})
def create_domain(
self,
domain: str,
ip: str = None,
dns_sec: str = "disabled"
) -> Dict:
"""
Create a DNS domain
Args:
domain: Domain name
ip: Default IP address for A record (optional)
dns_sec: DNSSEC status ("enabled" or "disabled")
Returns:
Created domain details
"""
data = {"domain": domain, "dns_sec": dns_sec}
if ip:
data["ip"] = ip
response = self.client.post("domains", data)
return response.get("domain", {})
def update_domain(self, domain: str, dns_sec: str) -> None:
"""
Update domain DNSSEC setting
Args:
domain: Domain name
dns_sec: DNSSEC status ("enabled" or "disabled")
"""
self.client.put(f"domains/{domain}", {"dns_sec": dns_sec})
def delete_domain(self, domain: str) -> None:
"""
Delete a DNS domain
Args:
domain: Domain name to delete
"""
self.client.delete(f"domains/{domain}")
def get_soa(self, domain: str) -> Dict:
"""
Get SOA record info
Args:
domain: Domain name
Returns:
SOA record details
"""
return self.client.get(f"domains/{domain}/soa")
def update_soa(
self,
domain: str,
nsprimary: str = None,
email: str = None
) -> None:
"""
Update SOA record
Args:
domain: Domain name
nsprimary: Primary nameserver
email: Admin email
"""
data = {}
if nsprimary:
data["nsprimary"] = nsprimary
if email:
data["email"] = email
self.client.patch(f"domains/{domain}/soa", data)
def get_dnssec(self, domain: str) -> Dict:
"""
Get DNSSEC info
Args:
domain: Domain name
Returns:
DNSSEC details
"""
return self.client.get(f"domains/{domain}/dnssec")
# Record management
def list_records(
self,
domain: str,
per_page: int = 100,
cursor: str = None
) -> Dict:
"""
List DNS records for a domain
Args:
domain: Domain name
per_page: Number of items per page
cursor: Pagination cursor
Returns:
Dict with 'records' list and 'meta' pagination
"""
params = {"per_page": per_page}
if cursor:
params["cursor"] = cursor
return self.client.get(f"domains/{domain}/records", params=params)
def list_all_records(self, domain: str) -> List[Dict]:
"""List all records for domain (auto-paginate)"""
return self.client.paginate(f"domains/{domain}/records", "records")
def get_record(self, domain: str, record_id: str) -> Dict:
"""
Get a specific DNS record
Args:
domain: Domain name
record_id: Record ID
Returns:
Record details
"""
response = self.client.get(f"domains/{domain}/records/{record_id}")
return response.get("record", {})
def create_record(
self,
domain: str,
name: str,
record_type: str,
data: str,
ttl: int = 300,
priority: int = None
) -> Dict:
"""
Create a DNS record
Args:
domain: Domain name
name: Record name (subdomain or "@" for root)
record_type: Record type (A, AAAA, CNAME, MX, TXT, NS, SRV, CAA, SSHFP)
data: Record data/value
ttl: Time to live in seconds (default 300)
priority: Priority (required for MX and SRV)
Returns:
Created record details
Example:
# A record
client.dns.create_record("example.com", "www", "A", "192.168.1.1")
# MX record
client.dns.create_record("example.com", "@", "MX", "mail.example.com", priority=10)
# TXT record (SPF)
client.dns.create_record("example.com", "@", "TXT", "v=spf1 include:_spf.example.com ~all")
"""
record_data = {
"name": name,
"type": record_type,
"data": data,
"ttl": ttl
}
if priority is not None:
record_data["priority"] = priority
response = self.client.post(f"domains/{domain}/records", record_data)
return response.get("record", {})
def update_record(
self,
domain: str,
record_id: str,
name: str = None,
data: str = None,
ttl: int = None,
priority: int = None
) -> None:
"""
Update a DNS record
Args:
domain: Domain name
record_id: Record ID to update
name: New record name
data: New record data
ttl: New TTL
priority: New priority
"""
record_data = {}
if name is not None:
record_data["name"] = name
if data is not None:
record_data["data"] = data
if ttl is not None:
record_data["ttl"] = ttl
if priority is not None:
record_data["priority"] = priority
self.client.patch(f"domains/{domain}/records/{record_id}", record_data)
def delete_record(self, domain: str, record_id: str) -> None:
"""
Delete a DNS record
Args:
domain: Domain name
record_id: Record ID to delete
"""
self.client.delete(f"domains/{domain}/records/{record_id}")
# Convenience methods
def create_a_record(
self,
domain: str,
name: str,
ip: str,
ttl: int = 300
) -> Dict:
"""Create an A record"""
return self.create_record(domain, name, "A", ip, ttl)
def create_aaaa_record(
self,
domain: str,
name: str,
ip: str,
ttl: int = 300
) -> Dict:
"""Create an AAAA record"""
return self.create_record(domain, name, "AAAA", ip, ttl)
def create_cname_record(
self,
domain: str,
name: str,
target: str,
ttl: int = 300
) -> Dict:
"""Create a CNAME record"""
return self.create_record(domain, name, "CNAME", target, ttl)
def create_mx_record(
self,
domain: str,
name: str,
mail_server: str,
priority: int = 10,
ttl: int = 300
) -> Dict:
"""Create an MX record"""
return self.create_record(domain, name, "MX", mail_server, ttl, priority)
def create_txt_record(
self,
domain: str,
name: str,
text: str,
ttl: int = 300
) -> Dict:
"""Create a TXT record"""
return self.create_record(domain, name, "TXT", text, ttl)
def create_ns_record(
self,
domain: str,
name: str,
nameserver: str,
ttl: int = 300
) -> Dict:
"""Create an NS record"""
return self.create_record(domain, name, "NS", nameserver, ttl)