fastapi-fabric-audit 0.1.0

Immutable audit logging for FastAPI — fire-and-forget event emission, pluggable writers, and a query API.

fastapi-fabric-audit

Immutable audit logging for FastAPI — fire-and-forget event emission, pluggable writers, and a query API.

Install

pip install fastapi-fabric[auth,audit]

Quick start

from fastapi_fabric.audit.protocol import set_writer
from fastapi_fabric.audit.writers.database import DatabaseAuditWriter
from fastapi_fabric.audit.writers.queued import QueuedAuditWriter
from fastapi_fabric.audit.writer import emit_audit
from sqlalchemy.ext.asyncio import async_sessionmaker

# Wire at startup
session_factory = async_sessionmaker(engine, expire_on_commit=False)
writer = QueuedAuditWriter(DatabaseAuditWriter(session_factory))
set_writer(writer)
await writer.start()

# Emit in a route (fire-and-forget — not awaited, scheduled via BackgroundTasks)
@app.post("/posts")
async def create_post(background_tasks: BackgroundTasks, principal = Depends(get_current_principal)):
    post = await do_create()
    emit_audit(
        background_tasks,
        action="post.created",
        principal=principal,
        resource_type="post",
        resource_id=str(post.id),
    )
    return post

# Flush on shutdown
await writer.flush()

Router

from fastapi_fabric.audit import create_audit_router

app.include_router(create_audit_router())

Method	Path	Description
GET	/api/v1/audit	Query audit events (audit:read)
GET	/api/v1/audit/{id}	Get single event (audit:read)

Filter params: actor_id, actor_type, action (prefix with . for prefix match), resource_type, resource_id, from, to, has_flag.

Writers

File writer

Appends one JSON line per event to a daily-rotating file:

from fastapi_fabric.audit.writers.file import FileAuditWriter

set_writer(FileAuditWriter("logs/audit.jsonl", backup_count=90))

Database writer

Inserts into the audit_events table:

from fastapi_fabric.audit.writers.database import DatabaseAuditWriter

set_writer(DatabaseAuditWriter(session_factory, retention_days=30))

Purges events older than retention_days at startup.

Queued writer

Wraps any writer with an async queue to decouple writes from the request path:

from fastapi_fabric.audit.writers.queued import QueuedAuditWriter

writer = QueuedAuditWriter(inner_writer, maxsize=10000)
await writer.start()   # call at app startup
await writer.flush()   # call at app shutdown

Composite writer

Fans out to multiple writers simultaneously. One failing does not stop the others:

from fastapi_fabric.audit.writers.composite import CompositeAuditWriter

set_writer(QueuedAuditWriter(
    CompositeAuditWriter([
        FileAuditWriter("logs/audit.jsonl"),
        DatabaseAuditWriter(session_factory),
    ])
))

Emit API

Neither function is awaited — both schedule the write and return immediately, so they never add latency to the response path.

from fastapi_fabric.audit.writer import emit_audit, emit_audit_nowait

# With BackgroundTasks (recommended in routes)
emit_audit(background_tasks, action="user.deactivated", principal=principal,
           resource_type="user", resource_id=user_id)

# Without BackgroundTasks (e.g. in background jobs) — schedules an asyncio task
emit_audit_nowait(action="job.completed", principal=None, detail={"job_id": jid})

Actor fields (id, type, email, name, flags) are automatically populated from principal.

Config reference

audit:
  db_retention_days: 30      # events older than this are purged at startup
  queue_maxsize: 10000        # drops events (with warning) when full
  log_file: logs/audit.jsonl  # path for FileAuditWriter
  log_retention_days: 90      # rotated file backups to keep