muutils.logger.loggingstream

  1from __future__ import annotations
  2
  3import sys
  4import time
  5from dataclasses import dataclass, field
  6from typing import Any, Callable
  7
  8if sys.version_info >= (3, 12):
  9    from typing import override
 10else:
 11    from typing_extensions import override
 12
 13from muutils.logger.simplelogger import AnyIO, NullIO
 14from muutils.misc import sanitize_fname
 15
 16
 17@dataclass
 18class LoggingStream:
 19    """properties of a logging stream
 20
 21    - `name: str` name of the stream
 22    - `aliases: set[str]` aliases for the stream
 23            (calls to these names will be redirected to this stream. duplicate alises will result in errors)
 24            TODO: perhaps duplicate alises should result in duplicate writes?
 25    - `file: str|bool|AnyIO|None` file to write to
 26            - if `None`, will write to standard log
 27            - if `True`, will write to `name + ".log"`
 28            - if `False` will "write" to `NullIO` (throw it away)
 29            - if a string, will write to that file
 30            - if a fileIO type object, will write to that object
 31    - `default_level: int|None` default level for this stream
 32    - `default_contents: dict[str, Callable[[], Any]]` default contents for this stream
 33    - `last_msg: tuple[float, Any]|None` last message written to this stream (timestamp, message)
 34    """
 35
 36    name: str | None
 37    aliases: set[str | None] = field(default_factory=set)
 38    file: str | bool | AnyIO | None = None
 39    default_level: int | None = None
 40    default_contents: dict[str, Callable[[], Any]] = field(default_factory=dict)
 41    handler: AnyIO | None = None
 42
 43    # TODO: implement last-message caching
 44    # last_msg: tuple[float, Any]|None = None
 45
 46    def make_handler(self) -> AnyIO | None:
 47        if self.file is None:
 48            return None
 49        elif isinstance(self.file, str):
 50            # if its a string, open a file
 51            return open(
 52                self.file,
 53                "w",
 54                encoding="utf-8",
 55            )
 56        elif isinstance(self.file, bool):
 57            # if its a bool and true, open a file with the same name as the stream (in the current dir)
 58            # TODO: make this happen in the same dir as the main logfile?
 59            if self.file:
 60                return open(  # type: ignore[return-value]
 61                    f"{sanitize_fname(self.name)}.log.jsonl",
 62                    "w",
 63                    encoding="utf-8",
 64                )
 65            else:
 66                return NullIO()
 67        else:
 68            # if its neither, check it has `.write()` and `.flush()` methods
 69            if (
 70                (
 71                    not hasattr(self.file, "write")
 72                    or (not callable(self.file.write))
 73                    or (not hasattr(self.file, "flush"))
 74                    or (not callable(self.file.flush))
 75                )
 76                or (not hasattr(self.file, "close"))
 77                or (not callable(self.file.close))
 78            ):
 79                raise ValueError(f"stream {self.name} has invalid handler {self.file}")
 80            # ignore type check because we know it has a .write() method,
 81            # assume the user knows what they're doing
 82            return self.file  # type: ignore
 83
 84    def __post_init__(self):
 85        self.aliases = set(self.aliases)
 86        if any(x.startswith("_") for x in self.aliases if x is not None):
 87            raise ValueError(
 88                "stream names or aliases cannot start with an underscore, sorry"
 89            )
 90        self.aliases.add(self.name)
 91        self.default_contents["_timestamp"] = time.time
 92        self.default_contents["_stream"] = lambda: self.name
 93        self.handler = self.make_handler()
 94
 95    def __del__(self):
 96        if self.handler is not None:
 97            self.handler.flush()
 98            self.handler.close()
 99
100    @override
101    def __str__(self):
102        return f"LoggingStream(name={self.name}, aliases={self.aliases}, file={self.file}, default_level={self.default_level}, default_contents={self.default_contents})"