🚸 Examples

examples/configs/logger.yml:

logger:
  app_name: "fastapi-app"
  level:
    base: TRACE
  format_str: "[{time:YYYY-MM-DD HH:mm:ss.SSS Z} | {extra[level_short]:<5} | {name}:{line}]: {message}"
  http:
    std:
      msg_format_str: '<n><w>[{request_id}]</w></n> {client_host} {user_id} "<u>{method} {url_path}</u> HTTP/{http_version}" {status_code} {content_length}B {response_time}ms'
      err_msg_format_str: '<n><w>[{request_id}]</w></n> {client_host} {user_id} "<u>{method} {url_path}</u> HTTP/{http_version}" <n>{status_code}</n>'
      debug_msg_format_str: '<n>[{request_id}]</n> {client_host} {user_id} "<u>{method} {url_path}</u> HTTP/{http_version}"'
    file:
      format_str: '{client_host} {request_id} {user_id} [{datetime}] "{method} {url_path} HTTP/{http_version}" {status_code} {content_length} "{h_referer}" "{h_user_agent}" {response_time}'
      tz: "localtime"
    has_proxy_headers: false
    has_cf_headers: false
  intercept:
    mute_modules: ["uvicorn.access"]
  handlers:
    http_access_std_handler:
      enabled: true
      format: "[<c>{time:YYYY-MM-DD HH:mm:ss.SSS Z}</c> | <level>{extra[level_short]:<5}</level> ]: <level>{message}</level>"
    http_access_file_handler:
      enabled: true
      sink: "http/{app_name}.http-access.log"
    http_err_file_handler:
      enabled: true
      sink: "http/{app_name}.http-err.log"
    http_access_json_handler:
      enabled: true
      sink: "http.json/{app_name}.http-access.json.log"
    http_err_json_handler:
      enabled: true
      sink: "http.json/{app_name}.http-err.json.log"

examples/.env:

ENV=development
DEBUG=true

examples/config.py:

import os

from pydantic_settings import BaseSettings

from potato_util import io as io_utils
from beans_logging_fastapi import LoggerConfigPM

from logger import logger

_config_dict = {}
_configs_dir = os.path.join(os.getcwd(), "configs")
if os.path.isdir(_configs_dir):
    _config_dict = io_utils.read_all_configs(configs_dir=_configs_dir)


class MainConfig(BaseSettings):
    logger: LoggerConfigPM = LoggerConfigPM()


try:
    config = MainConfig(**_config_dict)
except Exception:
    logger.exception("Failed to load config:")
    raise SystemExit(1)


__all__ = [
    "MainConfig",
    "config",
]

examples/logger.py:

from beans_logging_fastapi import logger

__all__ = [
    "logger",
]

examples/router.py:

from pydantic import validate_call
from fastapi import FastAPI, APIRouter, HTTPException
from fastapi.responses import RedirectResponse

router = APIRouter()


@router.get("/")
def root():
    return {"Hello": "World"}


@router.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}


@router.get("/continue", status_code=100)
def get_continue():
    return {}


@router.get("/redirect")
def redirect():
    return RedirectResponse("/")


@router.get("/error")
def error():
    raise HTTPException(status_code=500)


@validate_call(config={"arbitrary_types_allowed": True})
def add_routers(app: FastAPI) -> None:
    """Add routers to FastAPI app.

    Args:
        app (FastAPI): FastAPI app instance.
    """

    app.include_router(router)

    return


__all__ = ["add_routers"]

examples/bootstrap.py:

# Standard libraries
from typing import Any
from collections.abc import Callable

# Third-party libraries
import uvicorn
from uvicorn._types import ASGIApplication
from pydantic import validate_call
from fastapi import FastAPI

from beans_logging_fastapi import add_logger

# Internal modules
from __version__ import __version__
from config import config
from lifespan import lifespan
from router import add_routers


def create_app() -> FastAPI:
    """Create FastAPI application instance.

    Returns:
        FastAPI: FastAPI application instance.
    """

    app = FastAPI(lifespan=lifespan, version=__version__)

    # Add logger before any other components:
    add_logger(app=app, config=config.logger)

    # Add any other components after logger:
    add_routers(app=app)

    return app


@validate_call(config={"arbitrary_types_allowed": True})
def run_server(app: FastAPI | ASGIApplication | Callable[..., Any] | str) -> None:
    """Run uvicorn server.

    Args:
        app (FastAPI            |
             ASGIApplication    |
             Callable[..., Any] |
             str                 , required): FastAPI application instance or ASGI application or import string.
    """

    uvicorn.run(
        app=app,
        host="0.0.0.0",  # nosec B104
        port=8000,
        access_log=False,  # Disable default uvicorn access log
        server_header=False,
        proxy_headers=False,
        forwarded_allow_ips="*",
    )

    return


__all__ = [
    "create_app",
    "run_server",
]

examples/main.py:

#!/usr/bin/env python

# Third-party libraries
from dotenv import load_dotenv

load_dotenv(override=True)

# Internal modules
from bootstrap import create_app, run_server  # noqa: E402
from logger import logger  # noqa: E402

app = create_app()


def main() -> None:
    """Main function."""

    run_server(app=app)
    return


if __name__ == "__main__":
    logger.info("Starting server from 'main.py'...")
    main()


__all__ = ["app"]