Stores¶
once.Status ¶
Bases: str, Enum
once.Record
dataclass
¶
once.Store ¶
Bases: Protocol
What Once needs from a storage backend.
Timestamps are ISO 8601 UTC strings with microseconds. A pending record is owned by whoever set its current updated_at value; claim and take return that value as the owner token, and finish only applies while it matches.
claim ¶
claim(key: str, args_hash: str | None) -> str | None
Insert a pending record. Returns an owner token, or None if the key exists.
finish ¶
finish(key: str, token: str, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> bool
Move a pending record we own to its final state.
take ¶
take(record: Record, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> str | None
Change a record only if it is unchanged since it was read.
renew ¶
renew(record: Record, args_hash: str | None) -> str | None
Turn an expired record back into a fresh pending one, if it is unchanged.
Returns an owner token, or None if someone else changed it first.
once.SQLiteStore ¶
Effect records in a SQLite file, using WAL mode and a busy timeout.
Each thread gets its own connection. Connections run in autocommit mode, so every statement is its own transaction.
A pending record is owned by whoever set its current updated_at value. That value is returned as a token by claim and take, and finish only applies when the token still matches. This keeps a slow owner from overwriting a record that was taken over after verify.
claim ¶
claim(key: str, args_hash: str | None) -> str | None
Insert a pending record. Returns an owner token, or None if the key exists.
finish ¶
finish(key: str, token: str, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> bool
Move a pending record we own to its final state.
take ¶
take(record: Record, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> str | None
Change a record only if it is unchanged since it was read.
Returns the new token, or None if someone else changed it first.
renew ¶
renew(record: Record, args_hash: str | None) -> str | None
Turn an expired record back into a fresh pending one, if it is unchanged.
Returns an owner token, or None if someone else changed it first.
records ¶
records(statuses: Iterable[Status] | None = None, limit: int | None = None) -> list[Record]
Records, most recently updated first.
purge ¶
purge(older_than: float, statuses: Iterable[Status]) -> int
Delete records with one of statuses not updated for older_than seconds.
once.postgres.PostgresStore ¶
Effect records in a Postgres table.
a libpq connection string or postgresql:// URL. The scheme can be
any case and can carry a driver suffix (postgresql+psycopg://).
table: table name, optionally schema qualified ("myschema.effects"). Created if missing, unless create is False. A table made before ttl support gets its expires_at column added either way.
connect_timeout: seconds to wait for a connection. statement_timeout: seconds any one statement may run before the server cancels it, 30 by default. None leaves the server's setting alone.
One connection per store, in autocommit mode, shared by all threads with a lock. Every statement is its own transaction, so no row stays locked between calls. Safety between processes and hosts comes from the primary key and the conditional updates, not from anything in this process. If the connection drops, the statement is retried once on a new one. A write that landed but lost its reply then looks like it lost a race, which is the safe side: the call reports pending instead of running twice.
claim ¶
claim(key: str, args_hash: str | None) -> str | None
Insert a pending record. Returns an owner token, or None if the key exists.
finish ¶
finish(key: str, token: str, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> bool
Move a pending record we own to its final state.
take ¶
take(record: Record, status: Status, result: str | None = None, error: str | None = None, expires_at: str | None = None) -> str | None
Change a record only if it is unchanged since it was read.
Returns the new token, or None if someone else changed it first.
renew ¶
renew(record: Record, args_hash: str | None) -> str | None
Turn an expired record back into a fresh pending one, if it is unchanged.
Returns an owner token, or None if someone else changed it first.
records ¶
records(statuses: Iterable[Status] | None = None, limit: int | None = None) -> list[Record]
Records, most recently updated first.
purge ¶
purge(older_than: float, statuses: Iterable[Status]) -> int
Delete records with one of statuses not updated for older_than seconds.