FastAPI structured logs

Provides a wrapper over the structlog for additional
doping in json format, as well as output of standard
logs to the console, configuration via .env files
via pedantic.

In addition, middleware and a sentry configuration
are available for use in FastAPI applications.

Installation

You can use pip:

pip install fastapi-structlog

or poetry:

poetry add fastapi-structlog

Using logging

To implement logging in an application, two things are necessary:
define the configuration and use setup_logger:

import structlog
from fastapi_structlog import setup_logger

log_settings = LogSettings(
    logger='test-log-lib',
    json_logs=False,
    debug=True,
    types=['console'],
)
setup_logger(log_settings)

log = structlog.get_logger()

log.info('Hello, World!')

In this case, the entire configuration is determined manually by
creating an instance of LogSettings.

If you want to get the configuration from the environment variables,
then add LogSettings as a nested model in your settings:

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        arbitrary_types_allowed=True,
        env_ignore_empty=True,
        env_nested_delimiter='__',
        extra='ignore',
    )
    log: LogSettings

In this case, the entire configuration is taken from
the .env file by means of Pedantic.

To use the library in the minimum version, see
example №1 (docs_src/example_1.py).

Logger configuration

The logger configuration is implemented using the pydantic model LogSettings.
You can easily integrate it into your application settings model.

Logger configuration parameters:

• Name of the logger (LOGGER), default default
• Logging level (LOG_LEVEL), default INFO
• Flag that activates logging in json format (JSON_LOGS), default True
• Logging of the traceback in string form (TRACEBACK_AS_STR), default True
• The name of the file (FILENAME), default None
• The interval of writing to the file (WHEN), one of the S, M, H, D, W,
  default D (see TimedRotatingFileHandler)
• The number of saved files (BACKUP_COUNT), default 1
  (see TimedRotatingFileHandler)
• Activate DEBUG mode (DEBUG), default False
• New name for the key event (EVENT_KEY), default message
  (see renaming the event key)
• Enable logging (ENABLE), default True
• Methods for logging (METHODS), default ["get","delete","post","put","patch","options","head"]
• Types of logging (TYPES), one of the console, internal, syslog, file, default ["console"]
• Number of days to store the log entry (TTL), default 90

Syslog configuration parameters:

• Syslog server address (SYSLOG__HOST), default None
• Syslog server port (SYSLOG__PORT), default 6514

Database configuration parameters:

• Use async mode (DB__IS_ASYNC), default True
• Database connection string (DB__URL), default None
• Database user (DB__USER), default postgres
• Database password (DB__PASSWORD), default None
• Database host (DB__HOST), default localhost
• Database port (DB__PORT), default 5432
• Database name (DB__NAME), default postgres

When using fastapi_structlog.setup_logger, the settings must be
passed manually. An instance of the LogSettings class must be
passed as an argument. You can import:

from fastapi_structlog import LogSettings

This may be convenient in case of expanding the settings.

The implementation of logging settings allows you to use it in
existing service settings in the form of pydantic models.
Just add another field to your settings. You can use BaseSettingsModel
as a base class for your settings. This class already includes the basic
configuration, for example:

• delimiter as __
• ignoring environment variables with empty values

from pathlib import Path

import structlog
from pydantic_settings import SettingsConfigDict

from fastapi_structlog import BaseSettingsModel, LogSettings, setup_logger

logger = structlog.get_logger()

class Settings(BaseSettingsModel):
    log: LogSettings

    model_config = SettingsConfigDict(
        secrets_dir='/run/secrets' if Path('/run/secrets').exists() else None,
    )


def main() -> None:
    settings = Settings()

    setup_logger(settings.log)

    logger.info('Start')


if __name__ == '__main__':
    main()

For more information about the integration of logging
settings, see example №4 (docs_src/example_4.py).

Using logging in a FastAPI application

The use of the library in FasAPI application is for the most part
similar to that described in the section
"Logger configuration". To embed it in your
application, it is enough in the main function of the file main.py (or at
any other entry point) before starting uvicorn (or another server) or
directly in lifespan, use the registrar configuration functions. In all other
places, it is enough to use structlog.get_logger(log_name) to get the logger.
It will be configured according to the settings specified when launching the
application.

For more information about integrating logging into a FasAPI
application, see example №2 (docs_src/example_2.py) or example №8
(docs_src/example_8.py).

Using middleware in a FastAPI application

The library provides the following middleware:

• fastapi_structlog.middleware.CurrentScopeSetMiddleware - Creating
  the current context
• fastapi_structlog.middleware.StructlogMiddleware - Adds a
  pass-through request ID to the logs context. It should be used
  together with asgi_correlation_id.middleware.CorrelationIdMiddleware
  (github) or other
  middleware for generating end-to-end IDs.
• fastapi_structlog.middleware.AccessLogMiddleware - Allows you
  to keep a log of access to the service.

You can add this middleware to a FasAPI application as follows:

app = FastAPI(
    title="Example API",
    version="1.0.0",
    middleware=[
        Middleware(CurrentScopeSetMiddleware),
        Middleware(CorrelationIdMiddleware),
        Middleware(StructlogMiddleware),
        Middleware(AccessLogMiddleware),
    ],
)

It's worth noting that CurrentScopeSetMiddleware should come
first, and StructlogMiddleware after CorrelationIdMiddleware!

In order for AccessLogMiddleware to use structlog, you need
to initialize the logger before starting uvicorn (or another server).
Otherwise, the standard logger will be used, which will be reported
in the corresponding warning.

For more information about integrating logging into a FasAPI
application, see example №5 (docs_src/example_5.py).

Logging format

The AccessLogMiddleware middleware allows the use of the following parameters:

Using any parameter requires the inclusion of an expression
in the format string %(PARAM_NAME)s.

The following format is used by default:

%(client_addr)s - "%(request_line)s" %(status_code)s %(L)ss - "%(a)s"

Using Sentry in a FastAPI application

Using Sentry is completely similar to using logging. You can use
fastapi_structlog.sentry.init_sentry (with the env_prefix parameter)
to configure Sentry using the built-in model
(fastapi_structlog.sentry.SentrySettings) settings. In this case,
the parameters will be taken from the environment variables via
pydantic. If necessary, you can specify release, app_slug
and version. These parameters will be used when generating the
release parameter for transmission to Sentry. The release is
formed either from the explicitly passed release argument or
in the app_slug@version format.

Another Sentry configuration option is to use
fastapi_structlog.sentry.setup_sentry. However, in this case,
you must explicitly pass an instance of the settings
fastapi_structlog.sentry.SentrySettings and the release parameter. In addition,
you can pass the failed_request_status_codes (status codes should be reported to Sentry)
and service_integration (additional Sentry integrations) parameters.

The Sentry configuration parameters are as follows:

• dsn - Name of the data source
• env - The name of the environment, see fastapi_structlog.sentry.Environment
• traces_sample_rate - Uniform sampling rate
• log_integration - Flag for using logging integration, by default True.
  For more information, see sentry docs.
• log_integration_event_level - Parameter event_level for LoggingIntegration, by default None.
• log_integration_level - - Parameter level for LoggingIntegration, by default None.
• sql_integration - Flag for using
  SQLAlchemy integration,
  by default True.

By inheriting from fastapi_structlog.sentry.SentrySettings, you can
extend the Sentry configuration by adding the necessary
parameters
(see example №7 (docs_src/example_7.py).).

All parameters are optional! If there is no dsn Sentry will be ignored!

The implementation of Sentry settings in the settings of the general
project is similar to the one described in the section
"Logger configuration".

For more information about integrating Sentry into the FasAPI
application, see example №3 (docs_src/example_3.py).

To use sentry_sdk with requests to other APIs, see example №6 (docs_src/example_6.py).

In addition, both functions accept the service_integration parameter.
This parameter is intended for inter-service interaction and can be
useful for repeated requests (for example, to track the delay before
a repeat request, to track subqueries to other services, to monitor
the time for data serialization, and more).

Logging into the database

If you want to save logs to a database, then you need to declare a table
model. The library provides a basic LogModel model that you can import as:

from fastapi_structlog.db_handler import LogModel

This model inherits SQLModel and is a data model, that is, it does
not have the table=True parameter. Therefore, you need to make a class
that will inherit this model and add table=True. You can also add new
fields or write your own model altogether. However, the fastapi_structlog
relies on it to be the Inheritor of the SQLModel:

class Log(LogModel, table=True):
    """Log table."""

You should use one of the logger initialization functions and pass
your model there, as well as the database connection string. This may be
a different database from your main one.

engine = create_async_engine(DB_URL)

queue_listener = setup_logger(
    settings.log,
    model=Log,
    db_url=settings.log.db.make_url(),
)

logger = structlog.get_logger()

Use lifespan to start and stop queue_listener, for example:

@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    async with engine.begin() as conn:
        await conn.run_sync(Log.metadata.create_all)

    if queue_listener:
        queue_listener.start()

    yield

    if queue_listener:
        queue_listener.stop()

You must pass lifespan as a parameter to the FastAPI class. This is the
end of logging into the database.

You can read the sample code in docs_src/example_8.py

Examples

For other usage docs_src, see docs_src:

• example_1 - simple example
• example_2 - example using fastapi and uvicorn
• example_3 - example using fastapi, uvicorn and sentry
• example_4 - example with the integration of logging settings into the application settings
• example_5 - example using middleware
• example_6 - example using Sentry and nested calls to other APIs
• example_7 - expanding settings
• example_8 - logging logs to the database with redefinition of some fields

Dependencies

The following dependencies are used here:

• Python 3.9
• pydantic (for validate the settings)
• structlog (for set up the logger)
• sentry_sdk
• SQLModel (for database integration)