Skip to content

Show: logging.config.dictConfig as attrs #1487

@jochumdev

Description

@jochumdev

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é

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions