fastapi-audit 0.1.0

Audit logging package for FastAPI applications

fastapi-audit

Audit logging package for FastAPI applications.

Features

• HTTP Request/Response Capture: Logs method, path, status code, response time, IP, and user agent
• JWT-based Actor Identification: Extracts actor ID, type, and email from JWT tokens
• ORM-level Diff Tracking: Captures INSERT, UPDATE, and DELETE operations across application sessions
• Tenant-Aware Context Support: Works with application-provided tenant context
• Sensitive Data Redaction: Automatically redacts passwords, tokens, secrets, and similar fields
• Fire-and-Forget Writes: Writes audit records asynchronously to avoid adding request latency
• Manual Audit Logging: Supports background tasks and non-HTTP contexts

Installation

From PyPI

pip install fastapi-audit

Local Development

pip install -e /path/to/audit

Quick Start

from fastapi import FastAPI
from fastapi_audit import AuditMiddleware, AuditConfig

app = FastAPI()

app.add_middleware(
    AuditMiddleware,
    config=AuditConfig(
        control_db_url="postgresql+asyncpg://user:pass@localhost/audit_db"
    )
)

The legacy audit import path still works for compatibility, but fastapi_audit is the preferred public import path.

Configuration

from fastapi_audit import AuditConfig

config = AuditConfig(
    # Required: Connection URL for the audit database
    control_db_url="postgresql+asyncpg://user:pass@localhost/audit_db",

    # Optional: Fields to redact (merged with defaults)
    redact_fields={"password", "token", "secret", "custom_field"},

    # Optional: Paths to exclude from audit logging
    exclude_paths={"/health", "/metrics"},

    # Optional: Map incoming actor_type values to canonical public values
    actor_type_aliases={"hashira": "platform_admin", "ops_admin": "platform_admin"},

    # Optional: Capture request/response bodies
    capture_request_body=True,
    capture_response_body=True,

    # Optional: Enable ORM diff capture
    capture_orm_diffs=True,

    # Optional: Log requests without authenticated actors
    log_anonymous=False,
)

Database Migration

Create the audit_logs table in your audit database:

# Using Alembic
alembic upgrade head

# Or run the SQL migration directly
# See migrations/versions/001_create_audit_logs.py

Tenant Context

If your application is tenant-aware, the middleware can read tenant context from:

# Example: in your own tenant-resolution middleware
request.state.tenant = Tenant(tenant_id="...", tenant_slug="example-tenant")

The tenant object should have tenant_id and tenant_slug attributes.

JWT Token Requirements

The middleware extracts actor information from JWT tokens in the Authorization header. Expected JWT claims:

Claim	Required	Description
sub	Yes	User ID
actor_type	No	platform_admin, tenant_user, or anonymous (default)
email	No	User email address

Example JWT payload:

{
  "sub": "user-123",
  "actor_type": "platform_admin",
  "email": "[e-mail]"
}

Legacy or organization-specific actor types can be mapped to canonical public values via AuditConfig.actor_type_aliases. By default, "hashira" maps to "platform_admin".

Manual Audit Logging

For background tasks or events outside the HTTP lifecycle:

from fastapi_audit import audit_log, ActorType
from sqlalchemy.ext.asyncio import AsyncSession

async def provision_tenant(db: AsyncSession):
    await audit_log(
        db=db,
        action="tenant.provisioned",
        actor_id="admin-user-id",
        actor_type=ActorType.PLATFORM_ADMIN,
        actor_email="[e-mail]",
        tenant_id="tenant-uuid",
        tenant_slug="example-tenant",
        metadata={"plan": "pro", "region": "us-east"}
    )

Environment Variables

All configuration can be set via environment variables with the AUDIT_ prefix:

export AUDIT_CONTROL_DB_URL="postgresql+asyncpg://..."
export AUDIT_CAPTURE_ORM_DIFFS="true"
export AUDIT_LOG_ANONYMOUS="false"

Architecture

HTTP Layer

The middleware intercepts requests and captures:

• Request metadata such as method, path, and query params
• Response metadata such as status code and response time
• Actor information from JWT claims
• Optional tenant context from request.state.tenant

ORM Layer

SQLAlchemy event listeners capture database changes:

• Class-level listeners on AsyncSession work across application sessions
• Context variables isolate diffs per request
• INSERT, UPDATE, and DELETE operations are recorded

Storage

All audit logs are written to a single audit_logs table in the audit database, providing a unified view of application activity.

Requirements

• Python 3.11+
• FastAPI / Starlette
• SQLAlchemy 2.0+ with async support
• Pydantic v2
• asyncpg

License

MIT