praval.observability.storage.sqlite_store

SQLite storage backend for traces.

Stores traces locally for querying and analysis.

Functions

get_trace_store()	Get the global trace store instance.
reset_trace_store()	Reset the global trace store to None.
set_trace_store(store)	Bind the store owned by the active explicit lifecycle.

Classes

SQLiteTraceStore(db_path, *[, busy_timeout_ms])	SQLite-based trace storage.
StorableSpan(*args, **kwargs)	Temporary input boundary retained until the O5 exporter migration.

class praval.observability.storage.sqlite_store.StorableSpan(*args, **kwargs)

Bases: Protocol

Temporary input boundary retained until the O5 exporter migration.

span_id: str

trace_id: str

parent_span_id: str | None

name: str

kind: _StoredValue

start_time: int

end_time: int | None

attributes: Dict[str, Any]

events: Sequence[_StoredEvent]

status: _StoredValue

status_message: str

duration_ms()

Return the completed duration in milliseconds.

Return type:

float

__init__(*args, **kwargs)

class praval.observability.storage.sqlite_store.SQLiteTraceStore(db_path, *, busy_timeout_ms=5000)

Bases: object

SQLite-based trace storage.

Stores spans in a local SQLite database with OpenTelemetry schema.

Parameters:

• db_path (str)

• busy_timeout_ms (int)

SCHEMA = '\n    CREATE TABLE IF NOT EXISTS spans (\n        span_id TEXT PRIMARY KEY,\n        trace_id TEXT NOT NULL,\n        parent_span_id TEXT,\n        name TEXT NOT NULL,\n        kind TEXT NOT NULL,\n        start_time INTEGER NOT NULL,\n        end_time INTEGER,\n        duration_ms REAL,\n        attributes TEXT,\n        events TEXT,\n        status TEXT,\n        status_message TEXT,\n        resource_attributes TEXT,\n        resource_schema_url TEXT,\n        scope_name TEXT,\n        scope_version TEXT,\n        scope_schema_url TEXT,\n        scope_attributes TEXT,\n        links TEXT,\n        trace_state TEXT,\n        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP\n    );\n\n    CREATE INDEX IF NOT EXISTS idx_spans_trace_id ON spans(trace_id);\n    CREATE INDEX IF NOT EXISTS idx_spans_parent ON spans(parent_span_id);\n    CREATE INDEX IF NOT EXISTS idx_spans_name ON spans(name);\n    CREATE INDEX IF NOT EXISTS idx_spans_start_time ON spans(start_time DESC);\n    CREATE INDEX IF NOT EXISTS idx_spans_status ON spans(status);\n    '

__init__(db_path, *, busy_timeout_ms=5000)

Initialize SQLite trace store.

Parameters:

• db_path (str) – Path to SQLite database file

• busy_timeout_ms (int)

store_span(span)

Store a completed span.

Parameters:

span (StorableSpan) – Span to store

Return type:

None

store_spans(spans)

Store multiple spans (batch operation).

Parameters:

spans (Sequence[StorableSpan]) – List of spans to store

Return type:

None

store_span_records(records)

Store normalized records produced by the official SDK exporter.

Return type:

None

Parameters:

records (Sequence[Dict[str, Any]])

get_trace(trace_id)

Get all spans for a trace.

Parameters:

trace_id (str) – Trace identifier

Return type:

List[Dict[str, Any]]

Returns:

List of span dictionaries

get_recent_traces(limit=10)

Get recent trace IDs.

Parameters:

limit (int) – Maximum number of trace IDs to return

Return type:

List[str]

Returns:

List of trace IDs (most recent first)

find_spans(agent_name=None, status=None, min_duration_ms=None, limit=100)

Query spans with filters.

Parameters:

• agent_name (Optional[str]) – Filter by agent name (matches span name)

• status (Optional[str]) – Filter by status (OK, ERROR, UNSET)

• min_duration_ms (Optional[float]) – Minimum duration in milliseconds

• limit (int) – Maximum number of spans to return

Return type:

List[Dict[str, Any]]

Returns:

List of span dictionaries

get_stats()

Get storage statistics.

Return type:

Dict[str, Any]

Returns:

Dictionary with storage stats

cleanup_old_traces(days=30)

Delete traces older than specified days.

Parameters:

days (int) – Number of days to retain

Return type:

int

Returns:

Number of spans deleted

cleanup_traces(*, max_age_days=None, keep_last_n=None, now_ns=None)

Delete complete traces outside age and count retention bounds.

Return type:

int

Parameters:

• max_age_days (int | None)

• keep_last_n (int | None)

• now_ns (int | None)

praval.observability.storage.sqlite_store.get_trace_store()

Get the global trace store instance.

Return type:

SQLiteTraceStore

Returns:

SQLiteTraceStore instance

praval.observability.storage.sqlite_store.set_trace_store(store)

Bind the store owned by the active explicit lifecycle.

Return type:

None

Parameters:

store (SQLiteTraceStore | None)

praval.observability.storage.sqlite_store.reset_trace_store()

Reset the global trace store to None.

This is primarily used for testing to ensure test isolation.

Return type:

None