store: Where sync state lives#

Where the core keeps sync state.

The store, the storage interface it keeps its data in, and the stores that ship.

The store interface, and the storage interface a store keeps its data in.

A store only keeps data. Every sync rule lives in Replica, and every rule about how a store behaves, such as its transactions, its checks and its errors, lives in StateStore. Where the data goes is a Storage, which implements the same small set of typed operations and nothing more.

class calendaring_sync.store.base.StateStore(storage: Storage)[source]#

Bases: object

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

It keeps them in a Storage, and adds every rule a store follows, whichever storage it uses. 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 where the storage 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 storage, which is a bug

No exception of the storage underneath reaches the caller.

Parameters:

storage – Where the store keeps its data.

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 copy or its local 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 copy in a conflict exactly when it’s in conflict, and no synced copy 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 copy 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 always returns 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:
class calendaring_sync.store.base.Storage[source]#

Bases: ABC

Where a StateStore keeps its data.

To keep sync state somewhere new, such as in another database, implement this and a StorageTransaction, and pass the storage to StateStore. The store adds every check, transaction rule, thread rule and error a store needs, so a storage only reads and writes.

abstractmethod begin(*, write: bool) → StorageTransaction[source]#

Start a transaction and return it.

The store runs one write transaction at a time, and transactions that only read beside it only if reads_beside_writes is set. If this raises, it leaves nothing open.

close() → None[source]#

Release what the storage holds open.

The store calls it once, after every transaction ended.

error(exc: Exception) → StoreError | None[source]#

Return the store error for an exception the storage raised, or None.

Turn the failures of whatever the storage uses, such as a file it can’t open, into StoreError or one of its subclasses. The store reports any exception this returns None for as a bug in the storage.

reads_beside_writes: bool = False#

Whether a transaction that only reads can run while another thread’s write transaction is open. With False, every transaction waits for the one before it to end. Set it to True only where a read sees the data as the last commit left it until it ends, whatever a write does meanwhile; reads in other threads then run at the same time too, so begin() and the transactions it returns must be safe for that.

class calendaring_sync.store.base.StorageTransaction(*, write: bool)[source]#

Bases: ABC

One transaction on a Storage, which Storage.begin() returns.

It holds whatever its transaction needs, such as a connection or a view of the data, so a storage keeps no state per thread. Its data operations are the ones StateStore offers, without the checks: StateStore calls them only inside this transaction, after it checked the arguments, the records and whether the transaction may write, and turns every exception they raise into a store error. Each returns what the StateStore method of the same name returns.

Parameters:

write – Whether the transaction may write.

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

Do what StateStore.append_entry() does, without its checks.

abstractmethod commit() → None[source]#

Keep what the transaction wrote, and end it.

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

Do what StateStore.delete_record() does, without its checks.

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

Do what StateStore.delete_scope() does, without its checks.

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

Do what StateStore.entries() does, without its checks.

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

Do what StateStore.feed_bounds() does, without its checks.

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

Do what StateStore.find_by_logical_key() does, without its checks.

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

Do what StateStore.find_record() does, without its checks.

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

Do what StateStore.get_record() does, without its checks.

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

Do what StateStore.get_scope() does, without its checks.

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

Do what StateStore.insert_record() does, without its checks.

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

Do what StateStore.put_scope() does, without its checks.

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

Do what StateStore.removed_records() does, without its checks.

abstractmethod rollback() → None[source]#

Discard what the transaction wrote, and end it.

It’s also called after commit() raised, so it ends whatever the transaction still holds open.

abstractmethod scope_records(scope_id: str) → list[Record][source]#

Return every record of the scope, ordered by local key.

scope_records_in(scope_id: str, states: frozenset[RecordState]) → list[Record][source]#

Return the scope’s records in one of states, ordered by local key.

This filters what scope_records() returns. A storage that can filter where it keeps its records, such as with an index, overrides it. It never gets an empty set.

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

Do what StateStore.scopes() does, without its checks.

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

Do what StateStore.trim_entries() does, without its checks.

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

Do what StateStore.update_record() does, without its checks.

write#

Whether the transaction may write.

A store that keeps everything in memory, and can hand its data to a keeper to save.

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

Bases: ABC

Keeps a MemoryStore’s data somewhere else, such as in a file.

Pass one to MemoryStore. It loads the data once, when the store opens, and saves it each time a write commits; reads come from memory. If save() raises, the write fails and the store keeps the data it saved last, as long as save() doesn’t stop halfway: write to a new file and then replace the old one, for example.

What MemoryData holds can change between releases, so data a keeper saved with one release may not load in another. Keep sync state that has to outlive an upgrade in SQLiteStore, whose file a newer release reads.

error(exc: Exception) → StoreError | None[source]#

Return the store error for an exception load() or save() raised.

Turn the failures of the storage underneath, such as a full disk, into StoreError; anything this returns None for is reported as a bug.

abstractmethod load() → MemoryData | None[source]#

Return the data saved last, or None for a store with nothing saved yet.

abstractmethod save(data: MemoryData) → None[source]#

Keep the data a write commits, in place of what was saved before.

class calendaring_sync.store.memory.MemoryStore(*, keeper: MemoryKeeper | None = None)[source]#

Bases: StateStore

A store held in memory, lost when the process ends unless a keeper saves it.

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 and small stores; use SQLiteStore for large ones.

Parameters:

keeper – Where to save the data, such as a file, so the next process can load it. None keeps it only in memory.

Raises:

StoreError – If the keeper can’t load the data, as its error() says.

Examples

A keeper that saves the data as pickled bytes, and a second store loaded from them:

>>> import pickle
>>> from calendaring_sync import Changed, ChangeSet, MemoryKeeper, MemoryStore, Replica
>>> class Pickled(MemoryKeeper):
...     def __init__(self, saved=None):
...         self.saved = saved
...     def load(self):
...         return pickle.loads(self.saved) if self.saved else None
...     def save(self, data):
...         self.saved = pickle.dumps(data)
>>> keeper = Pickled()
>>> listing = ChangeSet(
...     "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True
... )
>>> with Replica(MemoryStore(keeper=keeper)).session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(listing)
>>> store = MemoryStore(keeper=Pickled(keeper.saved))
>>> Replica(store).find_record("work", "a").content
b'Team meeting'
class calendaring_sync.store.memory.MemoryData[source]#

Bases: object

Everything a MemoryStore holds, as one object.

A MemoryKeeper saves and loads it whole. It’s made of plain containers holding the library’s frozen dataclasses, so it pickles; what’s inside isn’t part of the API.

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-28.