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.

Classes

SQLiteTraceStore(db_path)

SQLite-based trace storage.

class praval.observability.storage.sqlite_store.SQLiteTraceStore(db_path)

Bases: object

SQLite-based trace storage.

Stores spans in a local SQLite database with OpenTelemetry schema.

Parameters:

db_path (str)

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

Initialize SQLite trace store.

Parameters:

db_path (str) – Path to SQLite database file

store_span(span)

Store a completed span.

Parameters:

span (Span) – Span to store

Return type:

None

store_spans(spans)

Store multiple spans (batch operation).

Parameters:

spans (List[Span]) – List of spans to store

Return type:

None

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

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.reset_trace_store()

Reset the global trace store to None.

This is primarily used for testing to ensure test isolation.

Return type:

None