API Documentation

add_logger

Add and initialize logger middlewares and handlers to FastAPI application.

Parameters:

Name	Type	Description	Default
app	(FastAPI, required)	FastAPI application instance.	required
config	(LoggerConfigPM, required)	Logger configuration model.	required
has_proxy_headers	bool | None	Whether to use proxy headers. Defaults to None.	None
has_cf_headers	bool | None	Whether to use Cloudflare headers. Defaults to None.	None

Returns:

Name	Type	Description
Logger	Logger	Initialized Logger instance.

Source code in src/beans_logging_fastapi/_core.py

@validate_call(config={"arbitrary_types_allowed": True})
def add_logger(
    app: FastAPI,
    config: LoggerConfigPM,
    has_proxy_headers: bool | None = None,
    has_cf_headers: bool | None = None,
) -> "Logger":
    """Add and initialize logger middlewares and handlers to FastAPI application.

    Args:
        app               (FastAPI       , required): FastAPI application instance.
        config            (LoggerConfigPM, required): Logger configuration model.
        has_proxy_headers (bool | None   , optional): Whether to use proxy headers. Defaults to None.
        has_cf_headers    (bool | None   , optional): Whether to use Cloudflare headers. Defaults to None.

    Returns:
        Logger: Initialized Logger instance.
    """

    logger_loader = LoggerLoader(config=config)

    if has_proxy_headers is None:
        has_proxy_headers = config.http.has_proxy_headers

    if has_cf_headers is None:
        has_cf_headers = config.http.has_cf_headers

    app.add_middleware(ResponseHTTPInfoMiddleware)
    app.add_middleware(
        HttpAccessLogMiddleware,
        debug_msg_format_str=config.http.std.debug_msg_format_str,
        msg_format_str=config.http.std.msg_format_str,
    )
    app.add_middleware(
        RequestHTTPInfoMiddleware,
        has_proxy_headers=has_proxy_headers,
        has_cf_headers=has_cf_headers,
    )

    for _name, _handler in logger_loader.config.handlers.items():
        if _name == HTTP_ACCESS_STD_HANDLER_NAME:
            _handler.filter_ = http_std_filter
        elif (_name == HTTP_ACCESS_FILE_HANDLER_NAME) or (
            _name == HTTP_ERR_FILE_HANDLER_NAME
        ):
            _handler.filter_ = http_all_file_filter
            _handler.format_ = lambda record: http_file_format(
                record=record,
                format_str=config.http.file.format_str,
                tz=config.http.file.tz,
            )
        elif (_name == HTTP_ACCESS_JSON_HANDLER_NAME) or (
            _name == HTTP_ERR_JSON_HANDLER_NAME
        ):
            _handler.filter_ = http_all_file_filter
            _handler.format_ = http_json_format

    logger: Logger = logger_loader.load()
    return logger

LoggerConfigPM

Bases: LoggerConfigPM

Source code in src/beans_logging_fastapi/config.py

class LoggerConfigPM(BaseLoggerConfigPM):
    http: HttpConfigPM = Field(default_factory=HttpConfigPM)
    intercept: InterceptConfigPM = Field(default_factory=get_default_intercept)
    handlers: dict[str, LogHandlerPM] = Field(default_factory=get_default_handlers)

    @field_validator("handlers", mode="before")
    @classmethod
    def _check_handlers(cls, val: Any) -> dict[str, LogHandlerPM]:

        _default_handlers = get_default_handlers()

        if not val:
            val = _default_handlers
            return val

        if not isinstance(val, dict):
            raise TypeError(
                f"'handlers' attribute type {type(val).__name__} is invalid, must be a dict of <LogHandlerPM> or dict!"
            )

        for _key, _handler in val.items():
            if not isinstance(_handler, (LogHandlerPM, dict)):
                raise TypeError(
                    f"'handlers' attribute's '{_key}' key -> value type {type(_handler).__name__} is invalid, must be "
                    f"<LogHandlerPM> or dict!"
                )

            if isinstance(_handler, LogHandlerPM):
                val[_key] = _handler.model_dump(
                    by_alias=True, exclude_unset=True, exclude_none=True
                )

        _default_dict = {
            _key: _handler.model_dump(
                by_alias=True, exclude_unset=True, exclude_none=True
            )
            for _key, _handler in _default_handlers.items()
        }

        if _default_dict != val:
            val = utils.deep_merge(_default_dict, val)

        for _key, _handler in val.items():
            val[_key] = LogHandlerPM(**_handler)

        return val