Pools, deadlines and retries#

All importable from agensgraph, except where a module is named.

Pools#

class agensgraph.ConnectionPool( conninfo: ConninfoSource = '', *, graph: str | None = None, min_size: int = 4, max_size: int | None = 16, kwargs: dict[str, Any] | None = None, configure: Callable[[Connection[Any]], Awaitable[None]] | None = None, setup: Callable[[Connection[Any]], Awaitable[None]] | None = None, reset: Callable[[Connection[Any]], Awaitable[None]] | None = None, timeout: float = 30.0, max_waiting: int = 0, max_lifetime: float = 3600.0, max_idle: float = 600.0, reconnect_timeout: float = 300.0, reconnect_failed: Callable[[Any], None] | None = None, num_workers: int = 3, check_connections: bool = True, name: str | None = None, statement_timeout_gap: float = 0.5, )#

A pool of Connections, wrapping psycopg’s. graph is selected on each new connection; configure runs once per connection, setup on each borrow, reset on each return. The other arguments are psycopg’s. Creating it connects nothing: call open().

open(*, wait: bool = False, timeout: float = 30.0)#: Check the server with one connection of its own, then start the pool; with wait, until min_size exist.

wait(timeout: float = 30.0)#: Wait until min_size connections exist.

close(timeout: float = 5.0)#: Close the pool.

connection( *, timeout: float | None = None, deadline: Deadline | None = None, ) → Generator[Connection[Any]]#: Borrow a connection for a with block, within timeout or a Deadline.

execute_query(

query: Statement,

params: Params | None = None,

*,

timeout: float | None = None,

deadline: Deadline | None = None,

**kwargs: Any,

) → Result#: Borrow, run one statement with Connection.execute_query(), return.

invalidate() → int#: Retire every connection now held; returns the new generation.

drain()#: Close every connection and open the same number again.

check()#: Check the idle connections now, replacing any that fail.

resize(min_size: int, max_size: int | None = None)#: Change the pool’s size.

get_stats() → dict[str, int]#: psycopg’s counters with the pool’s own; every key always present.

pop_stats() → dict[str, int]#: The same, with the accumulating counters reset.

property generation: int#: The generation now in force.

property retired: int#: How many connections were closed for an older generation.

property interrupted: int#: How many were closed because a statement on them was interrupted.

property closed: bool#

property min_size: int#

property max_size: int#

property name: str#

class agensgraph.NullConnectionPool( conninfo: ConninfoSource = '', *, graph: str | None = None, min_size: int = 0, max_size: int | None = None, kwargs: dict[str, Any] | None = None, configure: Callable[[Connection[Any]], Awaitable[None]] | None = None, setup: Callable[[Connection[Any]], Awaitable[None]] | None = None, reset: Callable[[Connection[Any]], Awaitable[None]] | None = None, timeout: float = 30.0, max_waiting: int = 0, max_lifetime: float = 3600.0, max_idle: float = 600.0, reconnect_timeout: float = 300.0, reconnect_failed: Callable[[Any], None] | None = None, num_workers: int = 3, check_connections: bool = True, name: str | None = None, statement_timeout_gap: float = 0.5, )#: A pool that keeps no connections: one per borrow, closed after. The same interface.

class agensgraph.AsyncConnectionPool( conninfo: AsyncConninfoSource = '', *, graph: str | None = None, min_size: int = 4, max_size: int | None = 16, kwargs: dict[str, Any] | None = None, configure: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, setup: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, reset: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, timeout: float = 30.0, max_waiting: int = 0, max_lifetime: float = 3600.0, max_idle: float = 600.0, reconnect_timeout: float = 300.0, reconnect_failed: Callable[[Any], None] | None = None, num_workers: int = 3, check_connections: bool = True, name: str | None = None, statement_timeout_gap: float = 0.5, )#: ConnectionPool for asyncio. Its hooks are coroutine functions. open, wait, close, check, drain, resize and execute_query are awaited; connection() is an async with block.

class agensgraph.AsyncNullConnectionPool( conninfo: AsyncConninfoSource = '', *, graph: str | None = None, min_size: int = 0, max_size: int | None = None, kwargs: dict[str, Any] | None = None, configure: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, setup: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, reset: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None, timeout: float = 30.0, max_waiting: int = 0, max_lifetime: float = 3600.0, max_idle: float = 600.0, reconnect_timeout: float = 300.0, reconnect_failed: Callable[[Any], None] | None = None, num_workers: int = 3, check_connections: bool = True, name: str | None = None, statement_timeout_gap: float = 0.5, )#: NullConnectionPool for asyncio.

agensgraph.pool.COUNTERS = ('requests_num', 'requests_queued', 'requests_wait_ms', 'requests_errors', 'usage_ms', 'returns_bad', 'connections_num', 'connections_ms', 'connections_errors', 'connections_lost')#: The names of the pool counters that accumulate.

Deadlines#

See Timeouts and cancellation.

class agensgraph.Deadline(seconds: float | None = None, *, commit_reserve: float = 0.0)#

A budget of seconds, counted from when it is created. commit_reserve keeps part of it back from the statements.

classmethod none() → Deadline#: A budget with no limit.

remaining() → float | None#: Seconds left, or None without a limit; negative once spent.

spent() → float#: Seconds used.

available() → float | None#: What the statements may spend: what is left, less the reserve.

bounded(seconds: float | None) → float | None#: A wait of seconds, cut to what is left.

check(doing: str)#: Raise Expired, naming doing, if the budget is spent.

statement_timeout_ms(*, gap: float = 0.5) → int | None#: The statement_timeout it becomes, gap inside what is available; raises Expired if that is nothing.

property total: float | None#

property commit_reserve: float#

property unlimited: bool#

property expired: bool#

property can_commit: bool#: Whether the reserve is still there.

exception agensgraph.Expired#: A budget ran out before anything was sent. A subclass of TimeoutError, with spent and total.

Retries#

See Retries.

class agensgraph.Retryability#

What to do about a failure.

SAFE#: Run it again on the same connection.

RECONNECT#: Run it again on a new connection.

BACKPRESSURE#: Run it again, after a longer wait.

RESET_STATE#: Clear the stale state, keep the connection, run it again.

UNKNOWN#: Find out what happened before doing anything.

FATAL#: Do not run it again.

property is_retryable: bool#: True for the first four.

property needs_new_connection: bool#

property keeps_connection: bool#

property clears_prepared_statements: bool#

property wants_longer_delay: bool#

agensgraph.retryability( exc: BaseException, *, wrote: bool = False, merging: bool = False, ) → Retryability#: Classify an exception. wrote=True makes a lost connection UNKNOWN; merging=True makes a unique violation or a duplicate label from a MERGE race SAFE.

agensgraph.is_retryable(exc: BaseException, *, wrote: bool = False, merging: bool = False) → bool#: retryability(...).is_retryable.

class agensgraph.RetryPolicy( *, attempts: int = 3, base: float = 0.05, cap: float = 2.0, bucket: TokenBucket | None = None, rng: random.Random | None = None, )#

Decides whether and when to try again; the caller keeps the loop. attempts counts the first try. Waits are full jitter between zero and a ceiling doubling from base up to cap.

decide( exc: BaseException, *, number: int, wrote: bool = False, merging: bool = False, remaining: float | None = None, ) → Attempt#: An Attempt for the failure of attempt number.

succeeded()#: Credit the allowance after a success.

exhausted( exc: BaseException, *, attempts: int, previous: Sequence[BaseException], ) → BaseException#: Mark exc with its history and return it, for raising after the last attempt.

property attempts: int#

property bucket: TokenBucket#

class agensgraph.retry.Attempt#: RetryPolicy.decide()’s answer: number, recovery, retry, delay in seconds, and reason in words.

class agensgraph.TokenBucket(capacity: int = 100, *, refill: float = 120.0)#

An allowance of retries that time refills, in refill seconds from empty. Every RetryPolicy shares one per process unless given its own.

tokens() → float#: How many tokens are left.

spend(recovery: Retryability)#: Spend for a retry: 14 tokens, or 5 for BACKPRESSURE.

credit()#: Pay one back.

property allows_retry: bool#: Whether more than half is left.

property capacity: int#

Commit outcome#

See Losing a connection while committing.

class agensgraph.CommitOutcome#

What resolve_commit() found.

COMMITTED#: It landed.

ABORTED#: It did not.

IN_PROGRESS#: Not finished yet.

UNKNOWN#: The server no longer knows.

property is_settled: bool#: COMMITTED or ABORTED.

property safe_to_retry: bool#: Only ABORTED.