feat: caching

README.md

@@ -24,12 +24,6 @@ There is an [example configuration](cfg/example.json).
 
 ## TODOs
 
-Caching:
-
-- Add parameter `--cache` which saves retrieved DNS records as JSON files.
-- The method `update_record_destination` will then check if the arguments are the same as the saved JSON record.
-- Only if they differ, any requests to the API endpoint will be made.
-
 Alternative external IP detection:
 
 ```python

src/netcup_dns/datetime_util.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+from datetime import datetime, timezone
+
+
+def test():
+    dt = datetime.now()
+
+    print('non UTC:')
+    print(dt)
+
+    print('\nUTC:')
+    print(now())
+    print(to_str(now()))
+    print(now_str())
+    print(from_str(to_str(now())))
+
+    print('\nlocalized:')
+    print(dt.tzinfo)
+    dt = dt.replace(tzinfo=timezone.utc)
+    print(dt)
+
+
+def now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def now_str() -> str:
+    return to_str(now())
+
+
+def to_str(dt: datetime) -> str:
+    return dt.strftime(fmt())
+
+
+def from_str(dt_str: str) -> datetime:
+    dt = datetime.strptime(dt_str, fmt())
+    return dt.replace(tzinfo=timezone.utc)
+
+
+def fmt() -> str:
+    return '%Y%m%dT%H%M%S'
+
+
+def fmt_len() -> int:
+    return 13
+
+
+if __name__ == '__main__':
+    test()

src/netcup_dns/main.py

@@ -11,6 +11,7 @@ import requests
 from nc_dnsapi import Client, DNSRecord
 
 from netcup_dns.exception import UnknownIPException, MultipleRecordsException
+from netcup_dns.record_dst_cache import RecordDestinationCache
 
 
 def main():
@@ -19,6 +20,7 @@ def main():
     """
     args = parse_args()
     cfg_dir: Path = args.cfg_dir
+    cache: RecordDestinationCache | None = args.cache
 
     cfg_files = [file for file in cfg_dir.iterdir() if file.name.endswith('.json') and file.is_file()]
     for cfg_file in cfg_files:
@@ -48,7 +50,7 @@ def main():
                 else:
                     raise ValueError(f'DNS record type {type_} is not supported.')
 
-                if update_record_destination(api, domain, hostname, type_, destination=destination):
+                if update_record_destination(api, domain, hostname, type_, destination, cache):
                     print(f'Set {hostname}.{domain} {type_} record to {destination}')
                 else:
                     print(f'The {hostname}.{domain} {type_} record points already to {destination}')
@@ -62,12 +64,35 @@ def parse_args():
                         dest='cfg_dir',
                         default=Path('/etc/netcup-dns'),
                         type=Path)
+    parser.add_argument('--cache-directory',
+                        help='Path to cache directory. Retrieved/Set DNS records are cached there.',
+                        dest='cache_dir',
+                        default=Path.home().joinpath('.netcup-dns/cache'),
+                        type=Path)
+    parser.add_argument('--cache-validity-seconds',
+                        help='Value in seconds for how long cached DNS records are valid.'
+                             ' Set to `0` to disable caching.',
+                        dest='cache_validity_seconds',
+                        default=7200,
+                        type=int)
     args = parser.parse_args()
 
     if not args.cfg_dir.exists():
         raise Exception(f'The given config directory does not exist: {args.cfg_dir}')
+    if args.cache_validity_seconds < 0:
+        raise Exception(f'A negative cache validity duration is not allowed: {args.cache_validity_seconds}')
 
-def update_record_destination(api: Client, domain: str, hostname: str, type_: str, destination: str) -> bool:
+    if args.cache_validity_seconds == 0:
+        # Disable caching.
+        args.cache = None
+    else:
+        args.cache = RecordDestinationCache(args.cache_dir, args.cache_validity_seconds)
+
+    return args
+
+
+def update_record_destination(api: Client, domain: str, hostname: str, type_: str, destination: str,
+                              cache: RecordDestinationCache = None) -> bool:
     """
     Updates the `destination` of the DNS record identified by `domain`, `hostname` and `type`.
 
@@ -76,8 +101,16 @@ def update_record_destination(api: Client, domain: str, hostname: str, type_: st
     :param hostname:
     :param type_:
     :param destination:
+    :param cache:
     :return: True if `destination` differs from the old destination.
     """
+    # If caching is enabled.
+    if cache is not None:
+        # If a valid cache entry is available, check if the destination is still the same.
+        # If this is the case, we do not need to do anything else.
+        if cache.get(domain, hostname, type_) == destination:
+            return False
+
     try:
         record = get_record(api, domain, hostname, type_)
     except Exception as e:
@@ -86,10 +119,15 @@ def update_record_destination(api: Client, domain: str, hostname: str, type_: st
     if record.destination == destination:
         # The new destination is identical with the current one.
         # Thus, we don't need to call the api.update_dns_record() method.
+        cache.set(domain, hostname, type_, destination)
+        cache.write_to_file()
         return False
     else:
         record.destination = destination
         api.update_dns_record(domain, record)
+        # After a successful API call, update the cached destination.
+        cache.set(domain, hostname, type_, destination)
+        cache.write_to_file()
         return True
 
 

src/netcup_dns/record_dst_cache.py

@@ -0,0 +1,69 @@
+import json
+from datetime import timedelta
+from pathlib import Path
+
+from netcup_dns import datetime_util
+
+
+class RecordDestinationCache:
+    def __init__(self, cache_dir: Path, cache_validity_seconds: int):
+        self.cache_dir = cache_dir
+        self.cache_file = cache_dir.joinpath('record_destinations.json')
+
+        self.cache_validity_seconds = cache_validity_seconds
+
+        self.data: dict[str, tuple[str, str]] | None = None
+
+    def get(self, domain: str, hostname: str, type_: str) -> str | None:
+        data = self.read_from_file()
+        key = self._data_key(domain, hostname, type_)
+        date_str, destination = data.get(key, (None, None))
+
+        if destination is None:
+            return None
+
+        # Check if cached destination is still valid.
+        dt = datetime_util.from_str(date_str)
+        time_difference = datetime_util.now() - dt
+        zero = timedelta()
+        max_difference = timedelta(seconds=self.cache_validity_seconds)
+        #
+        if time_difference <= zero:
+            raise ValueError('Invalid dates')
+        if time_difference >= max_difference:
+            # This cache entry is outdated.
+            return None
+
+        return destination
+
+    def set(self, domain: str, hostname: str, type_: str, destination: str):
+        if self.data is None:
+            raise Exception('Can only modify data after it has been read first.')
+
+        key = self._data_key(domain, hostname, type_)
+        self.data[key] = (datetime_util.now_str(), destination)
+
+    @staticmethod
+    def _data_key(domain, hostname, type_) -> str:
+        return f'{hostname}.{domain}.{type_}'
+
+    def read_from_file(self) -> dict[str, tuple[str, str]]:
+        if self.data is not None:
+            return self.data
+
+        if self.cache_file.exists():
+            data = json.loads(self.cache_file.read_text(encoding='utf-8'))
+            if not isinstance(data, dict):
+                raise ValueError(f'Expected to read a dict from json file, but got {type(data)} instead.')
+            self.data = data
+        else:
+            self.data = {}
+
+        return self.data
+
+    def write_to_file(self):
+        if self.data is None:
+            raise Exception('Can only write data after it has been read first.')
+
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        self.cache_file.write_text(json.dumps(self.data), encoding='utf-8')