SQLite event store

SQLiteEventStore is an event sink that writes every event to a local SQLite database, so execution history survives restarts and can be inspected later. It uses only the Python standard library.

Usage#

from relprim import EventEmitter, SQLiteEventStore, resilient

event_store = SQLiteEventStore("relprim-events.db")


@resilient(retries=3, timeout=10, events=EventEmitter(sinks=(event_store,)))
async def call_provider(prompt: str) -> str:
    return await provider.generate(prompt)

The database and schema are created lazily, before the first operation that needs them. To fail at startup instead, initialize explicitly:

await event_store.initialize()

Parameter	Default	Notes
database	(required)	File path, as str or Path. ":memory:" is rejected.
busy_timeout_seconds	5.0	How long a connection waits for a lock. Finite, 0 or more.

Reading history#

history = await event_store.history(operation_name="call_provider", limit=100)

for stored in history:
    print(stored.sequence_id, stored.event.event_type, stored.event.payload)

history() returns a tuple of StoredEvent, each with a sequence_id assigned by SQLite and the original StructuredEvent. Results are always in ascending sequence order.

Argument	Default	Effect
operation_name	None	Only events of this operation.
event_types	None	Only these EventType values; an empty tuple returns nothing.
after_sequence_id	None	Only events after this sequence id, for pagination.
limit	100	Maximum number of events, greater than 0.

Filters combine:

from relprim import EventType

failures = await event_store.history(
    operation_name="call_provider",
    event_types=(EventType.ATTEMPT_FAILED, EventType.OPERATION_FAILED),
)

Pagination#

Continue from the last sequence id of the previous page:

page = await event_store.history(limit=100)
while page:
    for stored in page:
        handle(stored)
    page = await event_store.history(after_sequence_id=page[-1].sequence_id, limit=100)

Retention#

from datetime import UTC, datetime, timedelta

cutoff = datetime.now(UTC) - timedelta(days=30)
deleted = await event_store.delete_before(cutoff)

Events strictly older than the cutoff are deleted, and the number of deleted events is returned. The cutoff must be timezone-aware; a naive datetime, or one whose tzinfo returns no UTC offset, raises ValueError. RelPrim never runs retention automatically; your application decides when.

Storage format#

Each row holds the sequence id, event type, operation name, timestamp and the JSON payload. Timestamps are stored in UTC with microsecond precision and read back as UTC. Because event payloads only contain JSON scalars, every RelPrim event is stored and read back unchanged. Queries use bound parameters only.

Concurrency#

SQLite work runs in a worker thread, outside the event loop. Each operation opens its own connection, and the database uses WAL mode with the configured busy timeout. Several stores can write to the same file, including while the database is being created for the first time.

SQLite still serializes writers. The store suits normal application event volumes, not high-throughput telemetry pipelines. Do not place the database on a network filesystem.

Cancellation#

Cancelling the task that awaits emit(), history() or delete_before() raises asyncio.CancelledError immediately, but the statement already running in the worker thread is not interrupted: an event being written may still be committed. Connections are always closed and the store remains usable.

Failures#

The store follows normal event delivery rules: if writing fails, the sqlite3 error propagates to the caller unchanged and the operation stops. RelPrim does not hide persistence failures. Wrap the store if event storage must never affect business operations.

Schema compatibility#

The store keeps its schema version in a relprim_schema table and initializes or migrates it automatically.

• Databases written by RelPrim 0.10 and 0.11 use schema version 1 and are read without migration.
• Within 1.x, a database remains readable by the same or a later release.
• Reading a database with an older release is not supported once a newer release has stored event types the older one does not know.
• A database with a newer, unsupported schema version raises SQLiteEventStoreSchemaError instead of being downgraded.

Why :memory: is not supported#

Every operation uses a fresh connection, and an SQLite :memory: database belongs to a single connection, so later operations would not see earlier events. For tests, use a temporary file such as pytest’s tmp_path.