fastapi-celery-structlog 0.1.1

Structured logging utilities for FastAPI and Celery powered by structlog.

fastapi-celery-structlog

Structured logging helpers for FastAPI and Celery using structlog.

Features

• One-line logging setup for app and workers
• JSON logs in non-dev envs, readable console logs in dev
• Timed rotating file logs (logs/app.json.log)
• FastAPI request/response middleware with:
  • request id + correlation id propagation
  • response timing
  • optional response body capture for selected status codes
  • response body sanitization and size limits
• Celery task context binding via signals:
  • task id
  • task name
  • task args/kwargs
• Dynamic runtime resolution for:
  • env_value
  • base_dir

Installation

pip install fastapi-celery-structlog

Or with uv in this repo:

uv sync

Quick Start (FastAPI)

from fastapi import FastAPI
from fastapi_celery_structlog import configure_logging
from fastapi_celery_structlog.middlewares import (
    RequestIDMiddleware,
    StructlogRequestMiddleware,
)

configure_logging()

app = FastAPI()
app.add_middleware(StructlogRequestMiddleware, project_name="my-service")
app.add_middleware(RequestIDMiddleware)

Middleware order is intentional. FastAPI/Starlette executes middleware in reverse registration order on incoming requests, so RequestIDMiddleware must be registered after StructlogRequestMiddleware to run first at request time (and after it on response). If you reverse the registration order, StructlogRequestMiddleware builds its request log context before RequestIDMiddleware has populated scope["state"]["request_id"]/scope["state"]["correlation_id"], so request IDs will be missing in logs.

Quick Start (Celery)

from celery import Celery
from fastapi_celery_structlog.celery import configure_celery_logging

celery_app = Celery("my-worker")

configure_celery_logging(celery_app=celery_app)

This configures logging and binds task context using task_prerun/task_postrun.

Example Project

A runnable FastAPI + Celery demo is included at:

• examples/fastapi_celery_demo

See:

• examples/fastapi_celery_demo/README.md

Dynamic env_value and base_dir

Both configure_logging(...) and configure_celery_logging(...) accept optional:

• env_value
• base_dir

If omitted, values are resolved dynamically.

env_value resolution order

1. Explicit argument (env_value=...)
2. Environment variables (first match):
   • ENV
   • APP_ENV
   • FASTAPI_ENV
   • ENVIRONMENT
3. Fallback: "development"

base_dir resolution order

1. Explicit argument (base_dir=...)
2. Environment variables (first match):
   • FASTAPI_CELERY_STRUCTLOG_BASE_DIR
   • BASE_DIR
   • APP_BASE_DIR
3. Project root auto-detection by walking up from current path and finding:
   • pyproject.toml or
   • .git
4. Fallback: current working directory

Examples

# Fully dynamic
configure_logging()

# Explicit override
configure_logging(base_dir="/srv/my-service", env_value="production")

configure_celery_logging(
    celery_app=celery_app,
    base_dir="/srv/my-service",
    env_value="production",
    worker_hijack_root_logger=False,
)

API

configure_logging(base_dir=None, env_value=None) -> None

Configures Python logging + structlog processors for the application.

configure_celery_logging(celery_app=None, base_dir=None, env_value=None, worker_hijack_root_logger=False) -> None

Configures logging for Celery workers and wires Celery signal handlers for task context binding.

Middlewares

RequestIDMiddleware

• Reads incoming x-request-id / request-id
• Reads incoming x-correlation-id / correlation-id
• Generates missing IDs
• Adds both IDs to response headers

StructlogRequestMiddleware

Logs structured request/response events with environment, user, service, request, response, system, and duration fields.

Key options:

• project_name
• ip_logging_enabled
• log_4xx_level
• max_body_bytes
• capture_body_statuses
• capture_only_json
• include_endpoint_source
• cancelled_status_code
• cancelled_is_error

Log Output Behavior

• Development-like envs (dev, local, development):
  • console renderer for readability
• Other envs:
  • JSON renderer
• Always writes rotating JSON logs to:
  • logs/app.json.log

Celery Notes

• Task context vars are bound in task_prerun and cleared in task_postrun.
• If you pass a celery_app, worker_hijack_root_logger is set on celery_app.conf.

Running Tests

.venv/bin/python -m unittest discover -s tests -v

Project Structure

fastapi_celery_structlog/
  __init__.py
  runtime.py
  settings.py
  utils.py
  celery/
    __init__.py
    logger.py
  middlewares/
    __init__.py
    request_id.py
    request_response.py
examples/
  fastapi_celery_demo/
    api.py
    celery_app.py
    tasks.py