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), and so is API Gateway's 403 Forbidden for a brand-new key, which works within a minute or two (new_key_wait_s, 120 s).

• 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=, order=, after=, as_of=), query_all (a generator), count(db, where, as_of=) (a predicate's rows counted from the index, none read: {"count", "t"}), pull(db, pks, as_of=) (as_of=t: the database as it stood after t), request_index(db, att, required=) (att an attribute, or a list of 2 to 8 for a multi-attribute index: queried by tuples, ["=", ["a", "b"], [x, y]]; docs/query.md), await_index, drop_index, history(db, start=1, limit=), history_all, fork(db, t=, name=) (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), set_history_retention(db, days) (history kept 30 to 36500 days, or None for forever).
• Keys (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.
• row_id(n): a row named by its row id.

Keys and row ids. A primary key is a UUID (squuid()), a string or an int within a long's range. Every row also carries "stratus/id", a row id the database assigns when the row is made. transact answers {"t", "pks", "ids"}. Where a key goes - pull, delete=, a row's key attribute - row_id(n) names a row by its id; a bare int is always a primary key. Queries return rows in row id order (the order they were made), order="desc" newest first; next is a row id.

Requests go to <url>/v1/dbs/<id>/... with Authorization: Bearer <api_key>.

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 hosted product run locally.