Something went wrong. Try again.
This small application can be used to update the value of a DNS record.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889"""Module with the code to update a DNS record."""
import logging
import httpxfrom pydantic import SecretStrfrom pydantic_settings import BaseSettings, SettingsConfigDict
logger = logging.getLogger(__name__)
class Settings(BaseSettings): """Configuration of the updater."""
digital_ocean_token: SecretStr
model_config = SettingsConfigDict(env_prefix="dyns_")
def public_ip() -> str: """Get the public IP of the host.
It uses https://www.ipify.org/. """ response = httpx.get("https://api.ipify.org?format=json") response.raise_for_status() data = response.json() logger.debug(f"Public IP: {str(data)}") return response.json()["ip"]
def digital_ocean_client() -> httpx.Client: """Create a httpx client with parameters for DigitalOcean.""" settings = Settings() token = settings.digital_ocean_token.get_secret_value()
base_url = "https://api.digitalocean.com/v2" headers = {"Authorization": f"Bearer {token}"}
return httpx.Client(base_url=base_url, headers=headers)
def retrieve_record(name: str, domain: str) -> dict[str, str | int | None] | None: """Retrieve the record for the domain and name.""" with digital_ocean_client() as do_client: # looks for the record ID using the list endpoint # ------------------------------------------------------------------------------ response = do_client.get(f"/domains/{domain}/records") response.raise_for_status()
data = response.json() domain_records = data["domain_records"]
# handle pagination next = data["links"]["pages"].get("next") while next: response = do_client.get(next) response.raise_for_status() data = response.json() domain_records += data["domain_records"] next = data["links"]["pages"].get("next")
# filter the results records = [record for record in domain_records if record["name"] == name] if not records: logger.error(f"Record {name} not found in domain {domain}") return None record = records[0]
return record
def updater(name: str, domain: str) -> None: """Update the DNS record using the DigitalOcean API.""" record = retrieve_record(name=name, domain=domain) if not record: msg = f"Record {name}.{domain} not found" logger.error(msg) raise ValueError(msg)
with digital_ocean_client() as do_client: ip = public_ip() payload = {"type": "A", "data": ip} response = do_client.patch( f"/domains/{domain}/records/{record['id']}", json=payload ) response.raise_for_status() logger.info(f"DNS record updated with '{ip}'")