Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ requires-python = ">=3.10"
dynamic = ["version"]
dependencies = [
"structlog>=25.2.0",
"typing-extensions; python_version < '3.11'"
"typing-extensions; python_version < '3.11'",
]

[dependency-groups]
Expand All @@ -35,6 +35,7 @@ dev = [
"mkdocstrings[python]>=0.29.0",
"mypy>=1.15.0",
"pre-commit>=4.2.0",
"pydantic>=2",
"pytest>=9",
"pytest-coverage>=0.0",
]
Expand Down
3 changes: 2 additions & 1 deletion src/l2sl/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,15 @@
__version__ = "UNKNOWN"

from ._filter import (
LogLevel,
# LogLevel,
RecordFilter,
SimpleRecordFilter,
StdlibLogLevel,
StdlibLogLevelNumber,
StructlogLogLevel,
)
from ._forward import configure_stdlib_log_forwarding
from ._log_level import LogLevel
from ._parsers import (
Parser,
RegexpEventHandler,
Expand Down
147 changes: 147 additions & 0 deletions src/l2sl/_log_level.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
from __future__ import annotations

__all__ = ["LogLevel", "LogLevelNumber", "StructlogLogLevelName", "StdlibLogLevelName"]

import functools
from typing import Any, ClassVar, Literal, cast

try:
import pydantic
from pydantic_core import core_schema

_PYDANTIC_2_AVAILABLE = tuple(map(int, pydantic.__version__.split(".")[:3])) >= (2,)
except (ImportError, AttributeError):
_PYDANTIC_2_AVAILABLE = False

LogLevelNumber = Literal[0, 10, 20, 30, 40, 50]
StdlibLogLevelName = Literal["NOTSET", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
StructlogLogLevelName = Literal[
"notset", "debug", "info", "warning", "warn", "error", "exception", "critical"
]


@functools.total_ordering
class LogLevel:
_NUMBER_TO_STDLIB_NAME: ClassVar[dict[LogLevelNumber, StdlibLogLevelName]] = {
0: "NOTSET",
10: "DEBUG",
20: "INFO",
30: "WARNING",
40: "ERROR",
50: "CRITICAL",
}
_STRUCTLOG_NAME_TO_STDLIB_NAME: dict[StructlogLogLevelName, StdlibLogLevelName] = {
"notset": "NOTSET",
"debug": "DEBUG",
"info": "INFO",
"warning": "WARNING",
"warn": "WARNING",
"error": "ERROR",
"exception": "ERROR",
"critical": "CRITICAL",
}
_STDLIB_NAME_TO_NUMBER: ClassVar[dict[StdlibLogLevelName, LogLevelNumber]] = {
v: k for k, v in _NUMBER_TO_STDLIB_NAME.items()
}
_STDLIB_NAME_TO_STRUCTLOG_NAME: dict[StdlibLogLevelName, StructlogLogLevelName] = {
"NOTSET": "notset",
"DEBUG": "debug",
"INFO": "info",
"WARNING": "warning",
"ERROR": "error",
"CRITICAL": "critical",
}

def __init__(
self, level: LogLevelNumber | StdlibLogLevelName | StructlogLogLevelName
) -> None:
if (number := cast(LogLevelNumber, level)) in self._NUMBER_TO_STDLIB_NAME:
stdlib_name = self._NUMBER_TO_STDLIB_NAME[number]
structlog_name = self._STDLIB_NAME_TO_STRUCTLOG_NAME[stdlib_name]
elif (
stdlib_name := cast(StdlibLogLevelName, level)
) in self._STDLIB_NAME_TO_NUMBER:
number = self._STDLIB_NAME_TO_NUMBER[stdlib_name]
structlog_name = self._STDLIB_NAME_TO_STRUCTLOG_NAME[stdlib_name]
elif (
structlog_name := cast(StructlogLogLevelName, level)
) in self._STRUCTLOG_NAME_TO_STDLIB_NAME:
stdlib_name = self._STRUCTLOG_NAME_TO_STDLIB_NAME[structlog_name]
number = self._STDLIB_NAME_TO_NUMBER[stdlib_name]
else:
raise ValueError

self._number = number
self._stdlib_name = stdlib_name
self._structlog_name = structlog_name

@property
def number(self) -> LogLevelNumber:
return self._number

@property
def stdlib_name(self) -> StdlibLogLevelName:
return self._stdlib_name

@property
def structlog_name(self) -> StructlogLogLevelName:
return self._structlog_name

def __eq__(self, other: Any) -> bool:
if not isinstance(other, LogLevel):
try:
other = LogLevel(other)
except ValueError:
return NotImplemented
other: LogLevel

return self.number == other.number

def __lt__(self, other: Any) -> bool:
if not isinstance(other, LogLevel):
try:
other = LogLevel(other)
except ValueError:
return NotImplemented
other: LogLevel

return self.number < other.number

def __str__(self) -> str:
return self.stdlib_name

def __repr__(self) -> str:
return f"{type(self)}(number={self.number}, stdlib_name={self.stdlib_name}, structlog_name={self.structlog_name})"

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

if _PYDANTIC_2_AVAILABLE:

@classmethod
def __get_pydantic_core_schema__(
cls, _source_type: Any, _handler: pydantic.GetCoreSchemaHandler
) -> core_schema.CoreSchema:
json_schema = core_schema.chain_schema(
[
core_schema.literal_schema(
[
*cls._NUMBER_TO_STDLIB_NAME,
*cls._STDLIB_NAME_TO_NUMBER,
*cls._STRUCTLOG_NAME_TO_STDLIB_NAME,
]
),
core_schema.no_info_plain_validator_function(cls),
]
)

return core_schema.json_or_python_schema(
json_schema=json_schema,
python_schema=core_schema.union_schema(
[
core_schema.is_instance_schema(cls),
json_schema,
]
),
serialization=core_schema.plain_serializer_function_ser_schema(str),
)
90 changes: 90 additions & 0 deletions tests/test_log_level.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
import json
import operator

import pydantic
import pytest

from l2sl import LogLevel


class TestLogLevel:
@pytest.mark.parametrize(
("type", "value"),
[
*[("number", v) for v in LogLevel._NUMBER_TO_STDLIB_NAME],
*[("stdlib_name", v) for v in LogLevel._STDLIB_NAME_TO_NUMBER],
*[("structlog_name", v) for v in LogLevel._STRUCTLOG_NAME_TO_STDLIB_NAME],
],
)
def test_parsing(self, type, value):
log_level = LogLevel(value)
assert getattr(log_level, type) == value

@pytest.mark.parametrize(
("a", "op", "b"),
[
("info", operator.lt, "warning"),
("info", operator.lt, "WARNING"),
("info", operator.lt, 30),
("info", operator.eq, "info"),
("info", operator.eq, "INFO"),
("info", operator.eq, 20),
("info", operator.gt, "debug"),
("info", operator.gt, "DEBUG"),
("info", operator.gt, 10),
],
)
@pytest.mark.parametrize("as_log_level", ["left", "right", "both"])
def test_comparison(self, a, op, b, as_log_level):
match as_log_level:
case "left":
a = LogLevel(a)
case "right":
b = LogLevel(b)
case "both":
a = LogLevel(a)
b = LogLevel(b)
case _:
raise ValueError(f"{as_log_level=}")

assert op(a, b)

@pytest.mark.parametrize("level", ["info", "INFO", 20])
def test_str(self, level):
log_level = LogLevel(level)
assert str(log_level) == log_level.stdlib_name

@pytest.mark.parametrize("level", ["info", "INFO", 20])
def test_repr_smoke(self, level):
repr(level)

@pytest.mark.parametrize("level", ["info", "INFO", 20, LogLevel("info")])
def test_pydantic_validate_python(self, level):
log_level = level if isinstance(level, LogLevel) else LogLevel(level)

ta = pydantic.TypeAdapter(LogLevel)
assert ta.validate_python(level) == log_level

def test_pydantic_dump_python(self):
log_level = LogLevel("info")

ta = pydantic.TypeAdapter(LogLevel)
assert ta.dump_python(log_level) == str(log_level)

@pytest.mark.parametrize("level", ["info", "INFO", 20])
def test_int(self, level):
log_level = LogLevel(level)
assert int(log_level) == log_level.number

@pytest.mark.parametrize("level", ["info", "INFO", 20])
def test_pydantic_validate_json(self, level):
log_level = LogLevel(level)

ta = pydantic.TypeAdapter(LogLevel)
assert ta.validate_json(json.dumps(level)) == log_level

def test_pydantic_dump_json(self):
log_level = LogLevel("info")

ta = pydantic.TypeAdapter(LogLevel)
assert ta.dump_json(log_level) == json.dumps(str(log_level)).encode()
Loading
Loading