store: Where sync state lives#

A store keeps every scope, its records, and the change feed. Use SQLiteStore to keep state between runs, and MemoryStore in tests. In both, a read in one thread runs while another thread writes. To save the objects themselves somewhere else, such as with pickle or in an object database, subclass MemoryStore as its documentation shows.

The storage interface every store implements.

A store only keeps data. Every sync rule lives in Replica, so each store implements the same small set of typed operations and nothing more.

class calendaring_sync.store.base.StateStore[source]#

Bases: ABC

Where the core keeps each scope, its records, and the change feed.

Every operation runs inside transaction(): one that only reads runs inside any transaction, and one that writes needs a transaction opened with write=True. A store is safe to share between threads: one write transaction runs at a time, and in a store that sets _reads_beside_writes, transactions that only read run in other threads meanwhile instead of waiting for it.

Besides the errors each method lists, every operation, transaction() and close() can raise:

  • TypeError, for an argument of the wrong type, before anything is read or written; a record or scope to write is checked field by field

  • StoreError, if the store can’t read or write its data

  • UnexpectedStoreError, for any other failure inside the store, which is a bug

No exception of the storage underneath a store reaches the caller.

The public methods check that they run inside a transaction of the right kind, then call a method of the same name with a leading underscore. A store implements those, along with _begin(), _commit() and _rollback(), and needs no checks of its own. It turns the errors of the storage it uses into store errors by overriding _store_error(). Its __init__ calls super().__init__() first. A hook runs in the thread of the transaction it’s for, and _writing() says whether that transaction writes.

append_entry(scope_id: str, kind: FeedEntryKind, local_key: int | None, object_id: str | None, at: datetime) → FeedEntry[source]#

Write the next change feed entry and return it.

Its number is one more than the last number written. Runs inside a write transaction.

Raises:
close() → None[source]#

Close the store. Every later transaction raises StoreError.

Transactions open in other threads end first: closing waits for them. Closing a closed store does nothing.

Raises:

  • TransactionError – If this thread has a transaction open.

  • StoreError – If the store couldn’t release what it holds open. It counts as closed anyway.

delete_record(local_key: int) → None[source]#

Delete the record with this local key.

Runs inside a write transaction.

Raises:
delete_scope(scope_id: str) → None[source]#

Delete the scope and every record in it.

Runs inside a write transaction.

Raises:
entries(after: int, limit: int) → list[FeedEntry][source]#

Return up to limit kept entries numbered above after, in order.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

feed_bounds() → tuple[int, int][source]#

Return the number the feed is trimmed through, and the last number written.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

find_by_logical_key(scope_id: str, logical_key: str) → list[Record][source]#

Return the scope’s records with this logical key, ordered by local key.

A record matches if its synced side or its pending edit has the key. Several records can share one. Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

find_record(scope_id: str, object_id: str) → Record | None[source]#

Return the scope’s record with this object id, or None.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

get_record(local_key: int) → Record | None[source]#

Return the record with this local key, or None.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

get_scope(scope_id: str) → Scope | None[source]#

Return the scope with this id, or None.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

insert_record(record: Record) → int[source]#

Insert a new record and return the local key it got.

The record’s own local_key is ignored. A key is never given out twice, even after the record it went to was deleted. Runs inside a write transaction.

Raises:

  • RecordValueError – If the record holds what its state can’t. A record holds an edit exactly when it’s pending, in flight or in conflict, the server’s side of a conflict exactly when it’s in conflict, and no synced content exactly when its edit is a create. An edit holds a copy of the object unless it’s a delete, a conflict’s edit and server’s side match its kind, and only a refused create’s conflict names a holder. The record is checked first, so this is raised outside a transaction too.

  • TransactionRequiredError – If no transaction is open in this thread.

  • TransactionPermissionError – If the transaction only reads.

  • UnknownScopeError – If the record’s scope doesn’t exist.

  • DuplicateObjectError – If another record in the scope has the record’s object id.

put_scope(scope: Scope) → None[source]#

Insert the scope, or replace the one with its id.

Runs inside a write transaction.

Raises:
removed_records(through: int) → list[Record][source]#

Return the removed records no feed entry after through is about.

through is the number of a change feed entry. Each record keeps the number of the last entry about it, so these are the records removed by that entry or an earlier one, ordered by local key. Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

scope_records(scope_id: str, *, states: frozenset[RecordState] | None = None) → list[Record][source]#

Return the scope’s records, ordered by local key.

Runs inside any transaction.

Parameters:

  • scope_id – The scope to read.

  • states – Only the records in one of these states; None for every record. An empty set matches none, without asking the store, since not every database takes an empty list.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

scopes() → list[Scope][source]#

Return every scope, ordered by id.

Runs inside any transaction.

Raises:

TransactionRequiredError – If no transaction is open in this thread.

transaction(*, write: bool = True) → Iterator[None][source]#

Run the operations inside the block as one transaction.

Everything the block wrote is kept if it ends normally, and nothing is kept if it raises.

Parameters:

write – False for a block that only reads. Such a block can’t write, and a store may run it beside a write in another thread or process.

Raises:

  • TransactionError – If a transaction is already open in this thread.

  • StoreBusyError – If another process kept the store busy for too long.

  • StoreError – If the store was closed, or can’t start or keep the transaction.

trim_entries(through: int) → None[source]#

Delete every entry numbered at or below through.

feed_bounds() reports the highest through trimmed so far. Runs inside a write transaction.

Raises:
update_record(record: Record) → None[source]#

Replace the record with the same local key.

Runs inside a write transaction.

Raises:

A store that keeps everything in memory, and the base of stores that save it.

class calendaring_sync.store.memory.MemoryStore[source]#

Bases: StateStore

A store held in memory and lost when the process ends.

Each transaction works on its own view of the data: a read on the data the last write committed, a write on a copy of it, which becomes the committed data only when the transaction ends normally. So a read in one thread runs while another thread writes, and sees the data as it was when the read started. Lookups scan every record, so it suits test data, not large stores. Meant for tests; use SQLiteStore to keep state.

To keep the data somewhere else, such as a file written with pickle or an object database, subclass it and override the two methods that read and keep it: _load() returns the data the last write committed, and _save(data) keeps the data a write commits. The data is one object of plain containers holding the library’s frozen dataclasses, so it pickles. The checks, the transactions and every lookup come from the base classes.

_load runs at the start of every transaction, reads included, and can run in one thread while _save runs in another. So a store that keeps its data in a file reads the file once, when it opens, and leaves _load as it is: it passes what it read to super()._save(), and its own _save writes the file and then calls super()._save(), so reads come from memory. A store whose _load can’t run while _save does sets _reads_beside_writes to False.

If _save fails, the transaction fails and the last saved data stays, as long as _save doesn’t stop halfway: write to a new file and then replace the old one, for example. Turn the errors of the storage underneath into StoreError by overriding _store_error, as StateStore describes; anything else is reported as a bug.

Examples

A store that keeps its data as pickled bytes, read once when it opens, and a second store opened from those bytes:

>>> import pickle
>>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica
>>> class PickledStore(MemoryStore):
...     def __init__(self, saved=None):
...         super().__init__()
...         self.saved = saved
...         if saved:
...             super()._save(pickle.loads(saved))
...     def _save(self, data):
...         self.saved = pickle.dumps(data)
...         super()._save(data)
>>> first = PickledStore()
>>> listing = ChangeSet(
...     "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True
... )
>>> with Replica(first).session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(listing)
>>> Replica(PickledStore(first.saved)).find_record("work", "a").content
b'Team meeting'

A store kept in a SQLite database file.

class calendaring_sync.store.sqlite.SQLiteStore(path: str | PathLike[str], *, timeout: float = 5.0)[source]#

Bases: StateStore

A store in a SQLite database file, shared between threads and processes.

Writes take one connection, one transaction at a time, and SQLite’s write lock for their whole transaction, so two processes on one file never interleave. Reads take a second connection, one at a time, so a read in one thread doesn’t wait for another thread’s write. A database at ":memory:" or an empty path belongs to one connection, so it has only that one, and its transactions take turns.

In SQLite’s default journal mode, a commit locks reads out of the file. So it waits for the reads open in this process, and reads in another process make it wait up to timeout seconds, then raise StoreBusyError; the store keeps reads short for that. A write also keeps what it changes in memory until it commits, so that reads aren’t locked out before then: a full listing takes about as much memory again as its change set holds. In WAL mode reads and a commit don’t lock each other out, so a commit waits for no read and a write keeps nothing extra in memory. The mode is kept in the file, so set it once with PRAGMA journal_mode = WAL; WAL doesn’t work on a network filesystem.

Parameters:

  • path – The database file. It’s created, with the schema, if missing.

  • timeout – Seconds to wait for another process before giving up.

Raises:

  • TypeError – If path or timeout has the wrong type.

  • StoreVersionError – If a newer calendaring-sync wrote the file.

  • StoreBusyError – If another process kept the file busy past timeout.

  • StoreError – If the file can’t be opened or isn’t a store.

Last updated on 2026-10-21.