briefcase.storage
pip install briefcase-ai[storage]Two backends ship in the open-source package: SqliteBackend and
BufferedBackend. The native runtime must be initialized first.
SqliteBackend
import briefcasefrom briefcase import DecisionSnapshot, Input, Output, Snapshot, SnapshotQueryfrom briefcase.storage import SqliteBackend
briefcase.init()
backend = SqliteBackend.in_memory() # or SqliteBackend("decisions.db")
decision = DecisionSnapshot("classify_ticket")decision.add_input(Input("text", "reset my password", "string"))decision.add_output(Output("category", "account_access", "string"))
decision_id = backend.save_decision(decision)loaded = backend.load_decision(decision_id)
session = Snapshot("session")session.add_decision(decision)snapshot_id = backend.save(session)backend.load(snapshot_id)
backend.query(SnapshotQuery().with_function_name("classify_ticket"))backend.health_check()SqliteBackend(path)SqliteBackend.in_memory() .save(snapshot) -> snapshot_id .load(snapshot_id) .save_decision(decision) -> decision_id .load_decision(decision_id) .query(query) .delete(id) .health_check()BufferedBackend
Batches decision writes: decisions are held until buffer_size accumulate, then
committed to the wrapped backend in one transaction.
from briefcase.storage import BufferedBackend
with BufferedBackend(backend, buffer_size=100) as buffered: decision_id = buffered.save_decision(decision) buffered.pending() # decisions not yet written buffered.load_decision(decision_id) # readable while still buffered# leaving the block flushesBufferedBackend(backend, buffer_size) .save_decision(decision) -> str # id is valid immediately .flush() -> int # decisions written by this call .pending() -> int .load(snapshot_id) / .load_decision(decision_id) / .query(q) / .delete(id) .health_check()Pending decisions live in memory: a process that exits without flushing loses
them, and query does not see them. buffer_size=1 writes through. Dropping a
backend with unflushed decisions prints a warning to stderr.