Logging¶

class Loggable:
    """Mixin giving instances a logger that names them in each record."""

    @cached_property
    def logger(self) -> logging.LoggerAdapter[logging.Logger]:
        """Logger naming this instance in each record."""
        return ContextualAdapter(logging.getLogger("redsun"), self)

class BufferHandler(logging.Handler):
    """Retain the most recent log records, and announce each one as it arrives.

    A consumer built later in the session can still show earlier records.
    Application records and each service's records are kept apart, each
    dropping its oldest when full, so a noisy service drops only its own.

    Parameters
    ----------
    capacity : int
        How many application records to retain.
    service_capacity : int
        How many records of each service to retain.
    """

    sig_record = Signal(logging.LogRecord)

    def __init__(
        self,
        capacity: int = APPLICATION_CAPACITY,
        service_capacity: int = SERVICE_CAPACITY,
    ) -> None:
        super().__init__()
        self._capacity = capacity
        self._service_capacity = service_capacity
        self._records: deque[logging.LogRecord] = deque(maxlen=capacity)
        self._service_records: dict[str, deque[logging.LogRecord]] = {}

    @property
    def capacity(self) -> int:
        """How many application records the buffer retains."""
        return self._capacity

    @property
    def service_capacity(self) -> int:
        """How many records of each service the buffer retains."""
        return self._service_capacity

    @property
    def records(self) -> tuple[logging.LogRecord, ...]:
        """The retained application records, oldest first."""
        return tuple(self._records)

    @property
    def services(self) -> tuple[str, ...]:
        """The services a record has come from, in the order they first did."""
        return tuple(self._service_records)

    def service_records(
        self, service: str | None = None
    ) -> tuple[logging.LogRecord, ...]:
        """Return one service's retained records, or every service's, oldest first."""
        if service is not None:
            return tuple(self._service_records.get(service, ()))
        return tuple(
            heapq.merge(*self._service_records.values(), key=lambda r: r.created)
        )

    def emit(self, record: logging.LogRecord) -> None:
        """Retain *record* with the records of its source, and announce it."""
        service = service_of(record)
        if service is None:
            self._records.append(record)
        else:
            self._service_records.setdefault(
                service, deque(maxlen=self._service_capacity)
            ).append(record)
        self.sig_record.emit(record)

    def clear(self) -> None:
        """Drop every retained record."""
        self._records.clear()
        self._service_records.clear()

class SessionFileHandler(RotatingFileHandler):
    """Write the records of one run of a session to a file of its own.

    The file is in the user's log directory, in a folder named after the
    session, and named after the run: its start time and process. It rotates at
    `LOG_MAX_BYTES`, keeping `LOG_BACKUPS` older files.

    The application's file, ``<run>.log``, takes no service records, and
    opening it deletes the files of all but the session's `LOG_RUNS_KEPT` most
    recent runs. A service's file, ``<run>.<service>.log``, belongs to *run*,
    is installed with ``add_handler(handler, service)``, and is created when
    the service first logs.

    Parameters
    ----------
    session : str
        Name of the session.
    service : str | None
        The service whose records the file holds, ``None`` for the application.
    run : str | None
        The run, as the application handler's `run`. ``None`` starts a new
        run.
    """

    def __init__(
        self, session: str, service: str | None = None, run: str | None = None
    ) -> None:
        folder = Path(user_log_dir("redsun", appauthor=False)) / session_folder(session)
        folder.mkdir(parents=True, exist_ok=True)
        if run is None:
            started = datetime.now().astimezone().strftime("%Y-%m-%dT%H-%M-%S")
            run = f"{started}_{os.getpid()}"
        if service is None:
            _delete_old_runs(folder, keep=LOG_RUNS_KEPT - 1)
        self.run = run
        super().__init__(
            folder / (f"{run}.log" if service is None else f"{run}.{service}.log"),
            maxBytes=LOG_MAX_BYTES,
            backupCount=LOG_BACKUPS,
            encoding="utf-8",
            delay=service is not None,
        )
        if service is None:
            self.addFilter(lambda record: service_of(record) is None)

    @property
    def files(self) -> list[Path]:
        """The files of this run, oldest records first."""
        current = Path(self.baseFilename)
        rotated = [
            current.with_name(f"{current.name}.{index}")
            for index in range(self.backupCount, 0, -1)
        ]
        return [path for path in (*rotated, current) if path.exists()]