oj-persistence 0.1.0

Thread-safe persistence management with pluggable storage backends

oj-persistence

Thread-safe, async-first persistence with a single gatekeeper (Manager) in front of pluggable backends. Consumers register tables by name + spec; they never see a store, connection, handle, or file descriptor.

Installation

pip install oj-persistence

SQLite and InMemory need no extra dependencies. Install drivers only for the backends you actually use:

pip install "oj-persistence[dev]"    # pytest + aiosqlite + fakeredis

Quick start

from oj_persistence import Manager, Sqlite, InMemory

pm = Manager()

pm.register('users', Sqlite(path='data/app.db'))
pm.register('orders', Sqlite(path='data/app.db'))   # same path → shared connection
pm.register('cache', InMemory())

pm.create('users', 'u1', {'name': 'Alice', 'role': 'admin'})
pm.read('users', 'u1')                               # {'name': 'Alice', 'role': 'admin'}
pm.update('users', 'u1', {'name': 'Alice', 'role': 'owner'})
pm.upsert('users', 'u2', {'name': 'Bob'})
pm.delete('users', 'u2')

owners = pm.list('users', predicate=lambda v: v['role'] == 'owner')

pm.close()

Every sync method has an a-prefixed async counterpart with identical semantics:

await pm.aregister('users', Sqlite(path='data/app.db'))
await pm.acreate('users', 'u1', {'name': 'Alice'})
await pm.aread('users', 'u1')
async for v in pm.aiter('users'):
    ...
await pm.aclose()

---

Manager-as-gatekeeper

The Manager is the only thing that touches a backing resource. Consumers hand it a table name and a spec (Sqlite(path=...), InMemory(), ...), then interact only through the Manager's public API. No store object, no connection, no handle, no ref.

caller → Manager.create/read/update/upsert/delete/list/iter/join
                          ↓
                 table-name registry
                          ↓
         Backend  (SqliteBackend, InMemoryBackend, ...)

If you need an operation the Manager doesn't expose, the fix is to add it to the Manager — not to reach past it.

File-safety across Manager instances

File-safety is enforced at the file level, not the Manager-object level. A process-scoped _BackendRegistry guarantees each unique connection target (sqlite path, redis url, etc.) is backed by exactly one Backend. You can freely construct multiple Manager() instances — for test isolation, per-tenant namespacing, whatever. If two of them register Sqlite(path='app.db'), they share one underlying SqliteBackend with its single writer connection, reader pool, and WAL configuration. Two Managers cannot race on a file, ever.

Backends are reference-counted across Managers. When the last drop() / close() releases the final reference, the backend is closed and its resources released.

Register → register, drop, truncate, close

pm.register(table, spec, *, requires=..., replace=False)
pm.drop(table)          # delete data + unregister + release backend if last user
pm.truncate(table)      # clear data but keep registration
pm.close()              # release everything this Manager owns
pm.tables()             # ['users', 'orders', 'cache']
pm.exists('users')      # bool
pm.capabilities('users')  # frozenset[Capability]

register() raises TableAlreadyRegistered if the name is in use; pass replace=True to drop the previous registration first.

---

CRUDL semantics

Method	Behaviour
create	Raises KeyError if the key already exists
read	Returns None if the key is missing
update	Raises KeyError if the key does not exist
upsert	Create or overwrite
delete	No-op if the key is missing
list	Returns all values, optionally filtered by predicate
iter	Generator form of list; prefer for large tables

---

Capabilities

Every backend declares a frozenset[Capability] at the class level. The Manager checks them at register() time — no more "works on SQLite, throws NotImplementedError in-memory" surprises at the eventual call site.

class Capability(StrEnum):
    PAGINATION    = 'pagination'     # list_page(offset, limit)
    FIELD_INDEX   = 'field_index'    # add_index / list_by_field
    NATIVE_JOIN   = 'native_join'    # SQL JOIN pushdown for same-file tables
    NATIVE_UPSERT = 'native_upsert'  # in-place upsert without file rewrite

Backend	Capabilities
Sqlite	PAGINATION, FIELD_INDEX, NATIVE_JOIN, NATIVE_UPSERT
InMemory	PAGINATION, FIELD_INDEX, NATIVE_UPSERT

Declare what you need up-front and the Manager validates before you run:

from oj_persistence import Capability, UnsupportedOperation

# Fails at register() — InMemory has no NATIVE_JOIN
pm.register(
    'users',
    InMemory(),
    requires=frozenset({Capability.NATIVE_JOIN}),
)  # raises UnsupportedOperation

Calling a capability-gated method on a backend that doesn't declare it raises UnsupportedOperation at call time as a secondary safety net.

---

Backends

Sqlite(path, pool_size=4)

Local SQLite file with WAL mode, one writer connection, and a reader pool. path=':memory:' (the default) gives an in-memory SQLite database.

pm.register('users', Sqlite(path='data/app.db'))
pm.register('users', Sqlite(path='data/app.db', pool_size=8))

Multiple tables registered against the same path share one backend — reads across tables run concurrently, and joins push down into SQL (see Joins).

InMemory()

Dict-backed, single-lock. Fastest; not persistent. Each InMemory() spec gets its own backend — two InMemory() calls never share state. Useful for tests and ephemeral caches.

Other specs

Ndjson, Redis, SqlAlchemy, and TinyDb spec classes are exported and are targets for future backend work. Registering one today raises NotImplementedError — they ship in a later pass.

---

Pagination, field indexes, list_by_field

pm.register('users', Sqlite(path='app.db'))

for i in range(100):
    pm.create('users', f'u{i:03d}', {'i': i, 'role': 'admin' if i % 10 == 0 else 'viewer'})

# Pagination (offset/limit) — ordered by key
page = pm.list_page('users', offset=20, limit=10)

# Create a JSON-field index, then filter via pushed-down SQL
pm.add_index('users', '$.role')
admins = pm.list_by_field('users', '$.role', 'admin')

InMemory supports list_page and list_by_field too — the index call is a no-op there (dict scans are O(n) regardless).

---

Joins

pm.join(left, right, on, how='inner', where=None) pairs values from two tables. When both tables share the same backend instance and that backend declares Capability.NATIVE_JOIN, the join pushes down into the backend (SQL INNER JOIN / LEFT JOIN for SQLite). Otherwise the Manager materializes both sides and performs a nested-loop join in Python.

pm.register('users',  Sqlite(path='app.db'))
pm.register('orders', Sqlite(path='app.db'))   # same file → shared backend

pm.create('users',  'u1', {'id': 'u1', 'name': 'Alice'})
pm.create('orders', 'o1', {'user_id': 'u1', 'total': 150})

pairs = pm.join(
    'users', 'orders',
    on=lambda u, o: u['id'] == o['user_id'],
    how='left',
    where=lambda u, o: o['total'] > 100,
)
# -> [({'id': 'u1', 'name': 'Alice'}, {'user_id': 'u1', 'total': 150})]

Supported how values: inner, left, right, outer. RIGHT and OUTER joins always run the Python path (SQLite has no native syntax for them). The where filter is always applied in Python after the join.

For large tables, keep the tables on the same SQLite file so NATIVE_JOIN kicks in. Python-path joins are O(m × n).

---

Concurrency guarantees

SQLite files

• WAL mode (PRAGMA journal_mode=WAL) set at first open.
• One writer connection per file, serialized by an internal lock.
• Pool of reader connections (pool_size, default 4) — concurrent readers run truly in parallel.
• All I/O runs on a thread-pool executor, so async methods never block the event loop.

Resulting guarantees:

1. Concurrent aread / alist / aiter calls on the same or different tables in the same file do not block each other.
2. An in-flight write does not block any read. Readers see the last committed snapshot (WAL semantics).
3. Concurrent writes serialize (one at a time). Writers never wait for readers.
4. truncate / drop acquire the writer lock, so they cannot interleave with an in-flight write.

InMemory

One threading lock; reads and writes serialize. Cheap enough that contention isn't worth splitting.

What the Manager does not guarantee

• Cross-file transactions. A write to users (Sqlite) and a write to cache (InMemory) are not atomic together.
• Snapshot isolation across multiple async reads. Two await pm.aread(...) calls can see different snapshots if a writer lands between them. Callers needing a stable view should explicitly snapshot via await pm.alist(...).

---

Sync / async duality

The Manager exposes both flavors of every data method. Pick whichever matches your calling context:

Sync	Async
register	aregister
create	acreate
read	aread
update	aupdate
upsert	aupsert
delete	adelete
list	alist
iter	aiter
list_page	alist_page
list_by_field	alist_by_field
add_index	aadd_index
join	ajoin
truncate	atruncate
drop	adrop
close	aclose

The sync entry points run the async implementation on a Manager-owned background event loop — so they must not be called from inside a running event loop. Use the a-prefixed method there instead (the sync method raises RuntimeError to keep you honest).

---

Migration from v1

The v1 surface (PersistenceManager, AsyncPersistenceManager, *Store classes, configure() / create_group() / add_table(), StoreRef / GroupRef, store_context(), upsert-gating, Relation / JoinCondition) is gone as of v0.1.0. Rough mapping:

v1	v2
PersistenceManager() / AsyncPersistenceManager()	Manager() (unified sync + async)
pm.configure('sqlite', store_id='users', path='app.db')	pm.register('users', Sqlite(path='app.db'))
pm.register('users', SqliteStore('app.db'))	pm.register('users', Sqlite(path='app.db'))
pm.create_group('sqlite', ...) + pm.add_table(...)	Two pm.register(..., Sqlite(path=same)) calls share the backend
pm.read/write/list/...	Same names; a-prefixed variants for async
pm.unregister(store_id)	pm.drop(table)
Relation(..., JoinCondition(...))	pm.join(left, right, on=callable, how=...)
store_context() context manager	Removed — writes just work on the registered table
allow_inefficient=True upsert override	Removed (only native-upsert backends ship in v2)

Legacy v1 store classes (NdjsonFileStore, CsvFileStore, VersionedStore, etc.) still live under oj_persistence.store for direct programmatic use, but they are not wired into the v2 Manager. Prefer Sqlite / InMemory through the Manager; use the legacy stores only if you have an existing script that constructs them directly.

---

Utilities

oj_persistence.utils.streaming — stream bytes to a file atomically

import httpx
from oj_persistence.utils.streaming import stream_to_file, stream_to_file_sync

async with httpx.AsyncClient() as client:
    response = await client.get('https://example.com/data.ndjson.gz', stream=True)
    path = await stream_to_file(response.aiter_bytes(), '/tmp/data.ndjson.gz')

# Or sync, from an iterable or a .read()-capable object
import requests
r = requests.get('https://example.com/data.ndjson', stream=True)
path = stream_to_file_sync(r.iter_content(chunk_size=65536), '/tmp/data.ndjson')

Writes go to a UUID-named temp file in the same directory and are renamed atomically on success. On failure the temp file is deleted unless debug=True (or the OJ_PERSISTENCE_DEBUG=1 env var is set).

oj_persistence.utils.compression — codec resolution for file-backed stores

Legacy file stores under oj_persistence.store accept a compression parameter ('gzip', 'bz2', 'lzma', 'auto'). Auto-detection keys off the file extension (.gz, .bz2, .xz, .lzma).

---

Requirements

• Python >= 3.11
• sqlalchemy>=2.0, redis>=5.0, tinydb, ijson — installed by default; needed only by the legacy store classes or by future backends.

For testing without a live Redis server: pip install "fakeredis[lua]".

Version / stability

Current: v0.1.0. This is the first release of the v2 API. Expect some churn as the remaining backends (Ndjson, Redis, SqlAlchemy, TinyDb) are ported to the new Backend interface. The Manager surface itself is considered stable.