Sfoglia il codice sorgente

refactor: move logging into a custom Formatter

Sam Jaffe 1 mese fa
parent
commit
df4a2bc2c4
2 ha cambiato i file con 65 aggiunte e 0 eliminazioni
  1. 10 0
      src/cipy/__init__.py
  2. 55 0
      src/cipy/_logging.py

+ 10 - 0
src/cipy/__init__.py

@@ -2,16 +2,26 @@
 Entry point for cipy library, re-exporting all of the default items
 Entry point for cipy library, re-exporting all of the default items
 """
 """
 
 
+import logging
 import types
 import types
 import typing
 import typing
 
 
 import pydantic
 import pydantic
 
 
+import cipy._logging
+
 from cipy.action import Call, Composite, NodeScript, Script
 from cipy.action import Call, Composite, NodeScript, Script
 from cipy.common import Context, Factory, Inputs, Outputs, Ref, Status
 from cipy.common import Context, Factory, Inputs, Outputs, Ref, Status
 from cipy.shell import Shell
 from cipy.shell import Shell
 from cipy.workflow import Job, Matrix, MatrixParams, Workflow
 from cipy.workflow import Job, Matrix, MatrixParams, Workflow
 
 
+_handler = logging.StreamHandler()
+_handler.setFormatter(cipy._logging.CIFormatter("%(asctime)s [%(name)s] %(message)s"))
+
+logging.basicConfig(datefmt="%Y-%m-%dT%H:%M:%S.%fZ",
+                    handlers=[ _handler ],
+                    level=logging.INFO)
+
 __all__ = [
 __all__ = [
     "Call",
     "Call",
     "Composite",
     "Composite",

+ 55 - 0
src/cipy/_logging.py

@@ -0,0 +1,55 @@
+"""Logging utilities (internal)"""
+import logging
+import time
+
+from typing import Callable, ClassVar, Final
+
+GREY: Final[str] = "\x1b[38;20m"
+BOLD_PURPLE: Final[str] = "\x1b[35;1m"
+YELLOW: Final[str] = "\x1b[33;20m"
+RED: Final[str] = "\x1b[31;20m"
+BOLD_RED: Final[str] = "\x1b[31;1m"
+RESET: Final[str] = "\x1b[0m"
+
+
+class CIFormatter(logging.Formatter):
+    """A custom formatter for CI messages"""
+    DEFAULT_FORMAT: Final[ClassVar[str]] = (
+        "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
+    )
+
+    DEFAULT_COLOR_CODES: Final[ClassVar[dict[int, str]]] = {
+        logging.DEBUG: BOLD_PURPLE,
+        logging.INFO: GREY,
+        logging.WARNING: YELLOW,
+        logging.ERROR: RED,
+        logging.CRITICAL: BOLD_RED,
+    }
+
+    converter: Callable[[float | None], time.struct_time] = time.gmtime
+    _format: str
+    _color_codes: dict[int, str]
+
+    def __init__(
+        self,
+        fmt: str | None = None,
+        datefmt: str | None = None,
+        *,
+        colors: dict[int, str] | None = None,
+    ) -> None:
+        super().__init__(datefmt=datefmt)
+        if colors is None:
+            colors = {}
+
+        self._format = fmt if fmt else CIFormatter.DEFAULT_FORMAT
+        self._color_codes = CIFormatter.DEFAULT_COLOR_CODES | colors
+
+    def _dispatch(self, level: int) -> logging.Formatter:
+        log_fmt: str | None = self._color_codes.get(level)
+        log_fmt = self._format if log_fmt is None else f"{log_fmt}{self._format}{RESET}"
+        fmt = logging.Formatter(fmt=log_fmt, datefmt=self.datefmt)
+        fmt.converter = self.converter
+        return fmt
+
+    def format(self, record: logging.LogRecord) -> str:
+        return self._dispatch(record.levelno).format(record)