replica: The sync state#

The core’s entry point, which holds every sync rule on top of any store.

class calendaring_sync.replica.Replica(store: StateStore, *, clock: Callable[[], datetime] | None = None)[source]#

Bases: object

The sync state of every scope in one store.

This is the class your application and the adapter use: the adapter ingests what the server changed through a session(), and your application records its edits and reads records and the change feed. The store underneath only keeps the data.

Each method runs in a store transaction of its own, so two calls can see another thread’s or process’s write between them. To read several things at one point in time, open the store’s transaction() with write=False and call the store’s own methods inside it. A replica’s methods can’t run inside a transaction already open in the same thread; they raise TransactionError.

Every method raises TypeError for an argument of the wrong type, before it changes anything. Any method can also raise the store’s errors, StoreBusyError, StoreError and UnexpectedStoreError, none of which changes anything.

Parameters:

  • store – Where the state is kept.

  • clock – Returns the current time for every timestamp. Defaults to the current time in UTC.

Examples

A replica on a store in memory, with one scope that a full listing filled with one object. The examples on each method start from this replica. An adapter computes fingerprints from the content; these are short made-up ones.

>>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica
>>> replica = Replica(MemoryStore())
>>> listing = ChangeSet(
...     "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True
... )
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(listing)
>>> replica.find_record("work", "a").content
b'Team meeting'
>>> [(entry.kind, entry.object_id) for entry in replica.read_feed()]
[(<FeedEntryKind.CHANGED: 'changed'>, 'a')]
change(local_key: int, content: bytes, *, fingerprint: str, containers: frozenset[str] = frozenset({}), logical_key: str | None = None) → None[source]#

Record a change to an object.

A change back to the synced copy cancels the edit, a change after a delete replaces it, and a change to an object you created is still a create. Build fingerprint, containers and logical_key as for create().

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> replica.record(key).edit.content
b'Team meeting, moved'

A change back to the synced copy cancels the edit:

>>> replica.change(key, b"Team meeting", fingerprint="fp-a")
>>> replica.record(key).state
<RecordState.SYNCED: 'synced'>
create(scope_id: str, content: bytes, *, fingerprint: str, containers: frozenset[str] = frozenset({}), logical_key: str | None = None, object_id: str | None = None) → int[source]#

Record the creation of an object, and return its local key.

The record is pending until a push creates the object on the server. A create at the object id of a record the server removed takes that record over, as the server does when it has an object at that id again: it keeps its local key and drops the removed copy. Pass the fingerprint, containers and logical_key the adapter’s edit_fields() reads from the content; built any other way, an identical copy on the server looks different. An adapter’s own create does this for you.

Runs in a write transaction of its own.

Parameters:

  • scope_id – The scope the object belongs to.

  • content – The object’s content.

  • fingerprint – The content’s fingerprint.

  • containers – The containers the object belongs to.

  • logical_key – The object’s logical key, if it has one.

  • object_id – The object id the create chose, where the server takes ids from its clients; None where the server picks one.

Raises:

Examples

>>> key = replica.create("work", b"Lunch", fingerprint="fp-l", object_id="b")
>>> replica.record(key).state
<RecordState.PENDING: 'pending'>
>>> replica.record(key).edit.content
b'Lunch'
delete(local_key: int) → None[source]#

Record the deletion of an object.

Deleting an object you created that the server doesn’t have yet removes its record, leaving nothing to push.

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.delete(key)
>>> replica.record(key).edit.kind
<EditKind.DELETE: 'delete'>
discard(local_key: int) → None[source]#

Discard an object’s pending edit, so it matches the server’s copy again.

Discarding a create removes its record. A record with no edit is left as it is.

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.delete(key)
>>> replica.discard(key)
>>> replica.record(key).state
<RecordState.SYNCED: 'synced'>
drop_scope(scope_id: str) → DroppedScope[source]#

Drop a scope and every record in it.

Runs in a write transaction of its own.

Returns:

The records holding local work that the drop discarded.

Raises:

Examples

>>> dropped = replica.drop_scope("work")
>>> dropped.pending, dropped.conflicts
((), ())
>>> replica.scopes()
[]
feed_head() → int[source]#

Return the number of the last change feed entry written, 0 for none.

Start a new cursor here after reading every record.

Runs in a read transaction of its own.

Raises:

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

Examples

>>> replica.feed_head()
1
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, and removed records are included until they’re purged.

Runs in a read transaction of its own.

Raises:

Examples

>>> key = replica.create("work", b"Lunch", fingerprint="fp-l", logical_key="uid-1")
>>> found = replica.find_by_logical_key("work", "uid-1")
>>> [record.local_key for record in found] == [key]
True
find_record(scope_id: str, object_id: str) → Record | None[source]#

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

A removed record is returned until it’s purged.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.find_record("work", "a").state
<RecordState.SYNCED: 'synced'>
>>> replica.find_record("work", "z") is None
True
known_versions(scope_id: str) → dict[str, str][source]#

Return the version of the server’s copy of every object the store knows one for.

For a record in conflict that’s the server’s copy. An adapter compares these with the server’s during a full listing, and reports objects whose version matches as Unchanged instead of fetching them. Removed records aren’t included.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.known_versions("work")
{'a': 'v1'}
purge(through: int) → int[source]#

Purge removed records whose last feed entry is at or below through.

Their feed entries stay until trimmed.

Runs in a write transaction of its own.

Parameters:

through – The number of a change feed entry, such as the oldest cursor your application still reads from, so every part of it has seen these removals.

Returns:

How many records were purged.

Raises:

Examples

>>> from calendaring_sync import Removed
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {"a": Removed()}))
>>> replica.purge(replica.feed_head())
1
>>> replica.records("work")
[]
read_feed(after: int = 0, limit: int = 1000) → list[FeedEntry][source]#

Return the change feed entries after a cursor, oldest first.

Read the change feed shows how to keep a cursor so a crash never skips an entry.

Runs in a read transaction of its own.

Parameters:

  • after – The number of the last entry already handled.

  • limit – The most entries to return, at most FEED_PAGE_SIZE.

Raises:

Examples

>>> [(entry.number, entry.kind) for entry in replica.read_feed(after=0)]
[(1, <FeedEntryKind.CHANGED: 'changed'>)]
>>> replica.read_feed(after=replica.feed_head())
[]
record(local_key: int) → Record | None[source]#

Return the record with this local key, or None.

Runs in a read transaction of its own.

Raises:

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

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.record(key).object_id
'a'
records(scope_id: str, *, states: frozenset[RecordState] | None = None) → list[Record][source]#

Return the records in a scope, removed ones included, ordered by local key.

Runs in a read transaction of its own.

Parameters:

  • scope_id – The scope to read.

  • states – Only the records in one of these states; None for every record.

Raises:

Examples

>>> [record.object_id for record in replica.records("work")]
['a']
>>> replica.records("work", states=frozenset({RecordState.PENDING}))
[]
reset(scope_id: str) → None[source]#

Reset a scope, so its next sync is a full listing.

Records keep their state.

Runs in a write transaction of its own.

Raises:

Examples

>>> replica.reset("work")
>>> replica.scope("work").token is None
True
resolve(record: Record, resolution: KeepServer | KeepLocal | Merge | Delete | UpdateHolder) → None[source]#

Resolve one conflict, as your application decided from record.

Anything but the server’s copy becomes a pending edit over the server’s current copy, so the next sync pushes it against the server’s current version. Writes a CONFLICT_RESOLVED entry to the change feed. Runs in a write transaction of its own.

Parameters:

  • record – The record in conflict, as your application read it.

  • resolution – What to do with it.

Raises:

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

  • StaleConflictError – If the conflict changed since record was read: it ended, or its edit or the server’s copy moved on. A move of that copy’s version alone doesn’t count.

  • UnknownHolderError – For UpdateHolder, if the server named no record that’s still there, and no single record holds the refused edit’s logical key, or it was sent without one.

  • OpenConflictError – For UpdateHolder, if the holder is in conflict.

  • PushInFlightError – For UpdateHolder, if the holder’s push is in flight.

  • PendingEditError – For UpdateHolder, if the holder holds a pending edit of its own, which this would replace.

  • ValueError – If record isn’t in conflict, or the resolution doesn’t fit its kind: KeepLocal for an edit the server refused for its logical key, or UpdateHolder for any other.

Examples

>>> from calendaring_sync import KeepLocal
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, room 2", fingerprint="fp-mine")
>>> theirs = {"a": Changed("v2", "fp-theirs", b"Team meeting, room 3")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", theirs))
>>> record = replica.record(key)
>>> record.conflict.kind
<ConflictKind.BOTH_CHANGED: 'both_changed'>
>>> replica.resolve(record, KeepLocal())
>>> replica.record(key).state, replica.record(key).version
(<RecordState.PENDING: 'pending'>, 'v2')
resolve_conflicts(resolver: Callable[[Record], KeepServer | KeepLocal | Merge | Delete | UpdateHolder | None]) → ResolvedConflicts[source]#

Pass every open conflict to resolver, and resolve each as it answers.

The conflicts are read in one read transaction, then resolver runs outside any transaction, so it can take its time, and each answer is resolved with resolve() in a write transaction of its own. None leaves a record open. A record that changed after it was read is skipped, and comes back on the next call. An exception from resolver or resolve() stops the loop and keeps every resolution before it.

Parameters:

resolver – Your function. It gets each record in conflict, from every scope, and returns a resolution or None.

Returns:

Which conflicts were resolved, left open, or changed meanwhile.

Raises:

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

Examples

An application that takes the server’s copy in every conflict, without asking anyone, resolves them all in one call:

>>> from calendaring_sync import KeepServer
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, room 2", fingerprint="fp-mine")
>>> theirs = {"a": Changed("v2", "fp-theirs", b"Team meeting, room 3")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", theirs))
>>> replica.resolve_conflicts(lambda record: KeepServer()).resolved == (key,)
True
>>> replica.record(key).content
b'Team meeting, room 3'
scope(scope_id: str) → Scope[source]#

Return one scope.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.scope("work").token
't1'
scopes() → list[Scope][source]#

Return every scope, ordered by id.

Runs in a read transaction of its own.

Raises:

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

Examples

>>> [scope.id for scope in replica.scopes()]
['work']
session(scope_id: str, *, fingerprint_scheme: str, media_type: str, guarded_by_version: bool = False, unique_logical_key: bool = False) → Iterator[Session][source]#

Open a session for one scope, for the length of the with block.

The first session creates the scope. A different fingerprint scheme from the stored one resets the scope and clears every record’s version and fingerprint, so the next full listing fetches every object again. Objects whose content is the same as before aren’t reported as changed.

Opening the session and each ingest() run in a write transaction of their own.

Parameters:

  • scope_id – The scope’s id, chosen by the adapter.

  • fingerprint_scheme – The name of the adapter’s fingerprint scheme.

  • media_type – The media type of the scope’s content.

  • guarded_by_version – Whether the adapter’s pushes only apply if the object still has the version they were based on. A change or delete then waits for an ingest whenever its record has no version, so it never goes out unguarded.

  • unique_logical_key – Whether no two objects in the scope can share a logical key. A create whose answer never arrived is then recognized at the next ingest by its key and the data it was sent with, which a create the server picks the id for needs.

Raises:

Examples

>>> changes = {"a": Changed("v2", "fp-a2", b"Team meeting, moved")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", changes))
>>> replica.find_record("work", "a").content
b'Team meeting, moved'
trim_feed(through: int) → None[source]#

Delete every change feed entry numbered at or below through.

Runs in a write transaction of its own.

Raises:

Examples

>>> replica.trim_feed(replica.feed_head())
>>> replica.read_feed(after=replica.feed_head())
[]

Last updated on 2026-10-27.