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

# Audit client

> Python AuditClient API reference for compliance-grade audit logging.

# Audit client

```python theme={null}
import fact0

client = fact0.Client(api_key="f0_live_...")
audit = client.audit  # or client.audit directly via convenience methods
```

## Constructor options

| Parameter           | Default                | Description                               |
| ------------------- | ---------------------- | ----------------------------------------- |
| `api_key`           | required               | Write or read-scoped `f0_live_*` key      |
| `base_url`          | `https://api.fact0.io` | API origin                                |
| `batch_max_size`    | `100`                  | Max events per flush                      |
| `batch_max_wait_ms` | `500`                  | Max wait before flush                     |
| `raise_on_error`    | `False`                | Raise on transport failure                |
| `dead_letter_path`  | `None`                 | JSONL path for failed batches             |
| `sync`              | `False`                | Synchronous ingest (`X-Fact0-Sync: true`) |
| `poll_receipts`     | `True`                 | Poll async receipts after ingest          |

Environment variable: `FACT0_API_KEY` (optional if `api_key` is passed). Override the API origin with the `base_url` parameter.

## Write methods

### `log(**fields) -> None`

Validate and enqueue one event. Returns immediately.

### `log_batch(events: list[dict]) -> BatchResult | AsyncReceipt`

Send up to 1000 events. Uses batch endpoint.

### `flush() -> None`

Block until the in-memory buffer is drained.

### `close() -> None`

Stop background flusher and drain. Registered with `atexit`.

## Read methods

### `get_event(event_id: str) -> dict`

`GET /v1/events/{id}`

### `list_events(**filters) -> dict`

`GET /v1/events` - filters: `actor_id`, `actor_type`, `action`, `resource_id`, `outcome`, `from`, `to`, `page`, `page_size`.

### `get_receipt(receipt_id: str) -> dict`

`GET /v1/receipts/{id}`

### `wait_for_receipt(receipt_id: str, timeout_s=30) -> dict`

Poll until `status` is `committed` or `failed`.

## Verify methods

### `verify(from_=None, to=None, scan_all=False) -> dict`

`GET /v1/verify`

### `verify_event(event_id: str) -> dict`

`GET /v1/events/{id}/verify`

## Export methods

### `export_pdf(from_=None, to=None) -> bytes`

`GET /v1/export/pdf`

### `export_evidence_pack(from_=None, to=None) -> bytes`

`GET /v1/export/evidence-pack`

## Stream

### `stream_events() -> Iterator[dict]`

SSE iterator over `GET /v1/events/stream`.

## Error handling

| Exception               | When                                         |
| ----------------------- | -------------------------------------------- |
| `fact0.ValidationError` | Invalid event fields at `log()`              |
| `fact0.TransportError`  | HTTP failure (only if `raise_on_error=True`) |
| `fact0.Fact0Error`      | Base class                                   |

By default transport errors are logged and optionally written to dead-letter JSONL.

***

## Production Integration Recipes

### 1. Production FastAPI lifespan Hook

Always configure the `lifespan` hook to ensure the background batching workers are safely stopped and buffers are drained when the application shuts down.

```python theme={null}
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
import fact0

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize AsyncClient; it runs a background buffer worker
    app.state.fact0 = fact0.AsyncClient(
        api_key="f0_live_...",
        sync=False # defaults to async background batching
    )
    yield
    # CRITICAL: Close client on shutdown to drain all pending buffers
    await app.state.fact0.close()

app = FastAPI(lifespan=lifespan)

# Helper to inject client into routers
def get_fact0(request):
    return request.app.state.fact0
```

### 2. Dead-Letter Queue (DLQ) Recovery & Replay

If a network partition occurs and `dead_letter_path` is configured, Fact0 writes unsent events as JSONL rows to the specified file. Use this script to replay and re-submit them to the API once connectivity is restored:

```python theme={null}
import json
import fact0

def replay_dead_letters(file_path: str, api_key: str):
    client = fact0.Client(api_key=api_key, sync=True) # Use sync ingest for verification
    replayed_count = 0
    events = []
    
    with open(file_path, "r") as f:
        for line in f:
            if line.strip():
                events.append(json.loads(line))
                
    # Batch replay up to 1000 events per call
    for i in range(0, len(events), 1000):
        batch = events[i:i+1000]
        client.audit.log_batch(batch)
        replayed_count += len(batch)
        
    client.close()
    print(f"Successfully replayed {replayed_count} events from DLQ.")
```

### 3. Unit Testing & Mocking in pytest

Do not send live HTTP requests to `api.fact0.io` during unit tests. Mock the transport or the client using this fixture:

```python theme={null}
import pytest
from unittest.mock import MagicMock
import fact0

@pytest.fixture
def mock_fact0():
    client = MagicMock(spec=fact0.Client)
    client.audit = MagicMock()
    client.telemetry = MagicMock()
    return client

def test_delete_document_service(mock_fact0):
    # Your service function that takes client and deletes doc
    # delete_document(doc_id="doc_123", fact0_client=mock_fact0)
    
    # Assert your application code called log with correct attributes
    mock_fact0.audit.log.assert_called_once_with(
        actor={"id": "admin", "type": "human"},
        action="document.delete",
        resource={"id": "doc_123", "type": "document"},
        outcome="success"
    )
```