Minor cleanup post opencode

ipam_utils/utils.py

@@ -50,11 +50,7 @@ class IPAMUtils:
             data = response.json()
             return data.get("data")[0] if data.get("data") else None
         except Exception as e:
-            # Log exception here in a real scenario
-            print(
-                f"Warning: Could not retrieve IP info for {ip_address}: {e}",
-                file=sys.stderr,
-            )
+            logging.warning(f"Could not retrieve IP info for {ip_address}: {e}")
             return None
 
     def get_ip_id(self, ip_address: str) -> Optional[int]:
@@ -80,10 +76,7 @@ class IPAMUtils:
             data = response.json()
             return data.get("data")[0].get("id") if data.get("data") else None
         except Exception as e:
-            print(
-                f"Warning: Could not find subnet ID for {ip_address}: {e}",
-                file=sys.stderr,
-            )
+            logging.warning(f"Could not find subnet ID for {ip_address}: {e}")
             return None
 
     def update_ip_information(
@@ -132,15 +125,12 @@ class IPAMUtils:
                     # On successful creation, fetch the new record details
                     return self.get_ip_info(ip_address)
                 except Exception as e:
-                    print(
-                        f"Error creating new IP record {ip_address}: {e}",
-                        file=sys.stderr,
-                    )
+                    logging.info(f"Error creating new IP record {ip_address}: {e}")
+
                     return None
             else:
-                print(
-                    f"Could not determine subnet ID for {ip_address}. Skipping update.",
-                    file=sys.stderr,
+                logging.warning(
+                    f"Could not determine subnet ID for {ip_address}. Skipping update."
                 )
                 return None
 

kea_listener.py

@@ -5,6 +5,7 @@ import logging
 import random
 import socket
 import time
+from typing import TypedDict
 import uuid
 import urllib3
 import threading
@@ -14,6 +15,7 @@ import paho.mqtt.client as mqtt_client
 import paho.mqtt.publish as publish
 from ipam_utils import IPAMUtils
 from logging_setup import setup_split_logging
+from collections.abc import Callable
 
 
 @dataclass
@@ -26,13 +28,25 @@ class KeaListenerConfig:
         self.count = 0
         self.last_update_time = time.asctime()
 
+    def increment_count(self) -> None:
+        with self.lock:
+            self.count += 1
+            self.last_update_time = time.asctime()
+
 
 def on_connect(client, userdata, flags, rc, properties=None):
     logging.info("Connected with result code " + str(rc))
     client.subscribe(userdata.mqtt_topic)
 
 
-def extract_lease_ip_info(msg_dict) -> dict[str, str | None]:
+class LeaseInfo(TypedDict):
+    HOOK_TYPE: str
+    ip_address: str | None
+    hostname: str | None
+    mac_address: str | None
+
+
+def extract_lease_ip_info(msg_dict) -> LeaseInfo | None:
     ip_address = None
     hostname = None
     mac_address = None
@@ -57,7 +71,7 @@ def extract_lease_ip_info(msg_dict) -> dict[str, str | None]:
     }
 
 
-def _process_commit(userdata: KeaListenerConfig, lease_ip_info: dict):
+def _process_commit(userdata: KeaListenerConfig, lease_ip_info: LeaseInfo) -> None:
     """Handles IP lease renewal events (COMMIT)."""
     if logging.getLogger().isEnabledFor(logging.DEBUG):
         logging.debug(
@@ -75,7 +89,9 @@ def _process_commit(userdata: KeaListenerConfig, lease_ip_info: dict):
         logging.error("Error updating IP information: " + str(e))
 
 
-def _process_lease_change(userdata: KeaListenerConfig, lease_ip_info: dict):
+def _process_lease_change(
+    userdata: KeaListenerConfig, lease_ip_info: LeaseInfo
+) -> None:
     """Handles lease expiry/release/recovery events (CLEANUP)."""
     try:
         userdata.ipam_util.clear_ip_information(ip_address=lease_ip_info["ip_address"])
@@ -196,14 +212,6 @@ def main():
             auth=mqtt_auth,
             payload=mqtt_payload,
         )
-        # Commented out debugging section remains for context
-        # if logging.getLogger().isEnabledFor(logging.DEBUG):
-        #     logging.debug(
-        #         "MARK: last update: "
-        #         + userdata.last_update_time
-        #         + " updates processed: "
-        #         + str(userdata.count)
-        #     )
         time.sleep(60)
     client.loop_stop(force=False)