Metadata-Version: 2.4
Name: fastapi-fabric-audit
Version: 0.1.0
Summary: Immutable audit logging for FastAPI — fire-and-forget event emission, pluggable writers, and a query API.
Project-URL: Homepage, https://git.punakawanlab.id/PunakawanLab/fastapi-fabric
Project-URL: Repository, https://git.punakawanlab.id/PunakawanLab/fastapi-fabric
Author-email: Yoiq S Rambadian <yoiqsram@punakawanlab.id>
License-Expression: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: System :: Logging
Requires-Python: >=3.10
Requires-Dist: fastapi-fabric-core<0.2.0,>=0.1.0
Description-Content-Type: text/markdown

# fastapi-fabric-audit

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

## Install

```bash
pip install fastapi-fabric[auth,audit]
```

## Quick start

```python
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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.

```python
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

```yaml
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
```
