Audit client

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.

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:

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:

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"
    )

​Audit client

​Constructor options

​Write methods

​log(**fields) -> None

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

​flush() -> None

​close() -> None

​Read methods

​get_event(event_id: str) -> dict

​list_events(**filters) -> dict

​get_receipt(receipt_id: str) -> dict

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

​Verify methods

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

​verify_event(event_id: str) -> dict

​Export methods

​export_pdf(from_=None, to=None) -> bytes

​export_evidence_pack(from_=None, to=None) -> bytes

​Stream

​stream_events() -> Iterator[dict]

​Error handling

​Production Integration Recipes

​1. Production FastAPI lifespan Hook

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

​3. Unit Testing & Mocking in pytest