fastapi-celery-structlog 0.1.0

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(RequestIDMiddleware)
app.add_middleware(StructlogRequestMiddleware, project_name="my-service")

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