New logging metaclass

Author: codeSamuraii

lib/logging.py

@@ -1,6 +1,8 @@
 import logging
 
 
+# ---- FORMATTING ----
+
 class ColoredFormatter(logging.Formatter):
     """Custom formatter with colored output and specific formatting for transfers."""
 
@@ -59,12 +61,13 @@ def format(self, record: logging.LogRecord) -> str:
 
         return result + exc_text
 
-
 class HealthCheckFilter(logging.Filter):
     def filter(self, record):
         return '"GET /health HTTP/1.1" 200' not in record.getMessage()
 
 
+# ---- LOGGING SETUP ----
+
 def setup_logging() -> logging.Logger:
     """Configure all loggers to use our custom ColoredFormatter."""
     formatter = ColoredFormatter(
@@ -96,7 +99,6 @@ def setup_logging() -> logging.Logger:
 
     return root_logger
 
-
 def get_logger(logger_name: str) -> logging.Logger:
     console_handler = logging.StreamHandler()
     console_handler.setLevel(logging.DEBUG)
@@ -114,3 +116,38 @@ def get_logger(logger_name: str) -> logging.Logger:
     logger.propagate = False
 
     return logger
+
+
+# ---- PATCHING ----
+
+class HasLogging(type):
+    """Metaclass that automatically adds logging methods and a logger property."""
+
+    def __new__(mcs, name, bases, namespace, **kwargs):
+        name_from = kwargs.get('name_from', 'name')
+
+        @property
+        def logger(self):
+            if not hasattr(self, '_logger'):
+                class_name = self.__class__.__name__
+                fallback_name = class_name + str(id(self))[-4:]
+                logger_name = getattr(self, name_from, fallback_name)
+                self._logger = get_logger(logger_name)
+                if not hasattr(self, name_from):
+                    self._logger.warning(
+                        f"Object {class_name} does not have attribute '{name_from}', "
+                        f"using default name: {logger_name}"
+                    )
+            return self._logger
+
+        namespace['logger'] = logger
+
+        def make_log_method(level):
+            def log_method(self, msg, *args, **kwargs):
+                getattr(self.logger, level)(msg, *args, **kwargs)
+            return log_method
+
+        for level in {'debug', 'info', 'warning', 'error', 'exception', 'critical'}:
+            namespace[level] = make_log_method(level)
+
+        return super().__new__(mcs, name, bases, namespace)

lib/store.py

@@ -3,10 +3,10 @@
 import redis.asyncio as redis
 from typing import Optional, Annotated
 
-from lib.logging import get_logger
+from lib.logging import HasLogging, get_logger
 
 
-class Store:
+class Store(metaclass=HasLogging, name_from='transfer_id'):
     """
     Redis-based store for file transfer queues and events.
     Handles data queuing and event signaling for transfer coordination.
@@ -17,7 +17,6 @@ class Store:
     def __init__(self, transfer_id: str):
         self.transfer_id = transfer_id
         self.redis = self.get_redis()
-        self.log = get_logger(transfer_id)
 
         self._k_queue = self.key('queue')
         self._k_meta = self.key('metadata')
@@ -77,12 +76,12 @@ async def wait_for_event(self, event_name: str, timeout: float = 300.0) -> None:
         async def _poll_marker():
             while not await self.redis.exists(event_marker_key):
                 await asyncio.sleep(1)
-            self.log.debug(f">> POLL: Event '{event_name}' fired.")
+            self.debug(f">> POLL: Event '{event_name}' fired.")
 
         async def _listen_for_message():
             async for message in pubsub.listen():
                 if message and message['type'] == 'message':
-                    self.log.debug(f">> SUB : Received message for event '{event_name}'.")
+                    self.debug(f">> SUB : Received message for event '{event_name}'.")
                     return
 
         poll_marker = asyncio.wait_for(_poll_marker(), timeout=timeout)
@@ -98,7 +97,7 @@ async def _listen_for_message():
                 task.cancel()
 
         except asyncio.TimeoutError:
-            self.log.error(f"Timeout waiting for event '{event_name}' after {timeout} seconds.")
+            self.error(f"Timeout waiting for event '{event_name}' after {timeout} seconds.")
             for task in tasks:
                 task.cancel()
             raise
@@ -182,6 +181,6 @@ async def cleanup(self) -> int:
                 break
 
         if keys_to_delete:
-            self.log.debug(f"- Cleaning up {len(keys_to_delete)} keys")
+            self.debug(f"- Cleaning up {len(keys_to_delete)} keys")
             return await self.redis.delete(*keys_to_delete)
         return 0

lib/transfer.py

@@ -4,8 +4,10 @@
 from typing import AsyncIterator, Callable, Awaitable, Optional
 
 from lib.store import Store
-from lib.logging import get_logger
 from lib.metadata import FileMetadata
+from lib.logging import HasLogging, get_logger
+
+logger = get_logger('transfer')
 
 
 class FileTransferError(Exception):
@@ -26,7 +28,8 @@ def __repr__(self):
         return self.__class__.__name__ + f"({', '.join(map(repr, self.args))}, {self.kwargs})"
 
 
-class FileTransfer:
+class FileTransfer(metaclass=HasLogging, name_from='uid'):
+    """Handles file transfers, including metadata queries and data streaming."""
 
     DONE_FLAG = b'\x00\xFF'
     DEAD_FLAG = b'\xDE\xAD'
@@ -38,9 +41,6 @@ def __init__(self, uid: str, file: FileMetadata):
         self.bytes_uploaded = 0
         self.bytes_downloaded = 0
 
-        log = get_logger(self.uid)
-        self.debug, self.info, self.warning, self.error = log.debug, log.info, log.warning, log.error
-
     @classmethod
     async def create(cls, uid: str, file: FileMetadata):
         transfer = cls(uid, file)