attrs icon indicating copy to clipboard operation
attrs copied to clipboard

Show: logging.config.dictConfig as attrs

Open jochumdev opened this issue 2 months ago • 0 comments

Hey,

Thanks for attrs, I'm using it in one of my upcoming projects!

I did the exercise to generate a logging.config.dictConfig from dataclasses to learn attrs, it is basicaly untested as I'm switching to 3rd party logging library.

  • logging_config.py
from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any, override

import attrs.converters
from attrs import define, field

from . import validators

if TYPE_CHECKING:
    from collections.abc import Iterable


def _convert_log_level(lvl: int | str) -> int:
    """
    Converts a log level from int and str to int.

    Raises:
        KeyError: If the level is not known.

    Notes:
        Minimum Python version: Python 3.11+
    """
    if isinstance(lvl, int):
        return lvl

    return logging.getLevelNamesMapping()[lvl]


@define
class LogLevel:
    level: int = field(
        default=logging.NOTSET,
        converter=_convert_log_level,
        validator=validators.between(logging.NOTSET, logging.CRITICAL),
    )

    def __int__(self) -> int:
        return self.level

    @override
    def __str__(self) -> str:
        return logging.getLevelName(self.level)


@define(kw_only=True)
class LoggingConfig:
    level: LogLevel

    log_dir: Path | None = field(
        default=None,
        converter=attrs.converters.optional(Path),
        validator=validators.path(is_dir=True),
    )

    disable_existing_loggers: bool = field(
        default=False,
    )

    class Formatter:
        def to_dict_config(self) -> dict[str, Any]: ...

    @define(kw_only=True)
    class LoggingFormatter(Formatter):
        cls: str = field(
            default="logging.Formatter",
        )
        format: str = field(
            default="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        )
        datefmt: str = field(
            default="%d %b %y %H:%M:%S",
        )

        @override
        def to_dict_config(self) -> dict[str, Any]:
            return {
                "class": self.cls,
                "format": self.format,
                "datefmt": self.datefmt,
            }

    formatters: dict[str, Formatter] = field(
        default={
            "standard": LoggingFormatter(
                format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            ),
            "detailed": LoggingFormatter(
                format="%(asctime)s | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s",
            ),
            "json": LoggingFormatter(format="%(message)s"),
        },
    )

    class Handler:
        def to_dict_config(self) -> dict[str, Any]: ...

    @define(kw_only=True)
    class StreamHandler(Handler):
        cls: str = field(
            default="logging.StreamHandler",
        )
        level: LogLevel = field(
            default=LogLevel(logging.NOTSET),
        )
        formatter: str = field(
            default="standard",
        )
        stream: str = field(
            default="ext://sys.stdout",
        )

        @override
        def to_dict_config(self) -> dict[str, Any]:
            return {
                "class": self.cls,
                "level": int(self.level),
                "formatter": self.formatter,
                "stream": self.stream,
            }

    @define(kw_only=True)
    class RotatingFileHandler(Handler):
        cls: str = field(
            default="logging.handlers.RotatingFileHandler",
        )
        level: LogLevel = field(
            default=LogLevel(logging.NOTSET),
        )
        formatter: str = field(
            default="standard",
        )
        filename: Path = field(
            default=Path("out.log"),
        )
        max_bytes: int = field(
            default=10 * 1024 * 1024,
        )
        backup_count: int = field(
            default=5,
        )
        encoding: str = field(
            default="utf8",
        )

        @override
        def to_dict_config(self) -> dict[str, Any]:
            return {
                "class": self.cls,
                "level": int(self.level),
                "formatter": self.formatter,
                "filename": str(self.filename),
                "maxBytes": self.max_bytes,
                "backupCount": self.backup_count,
                "encoding": self.encoding,
            }

    handlers: dict[str, Handler] = field(
        default={
            "console": StreamHandler(),
            "file": RotatingFileHandler(),
            "json_file": RotatingFileHandler(
                formatter="json",
                filename=Path("out.json"),
            ),
        }
    )

    @define(kw_only=True)
    class Logger:
        handlers: Iterable[str]
        level: LogLevel = field(
            default=LogLevel(logging.NOTSET),
        )

        def to_dict_config(self) -> dict[str, Any]:
            return {
                "handlers": self.handlers,
                "level": int(self.level),
            }

    root: Logger = field(
        default=Logger(handlers=["console", "file"]),
    )

    loggers: dict[str, Logger] = field(
        default={},
    )

    def to_dict_config(self) -> dict[str, Any]:
        """
        Return a dict in the format accepted by logging.config.dictConfig.
        """
        cfg: dict[str, Any] = {
            "version": 1,
            "disable_existing_loggers": self.disable_existing_loggers,
            "formatters": {k: v.to_dict_config() for (k, v) in self.formatters.items()},
            "handlers": {k: v.to_dict_config() for (k, v) in self.handlers.items()},
            "root": self.root.to_dict_config(),
            "loggers": {k: v.to_dict_config() for (k, v) in self.loggers.items()},
        }
        return cfg

  • validators.py
from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, Any, override

from attrs import define, field
from attrs.validators import and_, ge, le

if TYPE_CHECKING:
    from collections.abc import Callable

    from attrs import Attribute

    Validator = Callable[[Any, Attribute[Any], Any], None]


__all__ = ["between", "path"]


def between(low: int, high: int) -> Validator:
    return and_(ge(low), le(high))


@define(repr=False, slots=False, unsafe_hash=True)
class _PathValidator:
    exists: bool = field()
    is_dir: bool = field()

    def __call__(self, _: Any, attr: Attribute[Any], value: Any) -> None:
        if not isinstance(value, Path):
            msg = f"{attr.name} must be an instance of `pathlib.Path` (got {value!r})"
            raise TypeError(
                msg,
                attr,
                {"exists": self.exists, "is_dir": self.is_dir},
                value,
            )

        if self.exists and not value.exists():
            msg = f"{attr.name} must exist (got none existant path {value!r})"
            raise ValueError(
                msg,
                attr,
                {"exists": self.exists, "is_dir": self.is_dir},
                value,
            )

        if self.is_dir and not value.is_dir():
            msg = f"{attr.name} must be a directory (got none directory {value!r})"
            raise ValueError(
                msg,
                attr,
                {"exists": self.exists, "is_dir": self.is_dir},
                value,
            )

    @override
    def __repr__(self) -> str:
        return (
            f"<validate_path validator with exists={self.exists!r} dir={self.is_dir!r}>"
        )


def path(*, exists: bool = True, is_dir: bool = False) -> Validator:
    """
    A validator that checks if the path exists and optionaly if it is a directory.

    Raises:
        TypeError:
            If the value is not a `pathlib.Path`
        ValueError:
            With a humand readable error message.
    Note:
        Could be optimized by using stat.
    """
    return _PathValidator(exists, is_dir)

Maybe someone has a need for this.

Kind regards, René

jochumdev avatar Dec 02 '25 15:12 jochumdev