mgabarda.com
This small application can be used to update the value of a DNS record.
1"""Module with the code to update a DNS record."""
2
3import logging
4
5import httpx
6from pydantic import SecretStr
7from pydantic_settings import BaseSettings, SettingsConfigDict
8
9logger = logging.getLogger(__name__)
10
11
12class Settings(BaseSettings):
13 """Configuration of the updater."""
14
15 digital_ocean_token: SecretStr
16
17 model_config = SettingsConfigDict(env_prefix="dyns_")
18
19
20def public_ip() -> str:
21 """Get the public IP of the host.
22
23 It uses https://www.ipify.org/.
24 """
25 response = httpx.get("https://api.ipify.org?format=json")
26 response.raise_for_status()
27 data = response.json()
28 logger.debug(f"Public IP: {str(data)}")
29 return response.json()["ip"]
30
31
32def digital_ocean_client() -> httpx.Client:
33 """Create a httpx client with parameters for DigitalOcean."""
34 settings = Settings()
35 token = settings.digital_ocean_token.get_secret_value()
36
37 base_url = "https://api.digitalocean.com/v2"
38 headers = {"Authorization": f"Bearer {token}"}
39
40 return httpx.Client(base_url=base_url, headers=headers)
41
42
43def retrieve_record(name: str, domain: str) -> dict[str, str | int | None] | None:
44 """Retrieve the record for the domain and name."""
45 with digital_ocean_client() as do_client:
46 # looks for the record ID using the list endpoint
47 # ------------------------------------------------------------------------------
48 response = do_client.get(f"/domains/{domain}/records")
49 response.raise_for_status()
50
51 data = response.json()
52 domain_records = data["domain_records"]
53
54 # handle pagination
55 next = data["links"]["pages"].get("next")
56 while next:
57 response = do_client.get(next)
58 response.raise_for_status()
59 data = response.json()
60 domain_records += data["domain_records"]
61 next = data["links"]["pages"].get("next")
62
63 # filter the results
64 records = [record for record in domain_records if record["name"] == name]
65 if not records:
66 logger.error(f"Record {name} not found in domain {domain}")
67 return None
68 record = records[0]
69
70 return record
71
72
73def updater(name: str, domain: str) -> None:
74 """Update the DNS record using the DigitalOcean API."""
75 record = retrieve_record(name=name, domain=domain)
76 if not record:
77 msg = f"Record {name}.{domain} not found"
78 logger.error(msg)
79 raise ValueError(msg)
80
81 with digital_ocean_client() as do_client:
82 ip = public_ip()
83 payload = {"type": "A", "data": ip}
84 response = do_client.patch(
85 f"/domains/{domain}/records/{record['id']}", json=payload
86 )
87 response.raise_for_status()
88 logger.info(f"DNS record updated with '{ip}'")