- 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>
349 lines
8.7 KiB
Python
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)
|