Memory

Constructor

Memory(
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    path: str | os.PathLike[str] | None = None,
    validate_user_id: Callable[[str], None] | None = None,
    timeout: float | httpx.Timeout = 10.0,
    max_retries: int = 2,
    suppress_async_warning: bool | None = None,
    transport: httpx.BaseTransport | None = None,
)

Every argument is keyword-only.

• api_key — reserved for v0.2 remote backend; defaults to ATLASO_API_KEY env.
• base_url — reserved for v0.2; defaults to ATLASO_BASE_URL env or https://api.atlaso.dev.
• path — SQLite location. If None, resolved via the 5-step path walker (see Configuration).
• validate_user_id — optional callback called inside for_user() and every lower-level call; raise to block.
• timeout — httpx timeout for the future remote backend.
• max_retries — transport-level retries (network + 5xx + 429).
• suppress_async_warning — silence SyncInAsyncWarning. Three levers: this arg > ATLASO_ASYNC_WARNINGS=0 > warnings.filterwarnings(...).
• transport — inject an httpx.BaseTransport for tests. See the warning below.

for_user(user_id)

Returns a UserHandle bound to user_id. The handle pre-fills user_id on every method call so callers never accidentally pass the wrong identity.

user = m.for_user("alice")
user.add("…")
user.recall("…")

add(text, *, user_id, ...)

Write a deposit. Returns AddResult.

add(
    text: str,
    *,
    user_id: str,
    polarity: Polarity = "open",
    evidence_grade: EvidenceGrade = "anecdotal",
    scope: Scope | None = None,
    tags: list[str] | None = None,
    contradicts: list[str] | None = None,
    author: str | None = None,
) -> AddResult

Raises InputValidationError, DepositRejectedError, MissingContradictsError.

recall(query, *, user_id, ...)

Dispersion-aware search. Returns a SearchResults container whose has_disagreement and is_confident flags are computed across all bags before slicing.

recall(
    query: str,
    *,
    user_id: str,
    limit: int = 10,         # 1..1000
    scope: Scope | None = None,
) -> SearchResults

get(deposit_id, *, user_id)

Fetch one deposit by id. Returns None on miss — deliberately distinct from NotFoundError, for dict.get muscle memory.

peek(user_id, *, limit=10)

REPL/Jupyter-friendly snapshot — recent deposits + FMI + a recent disagreement flag. Returns PeekView.

list_recent(*, user_id, limit=20, offset=0)

Newest-first paginated deposit list.

contradict(new_text, contradicts, *, reason, user_id)

Atomic supersede. Writes a new deposit and marks one or more existing deposits as superseded. reason is required; empty contradicts list raises InputValidationError.

retract(deposit_id, *, reason, user_id, hard_delete=False, force=False)

Soft retract by default (the deposit becomes invisible to recall but survives in the file). hard_delete=True is irreversible — it removes the row and clears the FTS index. force is accepted for forward-compatibility (reserved for v0.2 server-side protected-deposit flags) and currently has no effect on the local backend.

health(*, user_id, window_days=30)

Returns a Diagnostics with the Field Maturity Index and its four pillars.

add_many(items, *, user_id, on_gate_reject="skip")

Bulk import. Each AddItem requires an idempotency_key. Skip-and-report on gate rejection. See Idempotency.

close() / context manager

Memory is a context manager: with Memory() as m: ... closes the backend on exit. A caller-supplied transport is not closed for you.

update() is a typo-blocker

m.update("anything")
# AttributeError: deposits are immutable.
# Use m.contradict(new_text, contradicts=[old.id], reason="…") instead.