Python SDK — AuditRails Sync and Async Audit Client

The AuditRails Python SDK supports both synchronous and asynchronous Python applications. The synchronous client uses only the standard library (urllib) so it installs with zero additional dependencies. The async client adds httpx for non-blocking HTTP and is installed via an optional extra. Both clients are thread-safe, buffer events automatically, and flush on process exit so you never lose an audit record.

Installation

Sync (zero dependencies)
Async (httpx)

pip install auditrails

The synchronous client runs a background daemon thread and flushes via atexit. Requires Python 3.9+.

pip install auditrails[async]

Installs auditrails plus httpx. Use AsyncAuditRails in asyncio-based applications (FastAPI, aiohttp, etc.).

Initialization

Sync
Async

Create an AuditRails instance with a Config object. The only required field is api_key. The client starts its background flush thread immediately on construction.

from auditrails import AuditRails, Config

audit = AuditRails(Config(api_key="at_live_..."))

Instantiate the client once at module level (or in your application factory) and import it wherever you need it.

Create an AsyncAuditRails instance and call await audit.start() before logging any events. Use await audit.close() during application shutdown to flush remaining events.

from auditrails import AsyncAuditRails, Config

audit = AsyncAuditRails(Config(api_key="at_live_..."))

# Must be called before logging
await audit.start()

Configuration options

Option	Type	Default	Description
`api_key`	`str`	(required)	Your AuditRails API key.
`base_url`	`str`	`https://api.auditrails.io`	Override the API endpoint.
`batch_size`	`int`	`100`	Maximum events per HTTP request.
`flush_interval`	`float`	`1.0` (seconds)	How often the buffer is flushed, in seconds.
`max_retries`	`int`	`3`	Retry attempts on `5xx` responses.
`timeout`	`float`	`10.0` (seconds)	Per-request timeout in seconds.

Logging events

Buffered logging (recommended)

audit.log() adds an AuditEvent to an in-memory buffer and returns immediately. The buffer is flushed in the background on the flush_interval schedule. This method never raises an exception.

from auditrails import AuditRails, Config, AuditEvent

audit = AuditRails(Config(api_key="at_live_..."))

audit.log(AuditEvent(
    action="user.login",
    actor_id="user_123",
    resource="session/sess_abc",
    metadata={"ip": "203.0.113.1", "method": "oauth2"},
))

Direct (immediate) logging

audit.log_direct() sends the event immediately and returns the API response. It raises AuditRailsError on failure.

Sync
Async

from auditrails import AuditRails, Config, AuditEvent

response = audit.log_direct(AuditEvent(
    action="document.deleted",
    actor_id="user_456",
    resource="document/doc-789",
))
print(response.log_id)  # "01HXYZ..."

from auditrails import AsyncAuditRails, Config, AuditEvent

response = await audit.log_direct(AuditEvent(
    action="document.deleted",
    actor_id="user_456",
    resource="document/doc-789",
))
print(response.log_id)  # "01HXYZ..."

Direct batch logging

Send multiple events in a single HTTP request, bypassing the buffer. Raises on error.

await audit.log_batch_direct([
    AuditEvent(action="document.created", actor_id="user_123", resource="document/doc-001"),
    AuditEvent(action="document.shared",  actor_id="user_123", resource="document/doc-001"),
])

Manual flush

# Sync
audit.flush()

# Async
await audit.flush()

Framework integration

Django

Add AuditMiddleware to your MIDDLEWARE list in settings.py. The middleware logs every request after the response has been generated, capturing the final HTTP status code.

# myapp/middleware.py
from auditrails import AuditRails, Config, AuditEvent

audit = AuditRails(Config(api_key="at_live_..."))

class AuditMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)

        audit.log(AuditEvent(
            action=f"{request.method.lower()}.{request.resolver_match.url_name}",
            actor_id=str(request.user.id) if request.user.is_authenticated else None,
            resource=f"{request.method} {request.path}",
            metadata={
                "ip": request.META.get("REMOTE_ADDR"),
                "status": response.status_code,
            },
        ))

        return response

# settings.py
MIDDLEWARE = [
    # ... other middleware ...
    "myapp.middleware.AuditMiddleware",
]

FastAPI

Use a lifespan context manager to start and close the async client alongside your application.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from auditrails import AsyncAuditRails, Config, AuditEvent

audit = AsyncAuditRails(Config(api_key="at_live_..."))

@asynccontextmanager
async def lifespan(app: FastAPI):
    await audit.start()   # Start background tasks
    yield
    await audit.close()   # Flush remaining events on shutdown

app = FastAPI(lifespan=lifespan)

@app.delete("/documents/{doc_id}")
async def delete_document(doc_id: str, user_id: str):
    await do_delete(doc_id)

    audit.log(AuditEvent(
        action="document.deleted",
        actor_id=user_id,
        resource=f"document/{doc_id}",
    ))

    return {"status": "deleted"}

Error handling

log() never raises. For log_direct() and log_batch_direct(), catch AuditRailsError to inspect the failure details. AuditRailsTimeoutError is a subclass specifically for request timeouts.

from auditrails import AuditRailsError, AuditRailsTimeoutError, AuditEvent

try:
    audit.log_direct(AuditEvent(action="user.login", actor_id="user_123"))
except AuditRailsTimeoutError:
    print("Request timed out")
except AuditRailsError as e:
    print(e.status_code)  # HTTP status, e.g. 401
    print(e.code)         # Machine-readable code, e.g. "INVALID_API_KEY"
    print(e.request_id)   # AuditRails request ID for support
    print(e.doc_url)      # Link to relevant documentation
    print(e.retryable)    # True if retrying might succeed

Graceful shutdown

Sync: The background daemon thread is registered with Python’s atexit module. When your process exits normally, it automatically flushes any buffered events. You can also flush and shut down manually:

audit.close()  # Flush + stop background thread

Async: The atexit hook is not available for async clients. You must call await audit.close() explicitly — for example, in your framework’s shutdown lifecycle hook (as shown in the FastAPI lifespan example above).

await audit.close()  # Flush + stop background tasks

For async applications, always wire await audit.close() into your framework’s shutdown hook. If the event loop is torn down before close() is called, buffered events that have not yet been flushed will be lost.

​Installation

​Initialization

​Configuration options

​Logging events

​Buffered logging (recommended)

​Direct (immediate) logging

​Direct batch logging

​Manual flush

​Framework integration

​Django

​FastAPI

​Error handling

​Graceful shutdown

Installation

Initialization

Configuration options

Logging events

Buffered logging (recommended)

Direct (immediate) logging

Direct batch logging

Manual flush

Framework integration

Django

FastAPI

Error handling

Graceful shutdown