replica: Sessions, ingesting, and the change feed

Replica is the entry point. Open a session for a scope, ingest the change sets an adapter fetched, then read the change feed to learn what changed.

The core’s entry point: sessions, ingesting change sets, and reading the change feed.

Every sync rule lives here, on top of any StateStore.

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

Bases: object

The sync state of every scope in one store.

Any method can also raise the store’s errors: StoreBusyError when another process kept the store busy, which is safe to retry, StoreError when the store can’t read or write its data, and UnexpectedStoreError for a bug in the store. None of them changes anything.

Parameters:

• store – Where the state is kept.

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

drop_scope(scope_id: str) → DroppedScope

Drop a scope and every record in it.

Returns:

The records holding local work that the drop discarded.

Raises:

• UnknownScopeError – If there’s no scope with this id.

• SessionError – If a session for the scope is open.

feed_head() → int

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

Start a new cursor here after reading every record.

known_versions(scope_id: str) → dict[str, str]

Return the version of every object in a scope the store has one for.

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.

Raises:

UnknownScopeError – If there’s no scope with this id.

purge(through: int) → int

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

Their feed entries stay until trimmed.

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:

ValueError – If through is negative or past the last entry.

read_feed(after: int = 0, limit: int = 1000) → list[FeedEntry]

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

Save the new cursor only after handling the entries, so a crash in between reads them again instead of skipping them.

Parameters:

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

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

Raises:

• TrimmedFeedError – If entries after after were trimmed.

• ValueError – If after is negative or past the last entry, or limit is outside 1 to FEED_PAGE_SIZE.

record(local_key: int) → Record | None

Return the record with this local key, or None.

records(scope_id: str) → list[Record]

Return every record in a scope, removed ones included, ordered by local key.

Raises:

UnknownScopeError – If there’s no scope with this id.

reset(scope_id: str) → None

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

Records keep their state.

Raises:

• UnknownScopeError – If there’s no scope with this id.

• SessionError – If a session for the scope is open.

scope(scope_id: str) → Scope

Return one scope.

Raises:

UnknownScopeError – If there’s no scope with this id.

scopes() → list[Scope]

Return every scope, ordered by id.

session(scope_id: str, *, fingerprint_scheme: str, media_type: str) → Iterator[Session]

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 text is the same as before aren’t reported as changed.

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.

Raises:

• SessionError – If this replica already has a session open for the scope.

• ScopeMismatchError – If the scope’s content has another media type.

trim_feed(through: int) → None

Delete every change feed entry numbered at or below through.

Raises:

ValueError – If through is negative or past the last entry.

class calendaring_sync.replica.Session(replica: Replica, scope_id: str)

Bases: object

One sync of one scope, opened with Replica.session().

ingest(change_set: ChangeSet) → None

Ingest one change set, or one page of it, in one transaction.

The core applies each object in it: a new object gets a record, a changed one gets the new data, and a removed one is marked removed. An object whose fingerprint, containers and logical key didn’t change only gets its version and the server’s text, and isn’t reported. A truncated page of an incremental change set moves the token to its checkpoint. A full listing moves the token only on its last page, which also removes every record the listing never mentioned. If anything fails, nothing changes.

Raises:

• StaleChangeSetError – If the change set was fetched from a token the scope no longer has, or is a later page of a listing with another token, such as one another session has replaced. The session goes back to idle, so the adapter can fetch again and ingest from the start in the same session.

• SessionError – If the session has ended, already ingested its last page, or the page is a listing page in an incremental session or the other way round.

• ValueError – If the change set is for another scope, an incremental change set arrives for a scope with no token, or an Unchanged entry is outside a listing or names a record with no version.

property scope_id: str

The id of the session’s scope.

property state: SessionState

Where the session is.

class calendaring_sync.replica.SessionState(*values)

Bases: StrEnum

Where a session is.

IDLE = 'idle'

Nothing ingested yet.

INGESTED = 'ingested'

The last page was ingested, and the token is final.

INGESTING = 'ingesting'

A truncated page was ingested, and more pages follow.