Writing a memory backend

Built and verified

MemoryBackend, the conformance suite and the registration routes below exist and are tested. The testing extra installs pytest for you (pip install "coniti[testing]").

A new backend, such as a database or a remote service, subclasses MemoryBackend and implements eight underscore methods. The public methods check the request and result contracts for you, so you only write the storage.

# my_pkg/backend.py
from coniti.memory import (
    ClearRequest, ClearResult, CountRequest, CountResult, DeleteRequest,
    DeleteResult, GetRequest, GetResult, MemoryBackend, MetaGetRequest,
    MetaResult, MetaSetRequest, PutRequest, PutResult, ScanRequest, ScanResult,
)


class MyBackend(MemoryBackend):
    def _put(self, request: PutRequest) -> PutResult: ...
    def _get(self, request: GetRequest) -> GetResult: ...
    def _delete(self, request: DeleteRequest) -> DeleteResult: ...
    def _scan(self, request: ScanRequest) -> ScanResult: ...
    def _count(self, request: CountRequest) -> CountResult: ...
    def _get_meta(self, request: MetaGetRequest) -> MetaResult: ...
    def _set_meta(self, request: MetaSetRequest) -> MetaResult: ...
    def _clear(self, request: ClearRequest) -> ClearResult: ...

The base class provides the public put, get, delete, scan, count, get_meta, set_meta, clear and iter_all, plus async versions that run the sync call in a worker thread. A remote backend can override the async methods natively, and can extend (not replace) the request and result models by setting the matching class attributes, for example scan_result.

Give the class a params_model (a subclass of Params) if it takes settings.

The contract to honour

See memory for the full list. In short:

• put is all-or-nothing; an existing or repeated id raises DuplicateRecord.
• Unknown ids go in missing; get and delete never raise for them.
• scan is ordered by (timestamp, write order), half-open, paginated, with opaque cursors that resume exactly, and InvalidCursor for any cursor you did not issue for that request (including the empty string).
• Namespaces never see each other.
• Records are copied in and out.
• dump and load (the memory file) are inherited from MemoryBackend; the suite checks that your backend round-trips through them, as a stream and as plain and gzip paths, and that a dump covers only the requested namespace. See backup-restore.

Run the conformance suite

The suite lives in coniti.testing.memory and needs pytest.

With pytest. Subclass MemoryBackendContract and implement make_backend(). pytest then runs every check against a fresh backend:

# tests/test_my_backend.py
from coniti.testing.memory import MemoryBackendContract
from my_pkg.backend import MyBackend


class TestMyBackend(MemoryBackendContract):
    def make_backend(self):
        return MyBackend()

Without the pytest runner. run_contract(make_backend) runs the same checks and returns a dict mapping each check name to its failure, or None if it passed:

from coniti.testing.memory import run_contract
from my_pkg.backend import MyBackend

results = run_contract(MyBackend)
failures = {name: exc for name, exc in results.items() if exc is not None}
assert not failures, failures

The checks cover duplicates, missing ids, ordering, half-open ranges, pagination at every page size, cursor resumption after changes, bad cursors, namespace isolation, metadata, clear, copying, wrong request types, async parity, and a seeded random sequence of operations compared with a simple list model. The suite is itself tested: it catches six deliberately broken backends (partial batch writes, reversed pages, records that are not copied, raising on missing ids, leaking across namespaces, a bad end-of-scan cursor). A record(...) helper builds records for your own extra tests.

Register it

Bind by class path, with no registration:

[memory]
kind = "my_pkg.backend:MyBackend"

or give it a short name that others can use. Register it with @register and expose the module through the coniti.plugins entry-point group:

# my_pkg/coniti_plugins.py
from coniti import Params, register
from my_pkg.backend import MyBackend

register("memory", "my_store", Params)(MyBackend)  # or your own Params subclass

# your package's pyproject.toml
[project.entry-points."coniti.plugins"]
my_plugins = "my_pkg.coniti_plugins"

After installing your package, kind = "my_store" works. See plugins for how discovery behaves.