`session`: One sync of one scope#

A session, one sync of one scope: it ingests what changed, then pushes.

class calendaring_sync.session.Session(replica: Replica, scope_id: str, *, guarded_by_version: bool, unique_logical_key: bool)[source]#

Bases: object

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

A session ingests every page first, then the adapter pushes your pending edits: for each record to_push() lists, it claims it, sends it, and settles it with how the push ended.

Every method raises TypeError for an argument of the wrong type, before it changes anything.

Examples

The adapter pushes an edit after ingesting what changed on the server, here nothing:

>>> from calendaring_sync import Applied
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     for record in session.to_push():
...         claimed = session.claim(record.local_key)
...         # The adapter sends claimed.edit to the server here.
...         session.settle(claimed, Applied())
True
>>> replica.record(key).state, replica.record(key).content
(<RecordState.SYNCED: 'synced'>, b'Team meeting, moved')

claim(local_key: int, *, object_id: str | None = None) → Record | None[source]#

Mark a pending edit in flight, so no other process pushes it.

Send the record this returns, then report how the push ended with settle(). Runs in a write transaction of its own; when two processes claim one record, only one gets it.

Parameters:

local_key – The record to push.
object_id – For a create the server takes ids for, the id it’s about to be sent under. It’s stored with the claim, so the create is recognized by it even if the answer never arrives. An id a removed record holds passes to this one.

Returns:

The claimed record, or None when it no longer holds a pending edit to push: another process claimed it, the edit was discarded or resolved, or it waits for a version.

Raises:

TransactionError – If a store transaction is already open in this thread.
SessionError – If the session hasn’t ingested its last page, or has ended.
DuplicateObjectError – If a record that isn’t removed holds object_id.
OtherScopeError – If the record belongs to another scope.
ObjectIdAssignedError – If object_id is given for a record that has one.
LoneSurrogateError – If a string holds a lone surrogate.

Examples

A second claim of the same record gets nothing:

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     claimed = session.claim(key)
...     claimed.state, session.claim(key)
(<RecordState.IN_FLIGHT: 'in_flight'>, None)

ingest(change_set: ChangeSet) → None[source]#

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 content, 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. A complete change set, or the last page, also settles every push still in flight from what the server reported, and one it doesn’t mention is pending again. If anything fails, nothing changes.

Raises:

TransactionError – If a store transaction is already open in this thread.
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, has started pushing, or the page is a listing page in an incremental session or the other way round.
OtherScopeError – If the change set is for another scope.
ListingRequiredError – If an incremental change set arrives for a scope with no token.
UnknownVersionError – If an Unchanged entry names a record with no version.

property scope_id: str#: The id of the session’s scope.

settle(claimed: Record, outcome: Applied | Rejected | Gone | LogicalKeyTaken | Unknown) → bool[source]#

Record how the push of a claimed record ended.

Runs in a write transaction of its own, and writes a change feed entry for every outcome but Unknown. If another process already settled this push, for example by ingesting while it was in flight, the outcome only counts while the record still holds the same edit and no one has claimed it again.

Parameters:

claimed – The record claim() returned.
outcome – How the push ended.

Returns:

True if the store now agrees with the outcome, False if the push had already been settled another way and nothing was written.

Raises:

TransactionError – If a store transaction is already open in this thread.
SessionError – If the session has ended.
DuplicateObjectError – If Applied names an id a record that isn’t removed holds.
ValueError – If claimed isn’t a record this session’s claim() returned, or the outcome is one no push of its edit can have: Gone for a create, LogicalKeyTaken for a delete, Applied with an id for a delete, or Applied without one for a create that has none.

Examples

The server refused the write, so the edit is pending again:

>>> from calendaring_sync import Rejected
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     session.settle(session.claim(key), Rejected())
True
>>> replica.record(key).state
<RecordState.PENDING: 'pending'>

property state: SessionState#: Where the session is.

to_push() → list[Record][source]#

Return the pending edits ready to push, by local key.

With guarded_by_version, a change or delete of a record with no version waits for an ingest that gives it one, and isn’t listed. Runs in a read transaction of its own.

Raises:

TransactionError – If a store transaction is already open in this thread.
SessionError – If the session hasn’t ingested its last page, or has ended.

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     [record.local_key for record in session.to_push()] == [key]
True

class calendaring_sync.session.SessionState(*values)[source]#

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.

PUSHING = 'pushing'#: A push was claimed; no more pages can be ingested.

session: One sync of one scope#

`session`: One sync of one scope#