Skip to content

briefcase.bitemporal

Terminal window
pip install briefcase-ai[bitemporal]

Append-only store that tracks both valid time (when a fact is true) and transaction time (when it was recorded), so any past state can be reconstructed. An Iceberg-backed store is available via pip install briefcase-ai[bitemporal-iceberg].

BitemporalRecord, InMemoryBitemporalStore

from datetime import datetime, timezone
from briefcase.bitemporal import (
BitemporalRecord,
InMemoryBitemporalStore,
AsOfView,
append_correction,
)
store = InMemoryBitemporalStore()
now = datetime.now(timezone.utc)
record = BitemporalRecord.new(
key="config:max_retries",
valid_time=now,
value=3,
source="config-service",
)
store.append(record)
print(store.latest("config:max_retries").value, record.content_hash()[:12])
# Append-only correction (the original stays in history).
append_correction(store, record, 5, source="ops")
print(store.latest("config:max_retries").value)
print(len(store.history("config:max_retries")))
# Reconstruct the store as of a transaction time.
view = AsOfView(store, transaction_time=datetime.now(timezone.utc))
print(view.as_of("config:max_retries").value)
BitemporalRecord.new(key, valid_time, value, source, *, transaction_time=None,
decision=None, source_trust_level=None,
parent_record_id=None, metadata=None, record_id=None)
.content_hash() -> str
.record_id
InMemoryBitemporalStore()
.append(record)
.append_many(records)
.latest(key)
.history(key)
.as_of(key, *, transaction_time=None, valid_time=None)
.keys()
AsOfView(store, *, transaction_time=None, valid_time=None)
append_correction(store, original, corrected_value, *, source=None, ...)
batch_append(store, records, *, transaction_time=None)
stream_append(store, record)

Durable backends

from briefcase.bitemporal.backends import (
SqliteBitemporalBackend,
IcebergBitemporalBackend,
GlueIcebergBackend,
KdbBitemporalBackend,
)

Glue uses the bitemporal-glue extra. kdb+ uses the separately licensed kdb extra and is excluded from briefcase-ai[all].