This small application can be used to update the value of a DNS record.
feat: new check function and refactor in api calls
+54
-11
+35
-5
src/dyns/cli.py
+35
-5
src/dyns/cli.py
···
1
1
"""Module with CLI application."""
2
2
3
3
import argparse
4
+
import json
4
5
import logging
5
6
import sys
6
7
7
8
import dyns
8
9
9
-
from .updater import updater
10
+
from .api import public_ip, retrieve_record, updater
10
11
11
12
logger = logging.getLogger(__name__)
12
-
logging.basicConfig(level=logging.INFO)
13
13
14
14
15
15
def update_dns_record(args: argparse.Namespace) -> None:
···
17
17
parts = args.record.split(".")
18
18
domain = ".".join(parts[-2:])
19
19
name = ".".join(parts[:-2])
20
-
logger.debug(f"Updating for name {name} in domain {domain}")
21
20
22
-
updater(name=name, domain=domain)
21
+
if args.check:
22
+
record = retrieve_record(name=name, domain=domain)
23
+
if not record:
24
+
msg = f"Record {name}.{domain} not found"
25
+
logger.error(msg)
26
+
sys.exit(-1)
27
+
28
+
print("Record:\n", json.dumps(record, indent=2))
29
+
ip = public_ip()
30
+
print("Public IP:", ip)
31
+
if ip != record["data"]:
32
+
print("\033[91mThe record has to be updated!\033[0m")
33
+
else:
34
+
print("\033[92mThe record is updated!\033[0m")
35
+
36
+
else:
37
+
logger.debug(f"Updating for name {name} in domain {domain}")
38
+
updater(name=name, domain=domain)
23
39
24
40
25
41
def main():
···
33
49
parser.add_argument(
34
50
"record",
35
51
type=str,
52
+
nargs="?",
36
53
help="domain name (ex. 'home.example.com') to update with your public IP",
37
54
)
38
-
parser.add_argument("-v", "--version", help="show version", action="store_true")
55
+
parser.add_argument("--version", help="show version", action="store_true")
56
+
parser.add_argument(
57
+
"-v", help="increase verbosity of the output", action="store_true"
58
+
)
59
+
parser.add_argument(
60
+
"--check", help="checks the status of the record", action="store_true"
61
+
)
39
62
parser.set_defaults(func=update_dns_record)
40
63
args = parser.parse_args()
41
64
65
+
# check verbosity
66
+
if args.v:
67
+
logging.basicConfig(level=logging.INFO)
68
+
42
69
# just print version
43
70
if args.version:
44
71
print(dyns.__version__)
45
72
sys.exit(0)
73
+
elif not args.record:
74
+
print("You need to indicate a DNS record to update")
75
+
sys.exit(-1)
46
76
47
77
args.func(args)
+19
-6
src/dyns/updater.py
src/dyns/api.py
+19
-6
src/dyns/updater.py
src/dyns/api.py
···
29
29
return response.json()["ip"]
30
30
31
31
32
-
def updater(name: str, domain: str) -> None:
33
-
"""Update the DNS record using the DigitalOcean API."""
32
+
def digital_ocean_client() -> httpx.Client:
33
+
"""Create a httpx client with parameters for DigitalOcean."""
34
34
settings = Settings()
35
35
token = settings.digital_ocean_token.get_secret_value()
36
36
37
37
base_url = "https://api.digitalocean.com/v2"
38
38
headers = {"Authorization": f"Bearer {token}"}
39
39
40
-
# creates a client
41
-
with httpx.Client(base_url=base_url, headers=headers) as do_client:
40
+
return httpx.Client(base_url=base_url, headers=headers)
41
+
42
+
43
+
def 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:
42
46
# looks for the record ID using the list endpoint
43
47
# ------------------------------------------------------------------------------
44
48
response = do_client.get(f"/domains/{domain}/records")
···
63
67
return None
64
68
record = records[0]
65
69
66
-
# updates the record
67
-
# ------------------------------------------------------------------------------
70
+
return record
71
+
68
72
73
+
def 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:
69
82
ip = public_ip()
70
83
payload = {"type": "A", "data": ip}
71
84
response = do_client.patch(