stratus-db (Python)

The stratus service from Python 3.9+: typed JSON over urllib (docs/json.md), no dependencies.

The client is one file, stratus_db.py. Install it from PyPI (pip install stratus-db; a wheel only, published from this repository's release workflow with PyPI's attestations), or copy the file into your project. To pin it, use pip install --require-hashes with the wheel's hash, or compare the file's sha256 with the one published beside it.

import datetime as dt, os
from decimal import Decimal
from stratus_db import connect, kw, squuid

s = connect("https://api.stratus-db.com", api_key=os.environ["STRATUS_KEY"])
db = s.create_db(primary_key="invoice/id", name="invoices")["id"]

s.transact(db, tx=[{
    "invoice/id": squuid(),                        # every row carries its key
    "invoice/total": Decimal("1234.50"),           # exact, scale kept
    "invoice/due": dt.date(2026, 11, 1),
    "invoice/sent": dt.datetime.now(dt.timezone.utc),
    "invoice/status": kw("status/sent"),           # a keyword, not the string "status/sent"
}], tx_id="import-42")                             # a retry with the same id applies once

s.request_index(db, "invoice/status")
s.await_index(db, "invoice/status")
page = s.query(db, where=["=", "invoice/status", kw("status/sent")], limit=100)
for row in s.query_all(db, where=["range", "invoice/due", dt.date(2026, 11, 1), None]):
    ...

Values

A date or instant outside years 1-9999 reads as a Tagged, which writes back unchanged. A UUID, instant or keyword passed as a plain string is stored as a string and matches nothing of the intended type.

API

Failures raise StratusError (status, error - e.g. "invalid-query", "conflict" -, message). 429s are retried with backoff (retries).

• Catalog: list_dbs(deleted=False), db_by_name, db_record, create_db(primary_key, name=None), rename_db, delete_db.
• A database: status, transact(db, tx=, retract=, delete=, tx_id=, match=), query(db, where=, select=, limit=, after=), query_all (a generator), pull(db, pks), request_index, await_index, drop_index, history(db, start=1, limit=), history_all, fork(db, t=, name=) (hosted: a new database, this one as of t), excise(db, rows=, attributes=) (erasure: rows and (pk, att) out of the state, indexes, history and forks).
• Keys (hosted, with a key reaching every database): list_keys(), create_key(name=, scope=), revoke_key(key_id).
• Codec: encode, decode, to_json_value, from_json_value.

Addressing: "hosted" (default: <url>/v1/dbs/<id>/..., Authorization: Bearer <api_key>) or "function-urls" (the internal service's Function URLs, resolved through its catalog; caller for a local service).

Tests

python3 -m unittest tests.test_ejson (the codec, over velojity's own output in clients/fixtures/velojity-ejson.json), and scripts/test-python-client from the repo root (PYTHON= picks the interpreter): the codec, then the client against the local service.