Trailproof Class
The Trailproof class is the main entry point. It manages the hash chain, store, and optional signer.
Constructor
tp = Trailproof(
store="memory", # "memory" (default) or "jsonl"
path=None, # file path (required for jsonl store)
signing_key=None, # HMAC-SHA256 key (optional)
default_tenant_id=None, # applied to every event if not specified
)
emit()
Record a new event in the audit trail.
event = tp.emit(
event_type="myapp.user.login", # required
actor_id="user-42", # required
tenant_id="acme-corp", # required (or use default_tenant_id)
payload={"ip": "1.2.3.4"}, # required
trace_id="trace-abc", # optional
session_id="session-xyz", # optional
)
Returns: TrailEvent — the complete event with auto-generated fields.
Throws: ValidationError if required fields are missing or empty.
Behavior:
- Auto-generates
event_id (UUID v4) and timestamp (ISO-8601 UTC)
- Computes
hash using the hash chain engine
- Sets
prev_hash to the previous event’s hash (or genesis hash for the first event)
- If a signing key is configured, computes and sets
signature
- Appends the event to the store
query()
Search events with filters and pagination.
result = tp.query(
event_type="myapp.user.login", # optional
actor_id="user-42", # optional
tenant_id="acme-corp", # optional
trace_id="trace-abc", # optional
session_id="session-xyz", # optional
from_time="2025-01-01T00:00:00Z", # optional
to_time="2025-12-31T23:59:59Z", # optional
limit=100, # default 100
cursor=None, # for pagination
)
result.events # list[TrailEvent]
result.next_cursor # str | None
Returns: QueryResult { events, next_cursor }.
All filters are optional. No filters returns all events up to limit. Filters use exact match except from_time and to_time which are range filters.
verify()
Walk the hash chain and check every event’s hash.
result = tp.verify()
result.intact # bool -- True if no tampering
result.total # int -- number of events checked
result.broken # list[int] -- indices of broken events
Returns: VerifyResult { intact, total, broken }.
Empty chain returns { intact: true, total: 0, broken: [] }.
Verification does not throw on broken chains — it returns the result. Check result.intact to determine if the chain is valid.
get_trace() / getTrace()
Get all events for a specific trace ID, ordered by timestamp.
events = tp.get_trace("trace-abc")
# returns: list[TrailEvent]
Returns: List of TrailEvent objects matching the trace ID, ordered by timestamp.
flush()
Ensure all buffered events are persisted to disk.
No-op for the memory store. For the JSONL store, ensures all buffered writes are flushed to disk.