Overview
The SQLiteTraceHandler class provides the core SQLite-backed implementation for persisting guard execution traces, including pre/post-validation text, timing data, and exception messages.
Description
This module implements the low-level trace storage layer using SQLite. The SQLiteTraceHandler class extends TracerMixin and manages a guard_logs table with columns for guard name, start/end times, pre/post-validation text, and exception messages.
Key design considerations for multi-threaded safety:
- Write-Ahead Logging (WAL) mode is enabled for concurrent read/write support.
- Synchronous OFF is set for write performance (accepting the risk of losing log data on crash).
check_same_thread=False allows sharing connections across threads.
- A log retention system automatically truncates old entries beyond
LOG_RETENTION_LIMIT (100,000 entries) with cleanup no more frequent than every 10 seconds.
The module also registers custom SQLite adapters and converters for datetime.datetime objects, mapping them to Unix timestamps.
Usage
This class should not be used directly in most cases. Instead, use TraceHandler, which wraps SQLiteTraceHandler with a thread-safe singleton pattern. Direct use is appropriate only when implementing custom trace handling or testing.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/call_tracing/sqlite_trace_handler.py
- Lines: 1-235
Signature
LOG_RETENTION_LIMIT = 100000
TIME_BETWEEN_CLEANUPS = 10.0 # Seconds
def adapt_datetime(val):
"""Adapt datetime.datetime to Unix timestamp."""
def convert_timestamp(val):
"""Convert Unix epoch timestamp to datetime.datetime object."""
class SQLiteTraceHandler(TracerMixin):
def __init__(self, log_path: os.PathLike, read_mode: bool):
def _get_write_connection(cls, log_path: os.PathLike) -> sqlite3.Connection:
def _get_read_connection(cls, log_path: os.PathLike) -> sqlite3.Connection:
def _truncate(self, force: bool = False, keep_n: int = LOG_RETENTION_LIMIT):
def log(self, guard_name, start_time, end_time, prevalidate_text, postvalidate_text, exception_text):
def log_entry(self, guard_log_entry: GuardTraceEntry):
def log_validator(self, vlog: ValidatorLogs):
def clear_logs(self):
def tail_logs(self, start_offset_idx: int = 0, follow: bool = False) -> Iterator[GuardTraceEntry]:
Import
from guardrails.call_tracing.sqlite_trace_handler import SQLiteTraceHandler
I/O Contract
__init__ Parameters
| Parameter |
Type |
Description
|
log_path |
os.PathLike |
Path to the SQLite database file for storing trace logs.
|
read_mode |
bool |
If True, opens the database in read-only mode. If False, opens for writing with WAL mode enabled.
|
log Parameters
| Parameter |
Type |
Description
|
guard_name |
str |
Name of the guard being traced.
|
start_time |
float |
Epoch timestamp when the guard execution started.
|
end_time |
float |
Epoch timestamp when the guard execution ended.
|
prevalidate_text |
str |
The raw LLM output text before validation.
|
postvalidate_text |
str |
The text after validation processing.
|
exception_text |
str |
Any exception message encountered during validation.
|
tail_logs
| Parameter |
Type |
Default |
Description
|
start_offset_idx |
int |
0 |
Start printing entries after this ID. Negative values index from the end (e.g., -10 returns the last 10 entries).
|
follow |
bool |
False |
If True, continuously re-checks the database for new entries after initial results are exhausted.
|
| Return Type |
Description
|
Iterator[GuardTraceEntry] |
A generator yielding GuardTraceEntry objects for each log row.
|
Database Schema
| Column |
Type |
Description
|
id |
INTEGER PRIMARY KEY AUTOINCREMENT |
Auto-incrementing primary key.
|
guard_name |
TEXT |
Name of the guard.
|
start_time |
REAL |
Start time as epoch float.
|
end_time |
REAL |
End time as epoch float.
|
prevalidate_text |
TEXT |
Text before validation.
|
postvalidate_text |
TEXT |
Text after validation.
|
exception_message |
TEXT |
Exception message, if any.
|
Usage Examples
from guardrails.call_tracing.sqlite_trace_handler import SQLiteTraceHandler
# Writing logs (typically done internally by the framework)
writer = SQLiteTraceHandler("/tmp/guardrails_calls.db", read_mode=False)
writer.log(
guard_name="my_guard",
start_time=1700000000.0,
end_time=1700000001.5,
prevalidate_text="Raw LLM output here",
postvalidate_text="Validated output here",
exception_text="",
)
# Reading logs
reader = SQLiteTraceHandler("/tmp/guardrails_calls.db", read_mode=True)
for entry in reader.tail_logs(start_offset_idx=0, follow=False):
print(entry.guard_name, entry.prevalidate_text)
# Tailing the last 5 entries
for entry in reader.tail_logs(start_offset_idx=-5, follow=False):
print(entry)
# Clearing all logs
writer.clear_logs()
Related Pages