API reference

The public surface is intentionally tiny. Most use cases need only SecantusDBServer — talk to it through pymongo and you get the full MongoDB-shaped API.

Server

class secantus.SecantusDBServer(host: str = '127.0.0.1', port: int = 0, storage_path: str = './secantus-data', *, replica_set_name: str | None = 'secantus', require_auth: bool = False)

Bases: object

Storage

Storage is the WT-backed document/index store. It’s normally accessed via pymongo commands routed through the server, but the public methods are stable enough to call directly when a test wants explicit control — notably prune_ttl, explain_plan, collection_data_size, and index_sizes.

class secantus.storage.Storage(path: str = ':memory:', *, oplog_retention_seconds: float = 3600.0, oplog_max_entries: int = 100000, time_func: Callable[[], float] | None = None, enable_oplog: bool = True)

explain_plan(db: str, coll: str, filter: dict[str, Any] | None = None, *, sort: Mapping[str, Any] | None = None, hint: str | Mapping[str, Any] | None = None) → dict[str, Any]

Plan summary for what find_matching would do with these args.

No execution; mirrors the same routing decisions. Returns {"kind": "COLLSCAN"} or {"kind": "IXSCAN", "index_name", "key_pattern", "direction"}. direction is "forward" unless a sort spec inverts it relative to the chosen index.

collection_data_size(db: str, coll: str) → int

Sum of bson-encoded doc bytes for coll.

Used by collStats / dbStats for size / dataSize. Best-effort estimate — doesn’t include WT block overhead.

index_sizes(db: str, coll: str) → dict[str, int]

Map of index name → sum of packed entry-key bytes.

_id_ is reported separately as len(id_key) summed across the doc table, so callers can include it alongside secondary indexes for an accurate totalIndexSize.

prune_ttl(db: str, coll: str, *, now: datetime | None = None) → int

Delete docs whose indexed Date field is older than now - TTL.

For every index on coll with an expireAfterSeconds option, walks the collection and deletes docs whose indexed field resolves to a datetime older than now - expireAfterSeconds. Docs without the field, with non-date values, or with values inside the TTL window are left in place. Real MongoDB runs this on a 60s background sweeper; SecantusDB invokes it explicitly so tests can drive expiry with an injected now. Returns the number of docs pruned.

Cursors

class secantus.cursors.CursorRegistry(idle_ttl_seconds: float = 600.0, time_func: Callable[[], float] | None = None, tailable_idle_ttl_seconds: float = 1800.0)

class secantus.cursors.CursorNotFound(cursor_id: int)

CLI

secantus.cli.build_parser() → ArgumentParser

secantus.cli.main(argv: list[str] | None = None) → int

Module-level

secantus.__version__ = '0.3.0a26'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.