Exporters
Exporters control where decision records go the moment they’re captured — to the console, a file, memory, or a sink of your own.
base install For: observability & integration
@capture records every call, but on its own it has nowhere to send the record. An exporter is about streaming records out as they happen (for inspection, tests, or forwarding). A storage backend is about durable persistence you query later. Many setups use both.
How exporting fits
-
Capture —
@capturerecords aclassify_ticketcall as a lightweight dict. -
Wire an exporter —
briefcase.observe()configures the global exporter in one line (or passexporter=to@capture). -
Land it — the exporter writes the record where you pointed it: stderr, a
.jsonlfile, an in-memory list, or your own sink.
observe, setup, and all stock exporters ship in the base package — no extra required.
Emit records in one line
briefcase.observe() configures the global exporter and returns it. After
calling it, every @capture decision is sent to that exporter.
import briefcase
mem = briefcase.observe("memory")
@briefcase.capture(decision_type="ticket-classification", async_capture=False)def classify_ticket(text: str) -> str: # call your model here return "billing"
classify_ticket("My invoice is wrong")print(mem.records[0])observe() shorthands
briefcase.observe(exporter="console", *, level=None) accepts either a
BaseExporter instance or a shorthand string, and returns the configured
exporter.
| Argument | Result |
|---|---|
"console" (default) | ConsoleExporter — writes JSON lines to stderr |
"memory" | MemoryExporter — collects records in .records |
a path ending in .jsonl | JSONLFileExporter — appends to that file |
a BaseExporter instance | used as-is |
level= (optional) also enables Briefcase logging at that level — the same as
calling enable_logging().
import briefcase
# Each call replaces the global exporter.briefcase.observe("console") # JSON lines to stderr (default)briefcase.observe("memory") # collect in memorybriefcase.observe("decisions.jsonl") # append to a filebriefcase.observe("console", level="INFO") # also turn on loggingobserve() calls setup(exporter=...) under the hood, so
briefcase.setup(exporter=ConsoleExporter()) is equivalent to
briefcase.observe("console").
Which stock exporter?
| Exporter | Sends records to… | Reach for it when |
|---|---|---|
ConsoleExporter | a stream (sys.stderr by default) | Developing or debugging and you want to watch decisions live |
JSONLFileExporter | a .jsonl file (one record per line) | You want a durable, append-only local log you can grep or post-process |
MemoryExporter | an in-memory list on .records | Tests and notebooks — capture decisions, then assert on them without I/O |
ConsoleExporter
Writes each record as one line of JSON to a stream. The quickest way to confirm
@capture is producing records.
import sys
from briefcase import setupfrom briefcase.exporters import ConsoleExporter
setup(exporter=ConsoleExporter(sys.stdout, pretty=True))ConsoleExporter(stream=None, *, pretty=False) — stream defaults to
sys.stderr; pretty=True indents the JSON.
JSONLFileExporter
Appends records to a file as JSON Lines (one object per line). Durable,
append-only, and thread-safe, so it is safe to share across the background
export threads @capture spawns. Parent directories are created on demand.
import briefcase
briefcase.observe("decisions.jsonl") # or JSONLFileExporter("decisions.jsonl")JSONLFileExporter(path) — path is a string or pathlib.Path.
MemoryExporter
Collects records in a list on .records. Ideal for tests and notebooks where
you want to read the captured decisions back.
import briefcase
mem = briefcase.observe("memory")
@briefcase.capture(async_capture=False)def classify_ticket(text: str) -> str: # call your model here return "billing"
classify_ticket("My invoice is wrong")assert mem.records[0]["function_name"] == "classify_ticket"mem.clear() # drop all collected recordsMemoryExporter() exposes .records (a list) and .clear().
Custom exporters: ship to an external sink
Subclass BaseExporter to forward decisions anywhere your stack already
collects events — a log aggregator, a message queue, an analytics pipeline. You
implement three async methods; register the instance with observe() (it returns
it unchanged) or with setup(exporter=...).
from typing import Any
import briefcasefrom briefcase.exporters import BaseExporter
class WebhookExporter(BaseExporter): async def export(self, decision: Any) -> bool: # ship `decision` (a dict) to your external sink here # e.g. post to a collector, enqueue, or forward to a log pipeline return True
async def flush(self) -> None: ...
async def close(self) -> None: ...
exporter = briefcase.observe(WebhookExporter())
@briefcase.capture(decision_type="classification", async_capture=False)def classify_ticket(text: str) -> str: # call your model here return "account_access"
classify_ticket("Reset my password")export(decision)ships a single record; returnTrueon success.flush()flushes any buffered records.close()releases resources.
Record shape
Each record @capture hands to an exporter is a dict:
decision_id— a UUID stringdecision_type— the value you passed, or the function qualified namefunction_nameinputs/outputs— truncated reprs of the arguments and return valuestarted_at/ended_at— ISO 8601 timestampsexecution_time_mscontext_version— present only when you pass iterror— present only when the call raised
Key symbols
briefcase.observe(exporter="console", *, level=None)— configure and return the global exporter.briefcase.exporters.ConsoleExporter— JSON lines to a stream.briefcase.exporters.JSONLFileExporter— append JSON Lines to a file.briefcase.exporters.MemoryExporter— collect records in.records.briefcase.exporters.BaseExporter— base class for custom exporters.