> ## Documentation Index
> Fetch the complete documentation index at: https://docs.auditrails.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK — AuditRails Sync and Async Audit Client

> Integrate AuditRails into any Python application. Sync and async clients, zero dependencies for sync, Django and FastAPI integration examples included.

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

<Tabs>
  <Tab title="Sync (zero dependencies)">
    ```bash theme={null}
    pip install auditrails
    ```

    The synchronous client runs a background daemon thread and flushes via `atexit`. Requires **Python 3.9+**.
  </Tab>

  <Tab title="Async (httpx)">
    ```bash theme={null}
    pip install auditrails[async]
    ```

    Installs `auditrails` plus `httpx`. Use `AsyncAuditRails` in `asyncio`-based applications (FastAPI, aiohttp, etc.).
  </Tab>
</Tabs>

## Initialization

<Tabs>
  <Tab title="Sync">
    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.

    ```python theme={null}
    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.
  </Tab>

  <Tab title="Async">
    Create an `AsyncAuditRails` instance and call `await audit.start()` before logging any events. Use `await audit.close()` during application shutdown to flush remaining events.

    ```python theme={null}
    from auditrails import AsyncAuditRails, Config

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

    # Must be called before logging
    await audit.start()
    ```
  </Tab>
</Tabs>

### 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**.

```python theme={null}
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.

<Tabs>
  <Tab title="Sync">
    ```python theme={null}
    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..."
    ```
  </Tab>

  <Tab title="Async">
    ```python theme={null}
    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..."
    ```
  </Tab>
</Tabs>

### Direct batch logging

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

```python theme={null}
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

```python theme={null}
# 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.

```python theme={null}
# 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
```

```python theme={null}
# settings.py
MIDDLEWARE = [
    # ... other middleware ...
    "myapp.middleware.AuditMiddleware",
]
```

### FastAPI

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

```python theme={null}
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.

```python theme={null}
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:

```python theme={null}
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).

```python theme={null}
await audit.close()  # Flush + stop background tasks
```

<Note>
  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.
</Note>
