mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-03-01 17:00:48 +01:00
feat: add logger module (#37)
This commit is contained in:
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
import typer
|
import typer
|
||||||
|
|
||||||
|
from ..logger import configure_logging
|
||||||
from .config import get_custom_cli
|
from .config import get_custom_cli
|
||||||
from .pyproject import load_pyproject
|
from .pyproject import load_pyproject
|
||||||
|
|
||||||
@@ -27,4 +28,5 @@ if _config.get("fixtures") and _config.get("db_context"):
|
|||||||
@cli.callback()
|
@cli.callback()
|
||||||
def main(ctx: typer.Context) -> None:
|
def main(ctx: typer.Context) -> None:
|
||||||
"""FastAPI utilities CLI."""
|
"""FastAPI utilities CLI."""
|
||||||
|
configure_logging()
|
||||||
ctx.ensure_object(dict)
|
ctx.ensure_object(dict)
|
||||||
|
|||||||
@@ -1,15 +1,15 @@
|
|||||||
"""Fixture system with dependency management and context support."""
|
"""Fixture system with dependency management and context support."""
|
||||||
|
|
||||||
import logging
|
|
||||||
from collections.abc import Callable, Sequence
|
from collections.abc import Callable, Sequence
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import Any, cast
|
from typing import Any, cast
|
||||||
|
|
||||||
from sqlalchemy.orm import DeclarativeBase
|
from sqlalchemy.orm import DeclarativeBase
|
||||||
|
|
||||||
|
from ..logger import get_logger
|
||||||
from .enum import Context
|
from .enum import Context
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = get_logger()
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|||||||
@@ -1,4 +1,3 @@
|
|||||||
import logging
|
|
||||||
from collections.abc import Callable, Sequence
|
from collections.abc import Callable, Sequence
|
||||||
from typing import Any, TypeVar
|
from typing import Any, TypeVar
|
||||||
|
|
||||||
@@ -6,10 +5,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||||||
from sqlalchemy.orm import DeclarativeBase
|
from sqlalchemy.orm import DeclarativeBase
|
||||||
|
|
||||||
from ..db import get_transaction
|
from ..db import get_transaction
|
||||||
|
from ..logger import get_logger
|
||||||
from .enum import LoadStrategy
|
from .enum import LoadStrategy
|
||||||
from .registry import Context, FixtureRegistry
|
from .registry import Context, FixtureRegistry
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = get_logger()
|
||||||
|
|
||||||
T = TypeVar("T", bound=DeclarativeBase)
|
T = TypeVar("T", bound=DeclarativeBase)
|
||||||
|
|
||||||
|
|||||||
81
src/fastapi_toolsets/logger.py
Normal file
81
src/fastapi_toolsets/logger.py
Normal file
@@ -0,0 +1,81 @@
|
|||||||
|
"""Logging configuration for FastAPI applications and CLI tools."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import sys
|
||||||
|
from typing import Literal
|
||||||
|
|
||||||
|
__all__ = ["LogLevel", "configure_logging", "get_logger"]
|
||||||
|
|
||||||
|
DEFAULT_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||||
|
UVICORN_LOGGERS = ("uvicorn", "uvicorn.access", "uvicorn.error")
|
||||||
|
|
||||||
|
LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
||||||
|
|
||||||
|
|
||||||
|
def configure_logging(
|
||||||
|
level: LogLevel | int = "INFO",
|
||||||
|
fmt: str = DEFAULT_FORMAT,
|
||||||
|
logger_name: str | None = None,
|
||||||
|
) -> logging.Logger:
|
||||||
|
"""Configure logging with a stdout handler and consistent format.
|
||||||
|
|
||||||
|
Sets up a :class:`~logging.StreamHandler` writing to stdout with the
|
||||||
|
given format and level. Also configures the uvicorn loggers so that
|
||||||
|
FastAPI access logs use the same format.
|
||||||
|
|
||||||
|
Calling this function multiple times is safe -- existing handlers are
|
||||||
|
replaced rather than duplicated.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
level: Log level (e.g. ``"DEBUG"``, ``"INFO"``, or ``logging.DEBUG``).
|
||||||
|
fmt: Log format string. Defaults to
|
||||||
|
``"%(asctime)s - %(name)s - %(levelname)s - %(message)s"``.
|
||||||
|
logger_name: Logger name to configure. ``None`` (the default)
|
||||||
|
configures the root logger so all loggers inherit the settings.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
The configured Logger instance.
|
||||||
|
"""
|
||||||
|
formatter = logging.Formatter(fmt)
|
||||||
|
|
||||||
|
handler = logging.StreamHandler(sys.stdout)
|
||||||
|
handler.setFormatter(formatter)
|
||||||
|
|
||||||
|
logger = logging.getLogger(logger_name)
|
||||||
|
logger.handlers.clear()
|
||||||
|
logger.addHandler(handler)
|
||||||
|
logger.setLevel(level)
|
||||||
|
|
||||||
|
for name in UVICORN_LOGGERS:
|
||||||
|
uv_logger = logging.getLogger(name)
|
||||||
|
uv_logger.handlers.clear()
|
||||||
|
uv_logger.addHandler(handler)
|
||||||
|
uv_logger.setLevel(level)
|
||||||
|
|
||||||
|
return logger
|
||||||
|
|
||||||
|
|
||||||
|
_SENTINEL = object()
|
||||||
|
|
||||||
|
|
||||||
|
def get_logger(name: str | None = _SENTINEL) -> logging.Logger: # type: ignore[assignment]
|
||||||
|
"""Return a logger with the given *name*.
|
||||||
|
|
||||||
|
A thin convenience wrapper around :func:`logging.getLogger` that keeps
|
||||||
|
logging imports consistent across the codebase.
|
||||||
|
|
||||||
|
When called without arguments, the caller's ``__name__`` is used
|
||||||
|
automatically, so ``get_logger()`` in a module is equivalent to
|
||||||
|
``logging.getLogger(__name__)``. Pass ``None`` explicitly to get the
|
||||||
|
root logger.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
name: Logger name. Defaults to the caller's ``__name__``.
|
||||||
|
Pass ``None`` to get the root logger.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
A Logger instance.
|
||||||
|
"""
|
||||||
|
if name is _SENTINEL:
|
||||||
|
name = sys._getframe(1).f_globals.get("__name__")
|
||||||
|
return logging.getLogger(name)
|
||||||
118
tests/test_logger.py
Normal file
118
tests/test_logger.py
Normal file
@@ -0,0 +1,118 @@
|
|||||||
|
import logging
|
||||||
|
import sys
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from fastapi_toolsets.logger import (
|
||||||
|
DEFAULT_FORMAT,
|
||||||
|
UVICORN_LOGGERS,
|
||||||
|
configure_logging,
|
||||||
|
get_logger,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _reset_loggers():
|
||||||
|
"""Reset the root and uvicorn loggers after each test."""
|
||||||
|
yield
|
||||||
|
root = logging.getLogger()
|
||||||
|
root.handlers.clear()
|
||||||
|
root.setLevel(logging.WARNING)
|
||||||
|
for name in UVICORN_LOGGERS:
|
||||||
|
uv = logging.getLogger(name)
|
||||||
|
uv.handlers.clear()
|
||||||
|
uv.setLevel(logging.NOTSET)
|
||||||
|
|
||||||
|
|
||||||
|
class TestConfigureLogging:
|
||||||
|
def test_sets_up_handler_and_format(self):
|
||||||
|
logger = configure_logging()
|
||||||
|
|
||||||
|
assert len(logger.handlers) == 1
|
||||||
|
handler = logger.handlers[0]
|
||||||
|
assert isinstance(handler, logging.StreamHandler)
|
||||||
|
assert handler.stream is sys.stdout
|
||||||
|
assert handler.formatter is not None
|
||||||
|
assert handler.formatter._fmt == DEFAULT_FORMAT
|
||||||
|
|
||||||
|
def test_default_level_is_info(self):
|
||||||
|
logger = configure_logging()
|
||||||
|
|
||||||
|
assert logger.level == logging.INFO
|
||||||
|
|
||||||
|
def test_custom_level_string(self):
|
||||||
|
logger = configure_logging(level="DEBUG")
|
||||||
|
|
||||||
|
assert logger.level == logging.DEBUG
|
||||||
|
|
||||||
|
def test_custom_level_int(self):
|
||||||
|
logger = configure_logging(level=logging.WARNING)
|
||||||
|
|
||||||
|
assert logger.level == logging.WARNING
|
||||||
|
|
||||||
|
def test_custom_format(self):
|
||||||
|
custom_fmt = "%(levelname)s: %(message)s"
|
||||||
|
logger = configure_logging(fmt=custom_fmt)
|
||||||
|
|
||||||
|
handler = logger.handlers[0]
|
||||||
|
assert handler.formatter is not None
|
||||||
|
assert handler.formatter._fmt == custom_fmt
|
||||||
|
|
||||||
|
def test_named_logger(self):
|
||||||
|
logger = configure_logging(logger_name="myapp")
|
||||||
|
|
||||||
|
assert logger.name == "myapp"
|
||||||
|
assert len(logger.handlers) == 1
|
||||||
|
|
||||||
|
def test_default_configures_root_logger(self):
|
||||||
|
logger = configure_logging()
|
||||||
|
|
||||||
|
assert logger is logging.getLogger()
|
||||||
|
|
||||||
|
def test_idempotent_no_duplicate_handlers(self):
|
||||||
|
configure_logging()
|
||||||
|
configure_logging()
|
||||||
|
logger = configure_logging()
|
||||||
|
|
||||||
|
assert len(logger.handlers) == 1
|
||||||
|
|
||||||
|
def test_configures_uvicorn_loggers(self):
|
||||||
|
configure_logging(level="DEBUG")
|
||||||
|
|
||||||
|
for name in UVICORN_LOGGERS:
|
||||||
|
uv_logger = logging.getLogger(name)
|
||||||
|
assert len(uv_logger.handlers) == 1
|
||||||
|
assert uv_logger.level == logging.DEBUG
|
||||||
|
handler = uv_logger.handlers[0]
|
||||||
|
assert handler.formatter is not None
|
||||||
|
assert handler.formatter._fmt == DEFAULT_FORMAT
|
||||||
|
|
||||||
|
def test_returns_configured_logger(self):
|
||||||
|
logger = configure_logging(logger_name="test.return")
|
||||||
|
|
||||||
|
assert isinstance(logger, logging.Logger)
|
||||||
|
assert logger.name == "test.return"
|
||||||
|
|
||||||
|
|
||||||
|
class TestGetLogger:
|
||||||
|
def test_returns_named_logger(self):
|
||||||
|
logger = get_logger("myapp.services")
|
||||||
|
|
||||||
|
assert isinstance(logger, logging.Logger)
|
||||||
|
assert logger.name == "myapp.services"
|
||||||
|
|
||||||
|
def test_returns_root_logger_when_none(self):
|
||||||
|
logger = get_logger(None)
|
||||||
|
|
||||||
|
assert logger is logging.getLogger()
|
||||||
|
|
||||||
|
def test_defaults_to_caller_module_name(self):
|
||||||
|
logger = get_logger()
|
||||||
|
|
||||||
|
assert logger.name == __name__
|
||||||
|
|
||||||
|
def test_same_name_returns_same_logger(self):
|
||||||
|
a = get_logger("myapp")
|
||||||
|
b = get_logger("myapp")
|
||||||
|
|
||||||
|
assert a is b
|
||||||
Reference in New Issue
Block a user