update podman ddns to be more efficient

active/podman_ddns/update.py

@@ -6,10 +6,17 @@ export ROUTE53_RECORD=something.mydomain.com
 import logging
 import os
 import subprocess
-from typing import TYPE_CHECKING
+import yaml
+import sys
+from typing import TYPE_CHECKING, TypedDict
 
 import boto3
 
+try:
+    from yaml import CLoader as Loader
+except ImportError:
+    from yaml import Loader  # type: ignore
+
 if TYPE_CHECKING:
     from mypy_boto3_route53 import Route53Client
 
@@ -21,10 +28,20 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
-HOSTED_ZONE_ID = os.getenv("HOSTED_ZONE_ID")
-ROUTE53_RECORD = os.getenv("ROUTE53_RECORD")
-SKIP_IPV4 = os.getenv("SKIP_IPV4", "false").lower() == "true"
-SKIP_IPV6 = os.getenv("SKIP_IPV6", "false").lower() == "true"
+ROUTE53_RECORDS_FILE = os.getenv("ROUTE53_RECORDS_FILE")
+GLOBAL_SKIP_IPV4 = os.getenv("GLOBAL_SKIP_IPV4", "false").lower() == "true"
+GLOBAL_SKIP_IPV6 = os.getenv("GLOBAL_SKIP_IPV6", "false").lower() == "true"
+
+
+class RecordType(TypedDict):
+    record: str
+    hosted_zone_id: str
+    skip_ipv4: bool | None
+    skip_ipv6: bool | None
+
+
+class RecordYamlStruct(TypedDict):
+    records: list[RecordType]
 
 
 def get_ipv4() -> str:
@@ -92,18 +109,20 @@ def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str):
 
 
 def main():
-    if not HOSTED_ZONE_ID:
-        logger.error("HOSTED_ZONE_ID env var not found!")
+    if not ROUTE53_RECORDS_FILE:
+        logger.error("ROUTE53_RECORDS_FILE env var not found!")
         exit(1)
 
-    if not ROUTE53_RECORD:
-        logger.error("ROUTE53_RECORD env var not found!")
-        exit(1)
+    try:
+        with open(ROUTE53_RECORDS_FILE) as f:
+            records_file_contents: RecordYamlStruct = yaml.load(f, Loader)
+    except FileNotFoundError as e:
+        logger.error(e)
+        sys.exit(1)
 
-    logger.info(f"Attempting to update {ROUTE53_RECORD} from {HOSTED_ZONE_ID}.")
-
-    if SKIP_IPV4:
-        logger.warning("Skipping IPv4.")
+    if GLOBAL_SKIP_IPV4:
+        public_ipv4 = None
+        logger.warning("Globally skipping IPv4.")
     else:
         logger.info("Getting IPv4 address from ifconfig.me")
         public_ipv4 = get_ipv4()
@@ -111,14 +130,10 @@ def main():
             logger.error("Public IPv4 not found.")
             exit(1)
         logger.info(f"Public IPv4 is {public_ipv4}")
-        update_ipv4(
-            hosted_zone_id=HOSTED_ZONE_ID,
-            record=ROUTE53_RECORD,
-            public_ipv4=public_ipv4,
-        )
 
-    if SKIP_IPV6:
-        logger.warning("Skipping IPv6")
+    if GLOBAL_SKIP_IPV6:
+        public_ipv6 = None
+        logger.warning("Globally Skipping IPv6")
     else:
         logger.info("Getting IPv6 address from ifconfig.me")
         public_ipv6 = get_ipv6()
@@ -126,11 +141,32 @@ def main():
             logger.error("Public IPv6 not found.")
             exit(1)
         logger.info(f"Public IPv6 is {public_ipv6}")
-        update_ipv6(
-            hosted_zone_id=HOSTED_ZONE_ID,
-            record=ROUTE53_RECORD,
-            public_ipv6=public_ipv6,
-        )
+
+    for record in records_file_contents["records"]:
+
+        logger.info(f"Attempting to update {record['record']} from {record['hosted_zone_id']}.")
+
+        if record.get("skip_ipv4"):
+            logger.info(f"{record['record']} requested to skip IPv4")
+        elif GLOBAL_SKIP_IPV4 or not public_ipv4:
+            logger.info("Globally skipping IPv4")
+        else:
+            update_ipv4(
+                hosted_zone_id=record["hosted_zone_id"],
+                record=record["record"],
+                public_ipv4=public_ipv4,
+            )
+
+        if record.get("skip_ipv6"):
+            logger.info(f"{record['record']} requested to skip IPv6")
+        elif GLOBAL_SKIP_IPV6 or not public_ipv6:
+            logger.info("Globally skipping IPv6")
+        else:
+            update_ipv6(
+                hosted_zone_id=record["hosted_zone_id"],
+                record=record["record"],
+                public_ipv6=public_ipv6,
+            )
 
 
 if __name__ == "__main__":