Sync

`picopyn.synchronous`

Sync API based on psycopg for connection management and pools.

`apilevel = '2.0'` `module-attribute`

Supported DB-API level.

`connect = Connection.connect` `module-attribute`

DB-API 2.0 module-level constructor; alias for Connection.connect.

`paramstyle = 'format'` `module-attribute`

Parameter marker style: ANSI C printf, e.g. WHERE name = %s.

`threadsafety = 1` `module-attribute`

Threads may share the module, but not connections.

`Connection`

A representation of a database session.

Can be used as a context manager; the connection is closed on exit from the with block.

`autocommit` `property` `writable`

Always True: Picodata does not support explicit transaction control (COMMIT) yet, so every connection is opened in autocommit mode.

See commit and rollback, which are no-ops for the same reason.

Raises:

Type	Description
`ValueError`	When assigned a value other than `True`.

`cancel()`

Ask the server to cancel the query running on this connection.

Unlike close() it is safe to call from a thread that doesn't own the connection: the request goes over a separate cancellation channel, so the owner thread keeps using the connection and just gets an error from its query. Does nothing if there is no query to cancel.

Source code in picopyn/synchronous/connection.py

def cancel(self) -> None:
    """Ask the server to cancel the query running on this connection.

    Unlike [close()][picopyn.synchronous.connection.Connection.close] it is safe
    to call from a thread that doesn't own the connection: the request goes over a
    separate cancellation channel, so the owner thread keeps using the connection
    and just gets an error from its query. Does nothing if there is no query to
    cancel.
    """
    conn = self.conn
    if conn is None:
        return

    try:
        conn.cancel_safe()
    except Exception as e:
        # cancel_safe() uses a separate connection. Its failure alone says
        # nothing about the connection running the query.
        if conn.broken:
            self._notify_query_error(e)
        raise RuntimeError(
            f"Failed to cancel query on picodata instance {self.dsn}: {e}"
        ) from e

`close()`

Close the connection gracefully.

Source code in picopyn/synchronous/connection.py

def close(self) -> None:
    """
    Close the connection gracefully.
    """
    if self.conn is None:
        return

    conn = self.conn

    try:
        conn.close()
    except Exception as e:
        raise RuntimeError(
            f"Failed to disconnect from picodata instance {self.dsn}: {e}"
        ) from e
    finally:
        self.conn = None

`commit()`

No-op for SQLAlchemy compatibility.

DB-API defines this method as committing any pending transaction.

Source code in picopyn/synchronous/connection.py

def commit(self) -> None:
    """
    No-op for SQLAlchemy compatibility.

    DB-API defines this method as committing any pending transaction.
    """
    self._ensure_open()

`connect(dsn, on_query_metadata=None, **connect_kwargs)` `classmethod`

Create and return a new connection to Picodata.

Parameters:

Name	Type	Description	Default
`dsn`	`str`	The data source name (e.g., "postgresql://user:pass@host:port" or "postgresql://user:pass@host1:port1,host2:port2") for the picodata node.	required
`on_query_metadata`	`Callable[[PreparedStatementMetadata], None] \| None`	Internal -- used by `Pool`'s dedicated metadata connection to populate its cache. Not useful for typical application code.	`None`
`**connect_kwargs`	`Any`	Additional keyword arguments to pass to psycopg.connect(). SSL can be configured either via DSN query params or via kwargs such as `sslmode`, `sslrootcert`, `sslcert`, `sslkey`.	`{}`

Note

SSL should be configured using the appropriate parameters in a single source: either via DSN or by providing libpq SSL parameters in kwargs.

Examples:

Simple usage:

conn = Connection.connect("postgresql://admin:pass@localhost:5432")
cur = conn.cursor()
cur.execute("SELECT 1")
conn.close()

Or, using a context manager:

with Connection.connect("postgresql://admin:pass@localhost:5432") as conn:
    cur = conn.cursor()
    cur.execute("SELECT 1")

Multi-host DSN string:

conn = Connection.connect("postgresql://admin:pass@host1:5432,host2:5432")

SSL (kwargs-based):

conn = Connection.connect(
    "postgresql://admin:pass@localhost:5432",
    sslmode="verify-ca",
    sslrootcert="/path/to/ca.crt",
    sslcert="/path/to/client.crt",
    sslkey="/path/to/client.key",
)

SSL (DSN query params):

conn = Connection.connect(
    "postgresql://admin:pass@localhost:5432/db"
    "?sslmode=verify-ca&sslrootcert=/path/to/ca.crt"
    "&sslcert=/path/to/client.crt&sslkey=/path/to/client.key"
)

Source code in picopyn/synchronous/connection.py

@classmethod
def connect(
    cls,
    dsn: str,
    on_query_metadata: Callable[[PreparedStatementMetadata], None] | None = None,
    **connect_kwargs: Any,
) -> Self:
    """
    Create and return a new connection to Picodata.

    Args:
        dsn: The data source name (e.g., "postgresql://user:pass@host:port" or
            "postgresql://user:pass@host1:port1,host2:port2") for the picodata node.
        on_query_metadata: Internal -- used by [`Pool`][picopyn.synchronous.pool.Pool]'s
            dedicated metadata connection to populate its
            [cache][picopyn.query_metadata.QueryMetadataCache]. Not useful for typical
            application code.
        **connect_kwargs: Additional keyword arguments to pass to psycopg.connect().
            SSL can be configured either via DSN query params or via kwargs such as
            `sslmode`, `sslrootcert`, `sslcert`, `sslkey`.

    Note:
        SSL should be configured using the appropriate parameters in a single source:
        either via DSN or by providing libpq SSL parameters in kwargs.

    Examples:
        Simple usage:
        ```python
        conn = Connection.connect("postgresql://admin:pass@localhost:5432")
        cur = conn.cursor()
        cur.execute("SELECT 1")
        conn.close()
        ```

        Or, using a context manager:
        ```python
        with Connection.connect("postgresql://admin:pass@localhost:5432") as conn:
            cur = conn.cursor()
            cur.execute("SELECT 1")
        ```

        Multi-host DSN string:

        ```python
        conn = Connection.connect("postgresql://admin:pass@host1:5432,host2:5432")
        ```

        SSL (kwargs-based):

        ```python
        conn = Connection.connect(
            "postgresql://admin:pass@localhost:5432",
            sslmode="verify-ca",
            sslrootcert="/path/to/ca.crt",
            sslcert="/path/to/client.crt",
            sslkey="/path/to/client.key",
        )
        ```

        SSL (DSN query params):

        ```python
        conn = Connection.connect(
            "postgresql://admin:pass@localhost:5432/db"
            "?sslmode=verify-ca&sslrootcert=/path/to/ca.crt"
            "&sslcert=/path/to/client.crt&sslkey=/path/to/client.key"
        )
        ```
    """
    if dsn is None:
        raise ValueError("dsn can not be None")

    validate_ssl_source(dsn, connect_kwargs)

    autocommit = connect_kwargs.pop("autocommit", None)
    if autocommit is not None and autocommit is not True:
        raise ValueError("autocommit=False is not supported")

    connection = cls.__new__(cls)
    # DSN with hidden password
    connection.dsn = _redact_dsn(dsn)
    # origin DSN to connect
    connection._connect_dsn = dsn
    connection._connect_kwargs = connect_kwargs
    connection.conn = None
    # Note: on_query_metadata must be a plain function, not a coroutine --
    # same reason as _on_query_metadata_notice, which calls it
    connection._on_query_metadata = on_query_metadata
    connection._on_query_error = None
    connection.instance_uuid = None
    # is acquired connection must be closed on release
    connection.to_evict = False

    connect_kwargs = _with_picodata_startup_options(
        connection._connect_kwargs,
        connection._connect_dsn,
        request_query_metadata=connection._on_query_metadata is not None,
    )
    try:
        # With psycopg's default autocommit=False, even SELECT opens an implicit transaction.
        # See: https://www.psycopg.org/psycopg3/docs/basic/transactions.html#autocommit-transactions
        # Required until Picodata supports COMMIT.
        connection.conn = psycopg.connect(
            connection._connect_dsn,
            autocommit=True,
            **connect_kwargs,
        )
    except Exception as error:
        raise RuntimeError(
            f"Failed to connect to picodata instance using DSN {connection.dsn}: {error}"
        ) from error

    # register the Notice listener
    if connection._on_query_metadata is not None:
        assert connection.conn is not None
        connection.conn.add_notice_handler(connection._on_query_metadata_notice)

    return connection

`cursor()`

Return a new cursor object using the connection, which is used to manage the context of a fetch operation.

Returns:

Type	Description
`Cursor`	A new Cursor bound to this connection.

See Cursor for full documentation.

Source code in picopyn/synchronous/connection.py

def cursor(self) -> Cursor:
    """
    Return a new cursor object using the connection, which is used to
    manage the context of a fetch operation.

    Returns:
        A new Cursor bound to this connection.

    See [`Cursor`][picopyn.synchronous.cursor.Cursor] for full documentation.
    """
    self._ensure_open()

    assert self.conn is not None
    cursor = Cursor(self.conn.cursor())
    cursor._on_query_error = self._notify_query_error
    return cursor

`is_closed()`

Returns:

Type	Description
`bool`	True if there is no active underlying connection.

Source code in picopyn/synchronous/connection.py

def is_closed(self) -> bool:
    """
    Returns:
        True if there is no active underlying connection.
    """
    return self.conn is None or self.conn.closed

`prepare(query, parameters=None)`

Create a prepared statement for query.

Call this directly if you specifically want to force a Parse now -- this is what query metadata service uses on its dedicated connection to trigger Picodata's metadata Notice.

Parameters:

Name	Type	Description	Default
`query`	`QueryNoTemplate`	The SQL query text to prepare.	required
`parameters`	`Params \| None`	Optional DB-API parameters used only to let psycopg convert `%s` placeholders to PostgreSQL's `$1`/`$2` format before sending the Parse message. The query is not executed.	`None`

Returns:

Type	Description
`str`	The query text sent to Picodata in the Parse message.

Raises:

Type	Description
`InterfaceError`	If there is no active connection.
`Error`	If preparing the statement fails.

Source code in picopyn/synchronous/connection.py

def prepare(self, query: QueryNoTemplate, parameters: Params | None = None) -> str:
    """Create a prepared statement for `query`.

    Call this directly if you specifically want to force a `Parse` now --
    this is what
    [query metadata service][picopyn.synchronous.metadata_service.QueryMetadataService]
    uses on its dedicated connection to trigger Picodata's metadata Notice.

    Args:
        query: The SQL query text to prepare.
        parameters: Optional DB-API parameters used only to let psycopg
            convert `%s` placeholders to PostgreSQL's `$1`/`$2` format before
            sending the Parse message. The query is not executed.

    Returns:
        The query text sent to Picodata in the Parse message.

    Raises:
        InterfaceError: If there is no active connection.
        Error: If preparing the statement fails.
    """
    self._ensure_open()
    assert self.conn is not None

    try:
        encoding = getattr(self.conn.pgconn, "_encoding", "utf-8")
        cursor = self.conn.cursor()
        try:
            # psycopg exposes DB-API placeholder conversion on internals;
            # we need it here to send Parse without executing the statement.
            pg_query = PostgresQuery(psycopg.adapt.Transformer(cursor))
            pg_query.convert(query, parameters)
        finally:
            cursor.close()
        result = self.conn.pgconn.prepare(b"", pg_query.query, param_types=pg_query.types)
        if result.status != psycopg.pq.ExecStatus.COMMAND_OK:
            if result.status == psycopg.pq.ExecStatus.FATAL_ERROR:
                raise psycopg.errors.error_from_result(result, encoding=encoding)
            raise psycopg.InterfaceError(
                f"unexpected result {psycopg.pq.ExecStatus(result.status).name}"
                " from preparing statement"
            )
        return pg_query.query.decode(encoding)
    except psycopg.Error as error:
        self._notify_query_error(error)
        _raise_dbapi_error(error)

`rollback()`

No-op for SQLAlchemy compatibility.

DB-API defines this method as rolling back to the start of any pending transaction.

Source code in picopyn/synchronous/connection.py

def rollback(self) -> None:
    """
    No-op for SQLAlchemy compatibility.

    DB-API defines this method as rolling back to the start of any pending transaction.
    """
    self._ensure_open()

`Cursor`

DB-API cursor wrapper.

Supports query execution, cursor metadata and fetch methods.

`arraysize` `property` `writable`

Number of rows to fetch at a time when fetchmany() is called without size.

Returns:

Type	Description
`int`	The current arraysize.

`description` `property`

Return DB-API column metadata for the current result set.

Returns:

Type	Description
`Description \| None`	A tuple of per-column metadata (name, type_code, display_size, internal_size, precision, scale, null_ok), or None if there is no result set.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
[col[0] for col in cursor.description]
# ['id', 'item']

`rowcount` `property`

Return the number of rows produced or affected by the last operation.

Returns:

Type	Description
`int`	The row count, or -1 if it cannot be determined.

`statusmessage` `property`

Return the command status from the last operation.

Returns:

Type	Description
`str \| None`	The status message reported by the database, or None if no command
`str \| None`	status is available.

Examples:

cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
cursor.statusmessage
# 'INSERT 0 1'

`iter()`

Iterate over the current result set.

Returns:

Type	Description
`Iterator[tuple[object, ...]]`	An iterator over the remaining rows.

Examples:

cursor.execute("SELECT id, name FROM users")
for row in cursor:
    print(row)

Source code in picopyn/synchronous/cursor.py

def __iter__(self) -> Iterator[tuple[object, ...]]:
    """
    Iterate over the current result set.

    Returns:
        An iterator over the remaining rows.

    Examples:
        ```python
        cursor.execute("SELECT id, name FROM users")
        for row in cursor:
            print(row)
        ```
    """
    self._ensure_result_set()

    def iterate() -> Iterator[tuple[object, ...]]:
        try:
            yield from self._cursor
        except psycopg.Error as error:
            self._notify_query_error(error)
            raise

    return iterate()

`close()`

Close the cursor.

Source code in picopyn/synchronous/cursor.py

def close(self) -> None:
    """
    Close the cursor.
    """
    self._cursor.close()

`execute(operation, parameters=None)`

Prepare and execute a database operation (query or command).

Parameters:

Name	Type	Description	Default
`operation`	`QueryNoTemplate`	database operation to be prepared.	required
`parameters`	`Params \| None`	sequence or mapping that will be bound to variables in the operation.	`None`

Returns:

Type	Description
`Cursor`	This cursor, to allow chaining (e.g. `cursor.execute(query).fetchall()`).

Examples:

DDL example:

ddl = 'CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id);'
cursor.execute(ddl)

DML example:

cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))

Source code in picopyn/synchronous/cursor.py

def execute(self, operation: QueryNoTemplate, parameters: Params | None = None) -> "Cursor":
    """
    Prepare and execute a database operation (query or command).

    Args:
        operation: database operation to be prepared.
        parameters: sequence or mapping that will be bound to variables in the operation.

    Returns:
        This cursor, to allow chaining (e.g. `cursor.execute(query).fetchall()`).

    Examples:
        DDL example:

        ```python
        ddl = 'CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id);'
        cursor.execute(ddl)
        ```

        DML example:

        ```python
        cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
        ```
    """
    self._ensure_open()

    try:
        self._cursor.execute(operation, parameters)
    except psycopg.Error as error:
        self._notify_query_error(error)
        self._clear_stmt_cache_if_stale(error)
        _raise_dbapi_error(error)

    return self

`executemany(operation, seq_of_parameters)`

Prepare a database operation and execute it against all parameter sequences or mappings found in the sequence seq_of_parameters.

Parameters:

Name	Type	Description	Default
`operation`	`QueryNoTemplate`	database operation to be prepared.	required
`seq_of_parameters`	`Iterable[Params]`	all parameter sequences or mappings that will be bound to variables in the operation.	required

Examples:

cursor.executemany(
    'INSERT INTO "warehouse" VALUES (%s, %s)',
    [(1, "test1"), (2, "test2")],
)

Source code in picopyn/synchronous/cursor.py

def executemany(self, operation: QueryNoTemplate, seq_of_parameters: Iterable[Params]) -> None:
    """
    Prepare a database operation and execute it against all parameter sequences
    or mappings found in the sequence seq_of_parameters.

    Args:
        operation: database operation to be prepared.
        seq_of_parameters: all parameter sequences or mappings that will be bound to variables in the operation.

    Examples:
        ```python
        cursor.executemany(
            'INSERT INTO "warehouse" VALUES (%s, %s)',
            [(1, "test1"), (2, "test2")],
        )
        ```
    """
    self._ensure_open()
    try:
        self._cursor.executemany(operation, seq_of_parameters)
    except psycopg.Error as error:
        self._notify_query_error(error)
        self._clear_stmt_cache_if_stale(error)
        _raise_dbapi_error(error)

`fetchall()`

Fetch all remaining rows from the current result set.

Returns:

Type	Description
`list[tuple[object, ...]]`	A list of the remaining rows. Empty if no more rows are available.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
rows = cursor.fetchall()
# [(1, 'test')]

Source code in picopyn/synchronous/cursor.py

def fetchall(self) -> list[tuple[object, ...]]:
    """
    Fetch all remaining rows from the current result set.

    Returns:
        A list of the remaining rows. Empty if no more rows are available.

    Examples:
        ```python
        cursor.execute('SELECT * FROM "warehouse"')
        rows = cursor.fetchall()
        # [(1, 'test')]
        ```
    """
    self._ensure_result_set()
    try:
        return self._cursor.fetchall()
    except psycopg.Error as error:
        self._notify_query_error(error)
        raise

`fetchmany(size=None)`

Fetch the next set of rows from the current result set.

Parameters:

Name	Type	Description	Default
`size`	`int \| None`	Maximum number of rows to fetch. If omitted, `arraysize` is used.	`None`

Returns:

Type	Description
`list[tuple[object, ...]]`	A list of the fetched rows. Empty if no more rows are available.

Examples:

Use the previously configured arraysize:

cursor.arraysize = 100
rows = cursor.fetchmany()

Override the batch size for a single call:

rows = cursor.fetchmany(50)

Source code in picopyn/synchronous/cursor.py

def fetchmany(self, size: int | None = None) -> list[tuple[object, ...]]:
    """
    Fetch the next set of rows from the current result set.

    Args:
        size: Maximum number of rows to fetch. If omitted, `arraysize` is used.

    Returns:
        A list of the fetched rows. Empty if no more rows are available.

    Examples:
        Use the previously configured `arraysize`:

        ```python
        cursor.arraysize = 100
        rows = cursor.fetchmany()
        ```

        Override the batch size for a single call:

        ```python
        rows = cursor.fetchmany(50)
        ```
    """
    self._ensure_result_set()

    try:
        if size is None:
            return self._cursor.fetchmany()
        return self._cursor.fetchmany(size)
    except psycopg.Error as error:
        self._notify_query_error(error)
        raise

`fetchone()`

Fetch the next row from the current result set.

Returns:

Type	Description
`tuple[object, ...] \| None`	The next row, or None if no more rows are available.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
row = cursor.fetchone()
# (1, 'test')

Source code in picopyn/synchronous/cursor.py

def fetchone(self) -> tuple[object, ...] | None:
    """
    Fetch the next row from the current result set.

    Returns:
        The next row, or None if no more rows are available.

    Examples:
        ```python
        cursor.execute('SELECT * FROM "warehouse"')
        row = cursor.fetchone()
        # (1, 'test')
        ```
    """
    self._ensure_result_set()
    try:
        return self._cursor.fetchone()
    except psycopg.Error as error:
        self._notify_query_error(error)
        raise

`Pool`

A connection pool.

Connection pool can be used to manage a set of connections to the database. Connections are taken from the pool inside a with block and returned to it when the block is left.

Note

When cluster discovery is enabled, pool addresses are obtained from the _pico_peer_address system table; DSN hosts are used only for initial topology discovery and as a fallback for technical connections.

Note

Pool contains up to max_size connections to Picodata instances for usage, but in addition can contain several technical connections hidden from the user. In particular, for topology tracking and metadata cache service. These services can ignore the user-connection pool filters: the bootstrap-mode dsn filter never applies to them (any cluster node may be picked, and a node the pool holds no connections to is preferred), while a forbidden-tier node is used only when no allowed-tier node is online. The instance_role filter never applies to technical connections, so a leader-only user pool may use replicas for them, and a replica-only user pool may use leaders.

---
title: Pool lifecycle
---

flowchart TD
%% === Style ===
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

%% === Data objects ===
POOL[[Connection pool]]

%% === Lifecycle ===
START([Start]):::neutral
DONE([Pool closed]):::neutral

CFG[/"Pool settings<br>· DSN<br>· pool size<br>· forbidden tiers<br>· cluster discovery mode"/]

START --> CFG --> CREATE["Pool.__init__"] --> CONNECT["pool.open()"] --> UsePool
CONNECT <-. write .-> POOL

CONNECT_DETAILS_NOTE>"Open algorithm details"]:::note
click CONNECT_DETAILS_NOTE "#picopyn.synchronous.Pool.open" "Open algorithm details"
CONNECT_DETAILS_NOTE -.-> CONNECT

RECONCILE_NOTE>"pool membership changes<br>in the background:<br>connections are opened/closed<br>to match topology changes<br>(discovery mode: eligible cluster nodes,<br>bootstrap mode: eligible DSN nodes)"]:::note
click RECONCILE_NOTE "#picopyn.synchronous.Pool.topology" "Reconcile algorithm details"
RECONCILE_NOTE -.-> POOL

CYCLE_NOTE>"repeated for each<br>database operation"]:::note
CYCLE_NOTE -.-> UsePool

subgraph UsePool["Pool usage"]
    direction TB
    ACQUIRE["with pool.connection() as conn"] --> CURSOR["conn.cursor() -> cursor"] --> USE["Execute queries<br>· cursor.fetchone()<br>· cursor.execute()<br>· ..."] --> RELEASE["leaving the with block<br>releases the connection"]
end

ACQUIRE <-. delete .-> POOL
RELEASE <-. write .-> POOL
CLOSE["pool.close()"] <-. delete .-> POOL
UsePool --> CLOSE
CLOSE --> DONE

Parameters:

Name	Type	Description	Default
`dsn`	`str`	The data source name (e.g., "postgresql://user:pass@host:port") for the cluster.	required
`max_size`	`int`	Maximum number of connections in the pool. See details.	`10`
`enable_discovery`	`bool`	If True, the pool will automatically discover available picodata instances. If False, only the given `dsn` will be used.	`False`
`balance_strategy`	`Callable[[list[Connection]], Connection] \| None`	A custom strategy function to select a connection from the pool. See details.	`None`
`forbidden_tiers`	`str \| None`	A comma-separated list of Picodata node tiers for which user connections are forbidden (e.g., "arbiter,readonly"). If None, connections to all tiers are allowed. Technical connections ignore this filter.	`None`
`instance_role`	`InstanceRole`	Replicaset role of the instances user connections may use; see InstanceRole. Bootstrap mode still considers only instances listed in `dsn`. If topology tracking is disabled, roles are evaluated only at startup. This setting is independent of Picodata READ_PREFERENCE and does not restrict technical connections. Automatic shard-aware routing is disabled for `"replica"`, because it routes to current leaders.	`'any'`
`topology_update_interval`	`float \| None`	Interval in seconds between topology refresh cycles. See details.	`60.0`
`query_metadata_cache_size`	`int \| None`	Maximum number of queries to keep distribution key metadata for. See details.	`100`
`rebalance_pool_divisor`	`int`	Limits how many connections a single rebalance loop may migrate. See details.	`10`
`routing_acquire_timeout`	`float`	Time to wait for a free bucket master connection during shard-aware routing. See details.	`0.0`
`instance_ejection_failure_threshold`	`int \| None`	Number of connection failures from one instance within `instance_ejection_failure_window` required to eject it from pool selection. See instance ejection.	`None`
`instance_ejection_failure_window`	`float \| None`	Sliding window in seconds used to count connection failures for instance ejection. See instance ejection.	`None`
`instance_ejection_duration`	`float \| None`	Number of seconds an ejected instance stays out of pool selection. See instance ejection.	`None`
`**connect_kwargs`	`Any`	Additional keyword arguments to pass to psycopg.connect() (e.g. the libpq SSL parameters). Unless `prepare_threshold` is passed here, pool connections are opened with `prepare_threshold=0`, so the driver prepares every statement on its first execution.	`{}`

Note

Settings exposed as writable properties can be changed after open(). Setters raise RuntimeError once the pool starts closing.

Examples:

Pool with custom balance strategy:

def random_strategy(connections):
    import random
    return random.choice(connections)

pool = Pool(
    dsn="postgresql://admin:pass@localhost:5432",
    balance_strategy=random_strategy,
    max_size=5,
)

Pool with multi-host DSN string. With bootstrap mode (default), the pool connects only to the listed nodes, distributing connections across them in round-robin:

pool = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    max_size=10,
)

Pool with multi-host DSN string. In discovery mode, the listed nodes are used for initial topology discovery and as a fallback for technical connections:

pool = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    max_size=10,
    enable_discovery=True,
)

SSL (kwargs-based):

pool = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    sslmode="verify-ca",
    sslrootcert="/path/to/ca.crt",
    sslcert="/path/to/client.crt",
    sslkey="/path/to/client.key",
)

SSL (DSN query params):

pool = Pool(
    dsn=(
        "postgresql://admin:pass@host1:5432,host2:5432/db"
        "?sslmode=verify-ca"
        "&sslrootcert=/path/to/ca.crt"
        "&sslcert=/path/to/client.crt"
        "&sslkey=/path/to/client.key"
    ),
)

Source code in picopyn/synchronous/pool.py

class Pool:
    """A connection pool.

    Connection pool can be used to manage a set of connections to the database.
    Connections are taken from the pool inside a `with` block and returned to it
    when the block is left.

    Note:
        When cluster discovery is enabled, pool addresses are obtained from the `_pico_peer_address`
        system table; DSN hosts are used only for initial topology discovery and as a fallback
        for technical connections.

    Note:
        Pool contains up to `max_size` connections to Picodata instances for usage, but in addition
        can contain several technical connections hidden from the user. In particular, for
        [topology tracking][picopyn.synchronous.topology_tracker.TopologyTracker] and
        [metadata cache service][picopyn.synchronous.pool.Pool.get_query_metadata]. These services
        can ignore the user-connection pool filters: the bootstrap-mode `dsn` filter never applies
        to them (any cluster node may be picked, and a node the pool holds no connections to is
        preferred), while a forbidden-tier node is used only when no allowed-tier node is online.
        The `instance_role` filter never applies to technical connections, so a leader-only user
        pool may use replicas for them, and a replica-only user pool may use leaders.

    --8<-- "_pool_lifecycle_sync.md"

    Args:
        dsn: The data source name (e.g., "postgresql://user:pass@host:port") for the cluster.
        max_size: Maximum number of connections in the pool. See
            [details][picopyn.synchronous.pool.Pool.max_size].
        enable_discovery: If True, the pool will automatically discover available picodata
            instances. If False, only the given `dsn` will be used.
        balance_strategy: A custom strategy function to select a connection from the pool. See
            [details][picopyn.synchronous.pool.Pool.balance_strategy].
        forbidden_tiers: A comma-separated list of Picodata node tiers for which user
            connections are forbidden (e.g., "arbiter,readonly"). If None, connections to
            all tiers are allowed. Technical connections ignore this filter.
        instance_role: Replicaset role of the instances user connections may use; see
            [InstanceRole][picopyn.instance_role.InstanceRole]. Bootstrap mode still considers
            only instances listed in `dsn`. If topology tracking is disabled, roles are
            evaluated only at startup. This setting is independent of Picodata
            [READ_PREFERENCE](https://docs.picodata.io/picodata/stable/reference/db_config/#read_preference)
            and does not restrict technical connections. Automatic
            [shard-aware routing][picopyn.synchronous.pool.Pool.execute] is disabled for
            `"replica"`, because it routes to current leaders.
        topology_update_interval: Interval in seconds between topology refresh cycles. See
            [details][picopyn.synchronous.pool.Pool.topology_update_interval].
        query_metadata_cache_size: Maximum number of queries to keep distribution key metadata
            for. See [details][picopyn.synchronous.pool.Pool.query_metadata_cache_size].
        rebalance_pool_divisor: Limits how many connections a single rebalance loop may migrate.
            See [details][picopyn.synchronous.pool.Pool.rebalance_pool_divisor].
        routing_acquire_timeout: Time to wait for a free bucket master connection during
            shard-aware routing. See
            [details][picopyn.synchronous.pool.Pool.routing_acquire_timeout].
        instance_ejection_failure_threshold: Number of connection failures from one
            instance within `instance_ejection_failure_window` required to eject it from
            pool selection. See
            [instance ejection][picopyn.synchronous.pool.Pool.instance_ejection_enabled].
        instance_ejection_failure_window: Sliding window in seconds used to count
            connection failures for instance ejection. See
            [instance ejection][picopyn.synchronous.pool.Pool.instance_ejection_enabled].
        instance_ejection_duration: Number of seconds an ejected instance stays out
            of pool selection. See
            [instance ejection][picopyn.synchronous.pool.Pool.instance_ejection_enabled].
        **connect_kwargs: Additional keyword arguments to pass to psycopg.connect()
            (e.g. the libpq SSL parameters). Unless `prepare_threshold` is passed here,
            pool connections are opened with `prepare_threshold=0`, so the driver prepares
            every statement on its first execution.

    Note:
        Settings exposed as writable properties can be changed after open().
        Setters raise RuntimeError once the pool starts closing.

    Examples:
        Pool with custom balance strategy:

        ```python
        def random_strategy(connections):
            import random
            return random.choice(connections)

        pool = Pool(
            dsn="postgresql://admin:pass@localhost:5432",
            balance_strategy=random_strategy,
            max_size=5,
        )
        ```

        Pool with multi-host DSN string. With bootstrap mode (default), the pool connects only
        to the listed nodes, distributing connections across them in round-robin:

        ```python
        pool = Pool(
            dsn="postgresql://admin:pass@host1:5432,host2:5432",
            max_size=10,
        )
        ```

        Pool with multi-host DSN string. In discovery mode, the listed nodes are used for initial
        topology discovery and as a fallback for technical connections:

        ```python
        pool = Pool(
            dsn="postgresql://admin:pass@host1:5432,host2:5432",
            max_size=10,
            enable_discovery=True,
        )
        ```

        SSL (kwargs-based):

        ```python
        pool = Pool(
            dsn="postgresql://admin:pass@host1:5432,host2:5432",
            sslmode="verify-ca",
            sslrootcert="/path/to/ca.crt",
            sslcert="/path/to/client.crt",
            sslkey="/path/to/client.key",
        )
        ```

        SSL (DSN query params):

        ```python
        pool = Pool(
            dsn=(
                "postgresql://admin:pass@host1:5432,host2:5432/db"
                "?sslmode=verify-ca"
                "&sslrootcert=/path/to/ca.crt"
                "&sslcert=/path/to/client.crt"
                "&sslkey=/path/to/client.key"
            ),
        )
        ```
    """

    def __init__(
        self,
        dsn: str,
        max_size: int = 10,
        enable_discovery: bool = False,
        balance_strategy: Callable[[list[Connection]], Connection] | None = None,
        forbidden_tiers: str | None = None,
        topology_update_interval: float | None = 60.0,
        query_metadata_cache_size: int | None = 100,
        rebalance_pool_divisor: int = 10,
        routing_acquire_timeout: float = 0.0,
        instance_ejection_failure_threshold: int | None = None,
        instance_ejection_failure_window: float | None = None,
        instance_ejection_duration: float | None = None,
        instance_role: InstanceRole = "any",
        **connect_kwargs: Any,
    ) -> None:
        if max_size < 1:
            raise ValueError("max_size must be at least 1")
        if routing_acquire_timeout < 0:
            raise ValueError("routing_acquire_timeout cannot be negative")
        if rebalance_pool_divisor < 1:
            raise ValueError("rebalance_pool_divisor must be at least 1")
        if topology_update_interval is not None and topology_update_interval < 1.0:
            raise ValueError("topology_update_interval (if set) cannot be less than 1s")
        if query_metadata_cache_size is not None and query_metadata_cache_size < 1:
            raise ValueError("query_metadata_cache_size (if set) cannot be less than 1")
        instance_ejection_options = (
            instance_ejection_failure_threshold,
            instance_ejection_failure_window,
            instance_ejection_duration,
        )
        if any(value is not None for value in instance_ejection_options) and not all(
            value is not None for value in instance_ejection_options
        ):
            raise ValueError(
                "instance ejection failure threshold, window, and duration must be set together"
            )

        if instance_role not in ("any", "leader", "replica"):
            raise ValueError("instance_role must be one of: 'any', 'leader', 'replica'")

        # origin DSNs to connect
        self._connect_dsns: list[str] = _parse_multihost_dsn(dsn)
        # DSN with hidden password
        self._redacted_dsn = _redact_dsn(dsn)
        self._connect_kwargs = connect_kwargs
        self._pool_connect_kwargs = {"prepare_threshold": 0, **connect_kwargs}
        self._max_size = max_size
        self._rebalance_pool_divisor = rebalance_pool_divisor
        self._routing_acquire_timeout = routing_acquire_timeout
        self._instance_role = instance_role
        # query meta discovery is on only if configured; the service itself
        # (technical connection + cache) is created in open()
        self._query_metadata_cache_size = query_metadata_cache_size
        self._query_metadata_service: QueryMetadataService | None = None
        if all(value is not None for value in instance_ejection_options):
            assert instance_ejection_failure_threshold is not None
            assert instance_ejection_failure_window is not None
            assert instance_ejection_duration is not None
            self._instance_ejection_state: InstanceEjectionState | None = InstanceEjectionState(
                instance_ejection_failure_threshold,
                instance_ejection_failure_window,
                instance_ejection_duration,
            )
        else:
            self._instance_ejection_state = None
        self._pool: deque[Connection] = deque()
        self._used: set[Connection] = set()
        self._instance_uuid_by_address: dict[str, str] = {}
        self._forbidden_tiers = (
            set(t.strip() for t in forbidden_tiers.split(",") if t.strip())
            if forbidden_tiers
            else set()
        )
        self._condition = Condition()
        # pool shutdown state, guarded by `_condition`
        # None -> open; unset -> closing; set -> closed
        self._close_event: Event | None = None
        # serializes the shutdown work itself, so `_close_event` is only set once every
        # connection the shutdown owns is actually closed. Never taken while holding
        # `_condition`: the order is always `_shutdown_lock` -> `_condition`
        self._shutdown_lock = Lock()
        self._default_acquire_timeout_sec = 5
        # topology tracking is on if configured, and it always changes pool membership
        # (see _reconcile_pool_with_topology), wired up as the topology tracker's
        # refresh callback below. `enable_discovery` only decides which nodes are
        # candidates: the whole cluster or the DSN nodes only
        # (see _collect_candidate_uuids)
        self._topology_update_interval = topology_update_interval
        self._topology_tracker: TopologyTracker | None = None
        self._topology_disable: Future[None] | None = None
        self._metadata_disable: Future[None] | None = None
        # node discovery mode
        # if disabled, pool will be filled with given address connections
        # if enabled, pool will be filled with available picodata instances
        self._enable_discovery = enable_discovery
        # load balancing strategy:
        # if None, a simple round-robin strategy will be used.
        # otherwise, the provided callable will be used to select connections.
        if balance_strategy is not None and not callable(balance_strategy):
            raise ValueError("balance_strategy must be callable or None")
        self._balance_strategy = balance_strategy

    @property
    def routing_acquire_timeout(self) -> float:
        """Time in seconds that
        [shard-aware routing][picopyn.synchronous.pool.Pool.execute] waits for a free
        connection to the bucket master before falling back to an ordinary pool connection.
        `0` means a single attempt with no waiting. The value cannot be negative.

        The timeout is only spent when the pool does hold a connection to the bucket
        master and every such connection is busy. If the pool has no connection to that
        node at all, routing falls back to an ordinary pool connection at once: only a
        topology reconciliation can add one, and it runs on the topology refresh
        interval, not within the acquire timeout.

        Choose the value based on your workload. If the pool is overloaded and queries
        frequently wait for a free connection, set this value to `0` to avoid additional
        waiting. If queries rarely wait for a free connection, consider setting it to a
        positive value appropriate for your setup. Keep in mind that, unlike the async
        pool, waiting here occupies the calling thread.

        Can be changed: a new value applies to new routing attempts. An acquire already
        waiting keeps its original timeout.

        Raises:
            ValueError: If the new value is negative.
            RuntimeError: If the pool is closing or closed.
        """
        return self._routing_acquire_timeout

    @routing_acquire_timeout.setter
    def routing_acquire_timeout(self, value: float) -> None:
        with self._condition:
            self._check_open()
            if value < 0:
                raise ValueError("routing_acquire_timeout cannot be negative")
            self._routing_acquire_timeout = value

    @property
    def instance_ejection_enabled(self) -> bool:
        """Whether error-based instance ejection is enabled.

        Instance ejection temporarily excludes an instance from pool selection after repeated
        connection failures. It is enabled only if all three `instance_ejection_*` settings
        are passed to the constructor.

        When an instance reaches
        [instance_ejection_failure_threshold][picopyn.synchronous.pool.Pool.instance_ejection_failure_threshold]
        connection failures within
        [instance_ejection_failure_window][picopyn.synchronous.pool.Pool.instance_ejection_failure_window] seconds,
        the instance is excluded from pool selection for
        [instance_ejection_duration][picopyn.synchronous.pool.Pool.instance_ejection_duration] seconds. Its
        connections stay open, but the pool stops handing them out (see the exceptions below).

        Connection failures are failed or timed-out connection attempts and connections
        lost while in use; SQL errors are not counted.

        Instance ejection **cannot be enabled after construction**. Use
        [disable_instance_ejection()][picopyn.synchronous.pool.Pool.disable_instance_ejection] to disable it.

        --8<-- "docs/instance_ejection_selection.md"
        """
        return self._instance_ejection_state is not None

    @property
    def instance_ejection_failure_threshold(self) -> int | None:
        """Number of connection failures from one instance required for ejection.

        `None` if instance ejection is disabled.

        While instance ejection is [enabled][picopyn.synchronous.pool.Pool.instance_ejection_enabled],
        a new value applies to the next recorded failure; an ejection that is already active
        keeps its original deadline.

        Raises:
            ValueError: If the new value is not a positive integer.
            RuntimeError: If the pool is closing or closed, or instance ejection
                is disabled.
        """
        state = self._instance_ejection_state
        return state.failure_threshold if state is not None else None

    @instance_ejection_failure_threshold.setter
    def instance_ejection_failure_threshold(self, value: int) -> None:
        with self._condition:
            self._check_open()
            state = self._instance_ejection_state
            if state is None:
                raise RuntimeError("instance ejection is disabled")
            state.failure_threshold = value

    @property
    def instance_ejection_failure_window(self) -> float | None:
        """Sliding window in seconds used to count failures for instance ejection.

        `None` if instance ejection is disabled.

        While instance ejection is [enabled][picopyn.synchronous.pool.Pool.instance_ejection_enabled],
        a new value applies to the next recorded failure; an ejection that is already active
        keeps its original deadline.

        Raises:
            ValueError: If the new value is not a positive finite number.
            RuntimeError: If the pool is closing or closed, or instance ejection
                is disabled.
        """
        state = self._instance_ejection_state
        return state.failure_window if state is not None else None

    @instance_ejection_failure_window.setter
    def instance_ejection_failure_window(self, value: float) -> None:
        with self._condition:
            self._check_open()
            state = self._instance_ejection_state
            if state is None:
                raise RuntimeError("instance ejection is disabled")
            state.failure_window = value

    @property
    def instance_ejection_duration(self) -> float | None:
        """Number of seconds an ejected instance stays out of pool selection.

        `None` if instance ejection is disabled.

        While instance ejection is [enabled][picopyn.synchronous.pool.Pool.instance_ejection_enabled],
        a new value applies to ejections started after the change; an ejection that is already active
        keeps its original deadline.

        Raises:
            ValueError: If the new value is not a positive finite number.
            RuntimeError: If the pool is closing or closed, or instance ejection
                is disabled.
        """
        state = self._instance_ejection_state
        return state.ejection_duration if state is not None else None

    @instance_ejection_duration.setter
    def instance_ejection_duration(self, value: float) -> None:
        with self._condition:
            self._check_open()
            state = self._instance_ejection_state
            if state is None:
                raise RuntimeError("instance ejection is disabled")
            state.ejection_duration = value

    def disable_instance_ejection(self) -> None:
        """Disable error-based instance ejection.

        Existing ejections and recorded failures are discarded. Connections stay
        open and immediately return to pool selection. The operation is
        idempotent. See
        [instance ejection][picopyn.synchronous.pool.Pool.instance_ejection_enabled].

        Raises:
            RuntimeError: If the pool is closing or closed.
        """
        with self._condition:
            self._check_open()
            if self._instance_ejection_state is None:
                return
            self._instance_ejection_state = None
            self._condition.notify_all()

    @property
    def rebalance_pool_divisor(self) -> int:
        """How much of the pool a single
        [rebalance][picopyn.synchronous.Pool.topology--rebalance] loop may rebalance:
        at most `max(1, max_size // rebalance_pool_divisor)` connections are migrated
        from overloaded nodes to underloaded ones per reconcile, so even a pool
        smaller than the divisor still migrates one connection per refresh.
        A smaller value converges faster at the cost of more reconnects per refresh.
        Must be at least 1.

        Can be changed: a new value applies to the next rebalance loop.

        Raises:
            ValueError: If the new value is less than 1.
            RuntimeError: If the pool is closing or closed.
        """
        return self._rebalance_pool_divisor

    @rebalance_pool_divisor.setter
    def rebalance_pool_divisor(self, value: int) -> None:
        with self._condition:
            self._check_open()
            if value < 1:
                raise ValueError("rebalance_pool_divisor must be at least 1")
            self._rebalance_pool_divisor = value

    @property
    def balance_strategy(self) -> Callable[[list[Connection]], Connection] | None:
        """A custom strategy function to select a connection from the pool.
        If None, round-robin strategy is used.

        Can be changed: a new value applies to the next ordinary acquire.

        Raises:
            ValueError: If the new value is neither callable nor `None`.
            RuntimeError: If the pool is closing or closed.
        """
        return self._balance_strategy

    @balance_strategy.setter
    def balance_strategy(self, value: Callable[[list[Connection]], Connection] | None) -> None:
        with self._condition:
            self._check_open()
            if value is not None and not callable(value):
                raise ValueError("balance_strategy must be callable or None")
            self._balance_strategy = value

    @property
    def max_size(self) -> int:
        """Maximum number of connections in the pool.

        Can be changed before or after the pool is opened:

        - **increasing the limit** *requires topology tracking* and triggers a topology refresh
            to refill the pool. The setter does not wait for new connections.
        - **decreasing the limit** closes excess connections as they are released. *With
            topology tracking* it also triggers a topology refresh without waiting for it:
            the refresh closes excess free connections and marks excess in-use ones to close
            on release, starting from the most loaded nodes. *Without topology tracking* free
            connections stay until acquired and released. In-use connections stay until
            released in any case, so *the actual size can temporarily exceed this limit*.

        Raises:
            ValueError: If the new value is less than 1.
            RuntimeError: If the pool is closing or closed, or if the value is
                increased while topology tracking is disabled.
        """
        return self._max_size

    @max_size.setter
    def max_size(self, value: int) -> None:
        with self._condition:
            self._check_open()
            if value < 1:
                raise ValueError("max_size must be at least 1")
            if value == self._max_size:
                return
            if value > self._max_size and self._topology_update_interval is None:
                raise RuntimeError("Cannot increase max_size while topology tracking is disabled")
            self._max_size = value
            if self._topology_update_interval is not None and self._topology_tracker is not None:
                self._topology_tracker.trigger()

    @property
    def topology_update_interval(self) -> float | None:
        """Interval in seconds between topology refresh cycles for
        [Topology Tracker][picopyn.synchronous.topology_tracker.TopologyTracker].
        Pass None at initialization to disable periodic updates. With updates disabled
        [topology][picopyn.synchronous.pool.Pool.topology] is None and reconcile
        never runs.

        A new interval must be at least 1s. After open(), changing it triggers
        an immediate topology refresh, so the new interval applies without
        waiting for the previous timeout. A refresh already in progress is
        allowed to finish. Before open(), the value is saved for startup.

        Use [disable_topology_tracking()][picopyn.synchronous.pool.Pool.disable_topology_tracking]
        to stop it. Assigning the current value is a no-op.

        Raises:
            ValueError: If the new interval is less than 1s.
            RuntimeError: If the pool is closing or closed, or if the new value
                enables or disables topology tracking.
        """
        return self._topology_update_interval

    @topology_update_interval.setter
    def topology_update_interval(self, value: float | None) -> None:
        with self._condition:
            self._check_open()
            if value == self._topology_update_interval:
                return
            if value is None or self._topology_update_interval is None:
                raise RuntimeError(
                    "Changing topology tracking between enabled and disabled is not supported "
                    "by this property; use disable_topology_tracking() to disable it"
                )
            if value < 1.0:
                raise ValueError("topology_update_interval (if set) cannot be less than 1s")
            if self._topology_tracker is not None:
                self._topology_tracker.update_interval(value)
            self._topology_update_interval = value

    @property
    def query_metadata_cache_size(self) -> int | None:
        """Maximum number of distinct queries to keep Picodata's
        distribution key metadata for, on the pool's dedicated metadata-discovery connection
        (see [query meta usage][picopyn.synchronous.pool.Pool.get_query_metadata]).
        Oldest entries are evicted once exceeded. Must be at least 1 if set.
        Pass None at initialization to disable the metadata service: its technical connection is never
        opened and [get_query_metadata][picopyn.synchronous.pool.Pool.get_query_metadata] raises
        RuntimeError.

        Shrinking immediately evicts excess LRU entries, including entries for
        queries without metadata. The pending prepare limit changes too;
        already queued prepares are allowed to finish.
        Before open(), the value is saved for startup.

        Use [disable_query_metadata()][picopyn.synchronous.pool.Pool.disable_query_metadata]
        to stop it. Assigning the current value is a no-op.

        Raises:
            ValueError: If the new cache size is less than 1.
            RuntimeError: If the pool is closing or closed, or if the new value
                enables or disables the metadata service.
        """
        return self._query_metadata_cache_size

    @query_metadata_cache_size.setter
    def query_metadata_cache_size(self, value: int | None) -> None:
        with self._condition:
            self._check_open()
            if value == self._query_metadata_cache_size:
                return
            if value is None or self._query_metadata_cache_size is None:
                raise RuntimeError(
                    "Changing the metadata service between enabled and disabled is not supported "
                    "by this property; use disable_query_metadata() to disable it"
                )
            if value < 1:
                raise ValueError("query_metadata_cache_size (if set) cannot be less than 1")
            if self._query_metadata_service is not None:
                self._query_metadata_service.resize(value)
            self._query_metadata_cache_size = value

    def disable_topology_tracking(self) -> None:
        """Disable topology tracking and wait for its technical connection to close.

        Sets `topology_update_interval` and `topology` to None. New refreshes,
        reconcile and topology-based routing stop; user connections are kept.
        Previously scheduled evictions still take effect on release.

        Can be called before open(). Repeated calls wait for the same shutdown.
        This call blocks until the running refresh and source close finish.

        Raises:
            RuntimeError: If the pool is closing or closed.
        """
        with self._condition:
            self._check_open()
            if self._topology_update_interval is None:
                shutdown = self._topology_disable
                if shutdown is None:
                    return
                first_call = False
            else:
                self._topology_update_interval = None
                tracker = self._topology_tracker
                if tracker is None:
                    return
                shutdown = Future()
                self._topology_disable = shutdown
                first_call = True
        assert shutdown is not None
        if first_call:
            assert tracker is not None
            try:
                self._stop_topology_tracking(tracker)
            except BaseException as e:
                shutdown.set_exception(e)
            else:
                shutdown.set_result(None)
        try:
            shutdown.result()
        finally:
            with self._condition:
                if self._topology_disable is shutdown and shutdown.done():
                    self._topology_disable = None

    def _stop_topology_tracking(self, tracker: TopologyTracker) -> None:
        try:
            tracker.terminate()
            tracker.stop(timeout=None)
        except BaseException:
            tracker.terminate()
            with self._condition:
                if self._topology_tracker is tracker:
                    self._topology_tracker = None
            raise
        else:
            with self._condition:
                if self._topology_tracker is tracker:
                    self._topology_tracker = None

    def disable_query_metadata(self) -> None:
        """Disable query metadata and wait for its technical connection to close.

        Sets `query_metadata_cache_size` to None and cancels pending metadata
        requests. New execute() calls use ordinary pool selection, and
        get_query_metadata() raises RuntimeError. Explicit routing by sharding
        key remains available while topology tracking is enabled.

        Can be called before open(). Repeated calls wait for the same shutdown.
        This call blocks until the running prepare and connection close finish.

        Raises:
            RuntimeError: If the pool is closing or closed.
        """
        with self._condition:
            self._check_open()
            if self._query_metadata_cache_size is None:
                shutdown = self._metadata_disable
                if shutdown is None:
                    return
                first_call = False
            else:
                self._query_metadata_cache_size = None
                service = self._query_metadata_service
                if service is None:
                    return
                shutdown = Future()
                self._metadata_disable = shutdown
                first_call = True
        assert shutdown is not None
        if first_call:
            assert service is not None
            try:
                self._stop_query_metadata(service)
            except BaseException as e:
                shutdown.set_exception(e)
            else:
                shutdown.set_result(None)
        try:
            shutdown.result()
        finally:
            with self._condition:
                if self._metadata_disable is shutdown and shutdown.done():
                    self._metadata_disable = None

    def _stop_query_metadata(self, service: QueryMetadataService) -> None:
        try:
            service.stop()
        except BaseException:
            service.terminate()
            with self._condition:
                if self._query_metadata_service is service:
                    self._query_metadata_service = None
            raise
        else:
            with self._condition:
                if self._query_metadata_service is service:
                    self._query_metadata_service = None

    def __repr__(self) -> str:
        # Explicit repr backed by the already-redacted `_redacted_dsn`, so printing/logging
        # a Pool can never expose the passwords held in `_connect_dsns`.
        return f"Pool(dsn={self._redacted_dsn!r})"

    def _current_size(self) -> int:
        """Total number of connections owned by the pool for user-usage"""
        return len(self._pool) + len(self._used)

    def _check_open(self) -> None:
        """Raise if the pool is closing or already closed. Call it under `_condition`."""
        if self._close_event is None:
            return
        if self._close_event.is_set():
            raise RuntimeError("Connection pool is closed")
        raise RuntimeError("Connection pool is closing")

    def open(self) -> None:
        """
        Open the pool by creating up to `max_size` connections.

        This should be called before using the pool to ensure connections are available.

        --8<-- "_pool_init.md"

        <details id="pool-init-discovery-enabled"><summary>See open details in discovery mode</summary>
        --8<-- "_pool_init_discovery_enabled.md"
        </details>

        <details id="pool-init-discovery-disabled"><summary>See open details in bootstrap mode</summary>
        --8<-- "_pool_init_discovery_disabled.md"
        </details>

        Raises:
            RuntimeError: If the pool is closing or already closed, if discovery fails,
                or if fewer than `max_size` connections could be established.
        """
        with self._condition:
            self._check_open()

            if self._current_size() >= self._max_size:
                return

            # if node discovery is enabled, then connect to all alive picodata instances
            # (if they fit within the max_size limit)
            if self._enable_discovery:
                self._fill_pool_from_discovery()
            else:
                self._fill_pool_from_bootstrap_dsns()

            conn_count = self._current_size()
            if conn_count < self._max_size:
                while self._pool:
                    conn = self._pool.popleft()
                    try:
                        conn.close()
                    except Exception as e:
                        logger.warning("Pool connect cleanup: could not close connection: %s", e)
                raise RuntimeError(
                    f"Failed to initialize connection pool: only {conn_count} "
                    f"out of {self._max_size} connections established for DSN "
                    f"{self._redacted_dsn}"
                )

            # rotate the pool to randomize the order of connections.
            # this helps to distribute the initial load more evenly across nodes
            # when using round-robin or when multiple clients start simultaneously.
            shift = random.randint(0, len(self._pool) - 1)
            self._pool.rotate(shift)

            logger.info("Pool initialized with %d connections", conn_count)
            if self._topology_update_interval is not None and self._topology_tracker is None:
                self._topology_tracker = TopologyTracker(
                    PollingTopologySource(
                        lambda topology: self._get_technical_connection(topology=topology)
                    ),
                    self._topology_update_interval,
                )
                # register the pool's reconcile callback so connections are updated
                # to match the topology after each refresh
                self._topology_tracker.start(on_refresh=self._reconcile_pool_with_topology)
                logger.debug(
                    "Pool topology tracker started with update interval %ss",
                    self._topology_update_interval,
                )

            if self._query_metadata_cache_size is not None and self._query_metadata_service is None:
                # technical connection + cache for query meta discovery, decoupled from the
                # connections below that actually run queries
                self._query_metadata_service = QueryMetadataService(
                    connect=lambda on_query_metadata: self._get_technical_connection(
                        on_query_metadata
                    ),
                    cache_size=self._query_metadata_cache_size,
                )

                self._query_metadata_service.start()
                logger.debug(
                    "Pool query-metadata service started with cache size %s",
                    self._query_metadata_cache_size,
                )
                if self._instance_role == "replica" and self._topology_update_interval is not None:
                    logger.warning(
                        "Automatic shard-aware routing is disabled for instance_role='replica': "
                        "current replicaset masters are excluded from user pool connections"
                    )
        return

    def _fill_pool_from_discovery(self) -> None:
        """
        Fill the pool with online nodes from Picodata cluster if they pass filters.
        Current filters are `forbidden_tiers` and `instance_role`.
        """
        try:
            instance_addrs = self._discover_instances()
        except Exception as e:
            raise RuntimeError(
                f"Failed to discover instances using DSN {self._redacted_dsn}: {e}"
            ) from e

        # fill the connection pool with connections to all available nodes, up to the max_size.
        # this ensures the pool is evenly populated across all nodes.
        # if a node fails to connect, it is dropped from the rotation.
        # the loop exits early if no nodes remain to avoid an infinite loop.
        addrs = deque(instance_addrs)
        while self._current_size() < self._max_size and addrs:
            address = addrs[0]
            dsn = _replace_dsn_host(self._connect_dsns[0], address)

            logger.debug("Discovery: connecting to node %s", address)
            instance_uuid = self._instance_uuid_by_address.get(address)
            try:
                conn = Connection.connect(
                    dsn,
                    **self._pool_connect_kwargs,
                )
                conn._on_query_error = self._on_query_error
                conn.instance_uuid = instance_uuid
                self._pool.append(conn)
                logger.debug("Discovery: connected to node %s", address)
            except Exception as e:
                self._record_connection_failure(instance_uuid, e, during_connect=True)
                logger.warning("Discovery: could not connect to node %s for pool: %s", address, e)
                addrs.popleft()
                continue

            addrs.rotate(-1)

    def _fill_pool_from_bootstrap_dsns(self) -> None:
        """
        Fill the pool with online nodes from DSNs if they pass filters.

        Current filters are `forbidden_tiers` and `instance_role`. Nodes that is not online are skipped.
        """
        available = deque(self._connect_dsns)
        instance_info_by_address: dict[str, InstanceInfo] | None = None
        while self._current_size() < self._max_size and available:
            candidate = available[0]
            address = _dsn_hostinfo(candidate)
            conn: Connection | None = None
            instance_uuid: str | None = None
            try:
                if instance_info_by_address is not None:
                    instance_info = instance_info_by_address.get(address)
                    if not self._is_accepted_bootstrap_candidate(instance_info, address):
                        available.popleft()
                        continue
                    _, instance_uuid, _ = instance_info

                logger.debug("Bootstrap by DSN: connecting to %s", address)
                conn = Connection.connect(
                    candidate,
                    **self._pool_connect_kwargs,
                )
                conn._on_query_error = self._on_query_error
                conn.instance_uuid = instance_uuid

                if instance_info_by_address is None:
                    instance_info_by_address = self._fetch_bootstrap_instance_info(conn)
                    instance_info = instance_info_by_address.get(address)
                    if not self._is_accepted_bootstrap_candidate(instance_info, address):
                        conn.close()
                        available.popleft()
                        continue
                    _, instance_uuid, _ = instance_info

                conn.instance_uuid = instance_uuid
                self._pool.append(conn)
                available.rotate(-1)
            except Exception as e:
                self._record_connection_failure(instance_uuid, e, during_connect=True)
                if conn is not None:
                    try:
                        conn.close()
                    except Exception as close_error:
                        logger.warning(
                            "Bootstrap by DSN: could not close connection to %s after failure: %s",
                            _dsn_hostinfo(candidate),
                            close_error,
                        )
                logger.warning(
                    "Bootstrap by DSN: could not connect to %s: %s",
                    _dsn_hostinfo(candidate),
                    e,
                )
                available.popleft()

    def _is_accepted_bootstrap_candidate(
        self,
        instance_info: InstanceInfo | None,
        address: str,
    ) -> TypeGuard[tuple[str | None, str, str | None]]:
        return is_accepted_bootstrap_candidate(
            instance_info,
            address,
            self._forbidden_tiers,
            self._instance_role,
        )

    def _pick_refill_candidate(self, candidate_list: list[str]) -> str | None:
        """Pick the candidate node with the fewest current connections.

        See [pick_refill_candidate][picopyn.utils.balance.pick_refill_candidate].
        """
        with self._condition:
            return pick_refill_candidate(candidate_list, (*self._pool, *self._used))

    def _pick_rebalance_pair(self, candidate_list: list[str]) -> tuple[str, str] | None:
        """Pick `(overloaded_uuid, underloaded_uuid)` if the pool leans on some node.

        See [pick_rebalance_pair][picopyn.utils.balance.pick_rebalance_pair].
        """
        with self._condition:
            return pick_rebalance_pair(candidate_list, (*self._pool, *self._used))

    def _collect_candidate_uuids(self, topology: Topology) -> list[str]:
        """Collect uuids of the instances the pool is allowed to connect to.

        In bootstrap mode the candidates are limited to the `dsn` nodes.
        See [collect_candidate_uuids][picopyn.utils.balance.collect_candidate_uuids].
        """
        dsn_addresses = (
            None
            if self._enable_discovery
            else {_dsn_hostinfo(connect_dsn) for connect_dsn in self._connect_dsns}
        )
        return collect_candidate_uuids(
            topology,
            self._forbidden_tiers,
            dsn_addresses,
            self._instance_role,
        )

    def _technical_candidates(self, topology: Topology) -> list[tuple[str, str]]:
        """Collect candidates for a technical connection as a list of `(instance_uuid, address)`.

        See [technical_candidates][picopyn.utils.balance.technical_candidates].
        """
        with self._condition:
            return technical_candidates(topology, self._forbidden_tiers, (*self._pool, *self._used))

    def _get_technical_connection(
        self,
        on_query_metadata: Callable[[PreparedStatementMetadata], None] | None = None,
        *,
        topology: Topology | None = None,
    ) -> Connection | None:
        """Open a technical connection (topology tracking, query metadata etc) to the least
        loaded node the pool knows about.

        Technical connections are not part of user-usage connections (free or used), so they do
        not affect the connection counts used to pick a node.

        The node to connect to is picked from the given topology snapshot, or from the
        current pool topology if no snapshot is given:

        - pick the least loaded online allowed-tier node;
        - if there are none, pick the least loaded forbidden-tier node -- otherwise, once every
        allowed tier went offline, the tracker would have nowhere to connect and the pool could
        never learn that those tiers are back;
        - if there is no topology yet, fall back to the DSN list.

        Note:
            DSN list is also the only option before the very first topology refresh, when there is
            no snapshot to choose from yet.
        """
        topology_snapshot = topology if topology is not None else self.topology
        candidates = (
            self._technical_candidates(topology_snapshot) if topology_snapshot is not None else []
        )
        # we do not need to collect or recount candidates several times:
        # the technical connection is established only once (until it is closed),
        # so the connection counts do not change while we try to establish it
        for instance_uuid, address in candidates:
            dsn = _replace_dsn_host(self._connect_dsns[0], address)
            try:
                conn = Connection.connect(
                    dsn, on_query_metadata=on_query_metadata, **self._connect_kwargs
                )
            except Exception as e:
                logger.debug("Technical connection: could not connect to %s: %s", address, e)
                continue

            conn.instance_uuid = instance_uuid
            logger.debug("Technical connection: connected to %s", address)
            return conn

        if candidates:
            logger.warning(
                "Technical connection: none of %d topology candidate(s) reachable, "
                "falling back to DSN",
                len(candidates),
            )

        return self._get_dsn_connection(on_query_metadata)

    def _insert_randomly(self, conn: Connection) -> None:
        """Insert connection into the pool at a random position.

        See [insert_randomly][picopyn.utils.balance.insert_randomly].
        """
        insert_randomly(self._pool, conn)

    def _swap_connection(
        self, old_conn: Connection, new_node_uuid: str, new_node_address: str
    ) -> bool:
        """Open a connection to given address and put it into the pool in place of the old one.

        It is supposed that old connection was already taken out of pool: this method owns
        it and either hands it back to the pool or gets rid of it. The replacement is
        opened before old connection is closed, so while it is being opened the pool is one
        connection short.

        Returns:
            True if the migration went through, False if the replacement could not be
            opened -- old connection is back in the pool then, or closed if the pool
            is already closed.
        """
        dsn = _replace_dsn_host(self._connect_dsns[0], new_node_address)
        migrated = False
        connections_to_close: list[tuple[Connection, str]] = []

        try:
            new_conn = Connection.connect(
                dsn,
                **self._pool_connect_kwargs,
            )
            new_conn._on_query_error = self._on_query_error
        except Exception as e:
            self._record_connection_failure(new_node_uuid, e, during_connect=True)
            logger.warning("Rebalance: could not migrate connection to %s: %s", new_node_address, e)

            with self._condition:
                restore_old = self._close_event is None and self._current_size() < self._max_size
                if restore_old:
                    self._insert_randomly(old_conn)
                    self._condition.notify_all()
                else:
                    connections_to_close.append(
                        (old_conn, "Rebalance: could not drop the old connection: %s")
                    )
        else:
            new_conn.instance_uuid = new_node_uuid

            with self._condition:
                if (
                    self._close_event is None
                    and self._topology_update_interval is not None
                    and self._current_size() < self._max_size
                ):
                    self._insert_randomly(new_conn)
                    self._condition.notify_all()
                    migrated = True
                    connections_to_close.append(
                        (old_conn, "Rebalance: could not drop the old connection: %s")
                    )
                else:
                    restore_old = (
                        self._close_event is None and self._current_size() < self._max_size
                    )
                    if restore_old:
                        self._insert_randomly(old_conn)
                        self._condition.notify_all()
                    else:
                        connections_to_close.append(
                            (old_conn, "Rebalance: could not drop the old connection: %s")
                        )

                    connections_to_close.append(
                        (new_conn, "Rebalance: could not drop the replacement: %s")
                    )

        for conn, error_message in connections_to_close:
            try:
                conn.close()
            except Exception as e:
                logger.warning(error_message, e)

        return migrated

    def _close_excess_connections(self) -> None:
        """Trim connections above `max_size` (left after the limit was decreased).

        Free connections are closed right away, in-use ones are marked to evict.

        See [pick_excess_connections][picopyn.utils.balance.pick_excess_connections].
        """
        with self._condition:
            try:
                self._check_open()
            except RuntimeError:
                return
            excess_conns, used_to_evict = pick_excess_connections(
                self._pool, self._used, self._max_size
            )
            for conn in excess_conns:
                self._pool.remove(conn)
            for used_conn in used_to_evict:
                used_conn.to_evict = True

        # close excess connections outside the lock --
        # this must not block other threads waiting to acquire/release
        closed = 0
        for conn in excess_conns:
            try:
                conn.close()
                closed += 1
            except Exception as e:
                logger.warning("Rebalance: could not close excess connection: %s", e)

        if excess_conns:
            logger.debug(
                "Rebalance: found %s excess free connection(s), closed %s",
                len(excess_conns),
                closed,
            )

    def _rebalance_pool(self, topology: Topology, candidate_list: list[str]) -> None:
        """Migrate connections from overloaded nodes to underloaded ones.

        Before migrating, connections above `max_size` (left after the limit was
        decreased) are trimmed from the most loaded nodes first: free ones are closed,
        in-use ones are marked `to_evict`.

        One migration moves one connection from the most overloaded node to the most
        underloaded one. The balanced distribution is `max_size / node count` connections
        per each node, so a node takes as many migrations as needed to reach it.

        One rebalance loop does not do all the migrations at once. It stops after some
        limited migration attempts, where limit is
        `max(1, pool.max_size // pool.rebalance_pool_divisor)`, and leaves the rest to
        the next rebalance, so that a topology event does not trigger too many
        reconnections at once. A failed attempt (the node has no address, or the new
        connection could not be opened) spends the limit as well:

        max_size | nodes | pool before rebalance | migrations | limit | rebalance loops
        6        | 3     | 3/3/0                 | 2          | 1     | 2
        100      | 4     | 33/33/1/33            | 24         | 10    | 3
        100      | 10    | 12/11/11/.../0        | 10         | 10    | 1
        1000     | 3     | 1000/0/0              | 666        | 100   | 7

        Note:
            The table assumes there is a free connection to migrate on every step. A
            migration takes a connection out of the free list, so when every connection
            to the overloaded node is in use, the loop can only mark one of them to evict
            and stops: the counts do not change until the caller releases it. Under that
            kind of load the pool converges at one connection per refresh no matter what
            `rebalance_pool_divisor` is, and stays one connection short until the next
            refresh refills the freed slot.

        Args:
            topology: The topology snapshot providing addresses for the underloaded nodes.
            candidate_list: Uuids of the instances the pool is allowed to connect to. A node
                the pool cannot connect to is dropped from the list for this refresh.

        Note:
            Rebalance can stop early, before the limit is reached, when:

            - the pool is closed -- nothing is migrated;
            - every connection to the overloaded node is in use -- one of them is marked
                `to_evict` instead

            It does not start at all while some `to_evict` connection is still held by a
            caller.
        """
        self._close_excess_connections()

        with self._condition:
            if any(conn.to_evict for conn in self._used):
                logger.debug("Rebalance: skipping rebalance, some connections wait for eviction")
                return

        for _ in range(max(1, self._max_size // self._rebalance_pool_divisor)):
            pair = self._pick_rebalance_pair(candidate_list)
            if pair is None:
                return
            overloaded_uuid, underloaded_uuid = pair

            address = topology.instances[underloaded_uuid].address
            if address is None:
                candidate_list.remove(underloaded_uuid)
                continue

            with self._condition:
                try:
                    self._check_open()
                except RuntimeError:
                    return
                if self._topology_update_interval is None or self._current_size() > self._max_size:
                    return
                old_conn = self._pop_connection_by_instance_uuid(overloaded_uuid)
                if old_conn is None:
                    # if all connections to the overloaded node are in use, mark one of them to
                    # evict and stop current rebalance: the connection counts do not change until
                    # the caller releases the marked connection
                    conn_to_evict = next(
                        (
                            used_conn
                            for used_conn in self._used
                            if used_conn.instance_uuid == overloaded_uuid and not used_conn.to_evict
                        ),
                        None,
                    )
                    if conn_to_evict is not None:
                        conn_to_evict.to_evict = True
                    # the pool stays one connection short until the caller releases it and
                    # the next refresh refills the free slot to the least loaded node
                    logger.debug(
                        "Rebalance: no free connection to %s, evicting a busy one on release",
                        overloaded_uuid,
                    )
                    return

            if not self._swap_connection(old_conn, underloaded_uuid, address):
                # go to the next migration step without the node we could not reach
                candidate_list.remove(underloaded_uuid)
                continue

            logger.debug(
                "Rebalance: migrated one connection from %s to %s",
                overloaded_uuid,
                underloaded_uuid,
            )

    def _reconcile_pool_with_topology(self, topology: Topology) -> None:  # noqa: C901
        """
        Change the pool's connection list according to the current topology.

        Does nothing if the pool is closing/closed, or if the topology snapshot
        has no instances or no replicasets -- such a snapshot is not trustworthy.

        Finishes with a rebalance so that a full pool can still redistribute connections
        evenly across nodes.
        """
        with self._condition:
            try:
                self._check_open()
            except RuntimeError:
                logger.debug("Reconcile: pool is closing/closed, skipping")
                return

        # a snapshot without instances or replicasets means the tracker reconnected
        # and cleared the topology but failed to read it back -- a live cluster always
        # has both. Reconciling against it would close every pooled connection at once
        # so the pool is reconciled on the next refresh
        if not topology.instances or not topology.replicasets:
            logger.debug("Reconcile: topology snapshot is not trustworthy, skipping")
            return

        candidate_list = self._collect_candidate_uuids(topology)

        # `_collect_candidate_uuids` returns the candidates in topology order, so without the
        # shuffle every client sharing a Picodata cluster would break refill and rebalance ties
        # the same way and pile its connections onto the same nodes
        random.shuffle(candidate_list)
        logger.debug(
            "Reconcile: collected %s candidate(s): %s", len(candidate_list), candidate_list
        )
        candidate_set = set(candidate_list)

        # TODO we have to think about special case processing: instance changed its address
        # currently we guess it will leave pool because the connection to node (to old address)
        # will be broken and after the reconcile mechanism will refill pool
        with self._condition:
            # remove connections to dead nodes
            if self._topology_update_interval is None:
                return
            for conn in self._used:
                if conn.instance_uuid not in candidate_set:
                    conn.to_evict = True

            stale_conns = [conn for conn in self._pool if conn.instance_uuid not in candidate_set]
            for conn in stale_conns:
                self._pool.remove(conn)

            state = self._instance_ejection_state
            if state is not None:
                state.clear_not_in(topology.instances)

        # close stale connections outside the lock --
        # this must not block other threads waiting to acquire/release
        for conn in stale_conns:
            try:
                conn.close()
            except Exception as e:
                logger.warning("Reconcile: could not close stale connection: %s", e)

        logger.debug(
            "Reconcile: after closing dead connections there are %s/%s left",
            self._current_size(),
            self._max_size,
        )
        while candidate_list:
            with self._condition:
                if self._topology_update_interval is None:
                    return
                try:
                    self._check_open()
                except RuntimeError:
                    return
                if self._current_size() >= self._max_size:
                    break

            # pick a candidate that keeps connections balanced across nodes
            instance_uuid = self._pick_refill_candidate(candidate_list)
            if instance_uuid is None:
                break

            address = topology.instances[instance_uuid].address
            if address is None:
                candidate_list.remove(instance_uuid)
                continue

            # connect to the chosen node with DSN creds
            dsn = _replace_dsn_host(self._connect_dsns[0], address)
            try:
                conn = Connection.connect(
                    dsn,
                    **self._pool_connect_kwargs,
                )
                conn._on_query_error = self._on_query_error
            except Exception as e:
                self._record_connection_failure(instance_uuid, e, during_connect=True)
                logger.warning("Reconcile: could not connect to %s: %s", address, e)
                candidate_list.remove(instance_uuid)
                continue

            conn.instance_uuid = instance_uuid
            conn_to_close: Connection | None = None
            with self._condition:
                try:
                    self._check_open()
                except RuntimeError:
                    conn_to_close = conn
                else:
                    if (
                        self._topology_update_interval is None
                        or self._current_size() >= self._max_size
                    ):
                        conn_to_close = conn
                    else:
                        self._insert_randomly(conn)
                        self._condition.notify_all()

            if conn_to_close is not None:
                try:
                    conn_to_close.close()
                except Exception as e:
                    logger.warning("Reconcile: could not close unused connection: %s", e)
                return

        # a full pool never refills, so an imbalance left behind by a node that was
        # unreachable while the pool filled up is fixed here instead
        self._rebalance_pool(topology, candidate_list)

        left = self._current_size()
        if left == 0:
            logger.warning(
                "Reconcile: pool is left empty until next reconcile, no candidate node to connect to"
            )
        logger.debug("Reconcile: done with %s/%s pool connections", left, self._max_size)

    def _fetch_bootstrap_instance_info(
        self,
        conn: Connection,
    ) -> dict[str, InstanceInfo]:
        try:
            cursor = conn.cursor()
            try:
                cursor.execute(BOOTSTRAP_INSTANCE_INFO_QUERY)
                rows = [
                    cast(
                        InstanceInfoRow,
                        {
                            "address": address,
                            "tier": tier,
                            "instance_uuid": instance_uuid,
                            "current_state": current_state,
                            "current_master_uuid": current_master_uuid,
                        },
                    )
                    for address, tier, instance_uuid, current_state, current_master_uuid in cursor.fetchall()
                ]
            finally:
                cursor.close()

            instance_info_by_address, malformed_rows = _parse_instance_info_by_address(rows)
        except Exception:
            logger.debug(
                "Bootstrap by DSN: could not preload instance info",
                exc_info=True,
            )
            return {}

        if malformed_rows:
            logger.warning(
                "Failed to decode %d bootstrap instance info row(s)",
                len(malformed_rows),
            )

        return instance_info_by_address

    def _discover_instances(self) -> list[str]:
        # try each bootstrap DSN until one succeeds
        temp_conn: Connection | None = None
        last_error: Exception | None = None
        for dsn in self._connect_dsns:
            try:
                candidate = Connection.connect(dsn, **self._connect_kwargs)
                temp_conn = candidate
                break
            except Exception as e:
                last_error = e

        if temp_conn is None:
            raise RuntimeError(
                f"Could not connect to any bootstrap node {self._redacted_dsn}: {last_error}"
            ) from last_error

        try:
            rows = self._fetch_discovery_rows(temp_conn)
            online_addresses, self._instance_uuid_by_address = _collect_online_nodes(
                cast(Iterable[DiscoveryRow], rows), self._instance_role
            )
            if not online_addresses:
                if self._instance_role != "any" and self._forbidden_tiers:
                    raise ValueError(
                        "No online nodes available after applying pool filters: "
                        f"instance_role={self._instance_role!r}, "
                        f"forbidden_tiers={self._forbidden_tiers}"
                    )
                if self._instance_role != "any":
                    raise ValueError(
                        "No online nodes available after applying instance_role filter: "
                        f"{self._instance_role!r}"
                    )
                if self._forbidden_tiers:
                    raise ValueError(
                        "No online nodes available after applying forbidden_tiers filter: "
                        f"{self._forbidden_tiers}"
                    )
                raise ValueError("No online nodes discovered")
            return online_addresses
        finally:
            temp_conn.close()

    def _fetch_discovery_rows(self, conn: Connection) -> list[DiscoveryRow]:
        # all instance addresses excluding forbidden tiers
        if self._forbidden_tiers:
            placeholders = ", ".join("%s" for _ in range(len(self._forbidden_tiers)))
            query = DISCOVERY_INSTANCE_ADDRESSES_EXCLUDING_TIERS_QUERY_TEMPLATE.format(
                placeholders=placeholders
            )
            parameters = tuple(self._forbidden_tiers)
        else:
            # all instance addresses
            query = DISCOVERY_INSTANCE_ADDRESSES_QUERY
            parameters = None

        cursor = conn.cursor()
        cursor.execute(query, parameters)
        discovery_rows: list[DiscoveryRow] = []
        # The discovery SELECT returns rows as
        # (current_state, address, instance_uuid, current_master_uuid).
        for current_state, address, instance_uuid, current_master_uuid in cursor.fetchall():
            discovery_rows.append(
                cast(
                    DiscoveryRow,
                    {
                        "current_state": current_state,
                        "address": address,
                        "instance_uuid": instance_uuid,
                        "current_master_uuid": current_master_uuid,
                    },
                )
            )
        return discovery_rows

    def _get_dsn_connection(
        self, on_query_metadata: Callable[[PreparedStatementMetadata], None] | None = None
    ) -> Connection | None:
        # try each bootstrap DSN until one succeeds
        # used by the topology tracker and by query-metadata service
        last_error: Exception | None = None
        for dsn in self._connect_dsns:
            try:
                return Connection.connect(
                    dsn, on_query_metadata=on_query_metadata, **self._connect_kwargs
                )
            except Exception as e:
                last_error = e

        logger.warning("no bootstrap DSN is reachable: %s", last_error)
        return None

    def _has_non_ejected_usable_instance(self, state: InstanceEjectionState) -> bool:
        for conn in (*self._pool, *self._used):
            if conn.to_evict or conn.is_closed():
                continue
            instance_uuid = conn.instance_uuid
            if instance_uuid is None or not state.is_ejected(instance_uuid):
                return True
        return False

    def _next_ejection_deadline(
        self,
        state: InstanceEjectionState,
        connections: Iterable[Connection],
    ) -> float | None:
        deadlines: list[float] = []
        for conn in connections:
            instance_uuid = conn.instance_uuid
            if instance_uuid is None:
                continue
            deadline = state.ejected_until(instance_uuid)
            if deadline is not None:
                deadlines.append(deadline)
        return min(deadlines, default=None)

    def _eligible_idle_connections(self) -> tuple[list[Connection], float | None]:
        """Return idle connections available to automatic selection and the next
        ejection expiry that may make another connection available.

        Must be called with `self._condition` held.
        """
        state = self._instance_ejection_state
        if state is None:
            return list(self._pool), None

        eligible: list[Connection] = []
        ejected: list[Connection] = []
        for conn in self._pool:
            instance_uuid = conn.instance_uuid
            if instance_uuid is None or not state.is_ejected(instance_uuid):
                eligible.append(conn)
            else:
                ejected.append(conn)

        if eligible:
            return eligible, None

        # Do not fail open just because a healthy instance has no idle connection:
        # wait for it instead of immediately reusing an ejected one.
        if self._has_non_ejected_usable_instance(state):
            return [], self._next_ejection_deadline(state, ejected)

        # Every usable instance is ejected, so use an ejected idle connection
        # rather than making the pool unavailable.
        return ejected, None

    def _acquire_raw(self, timeout: float | None = None) -> Connection:
        """
        Internal sync counterpart of async Pool._acquire_raw.

        If no connections are available, this method will wait until one is released.

        :param timeout: Maximum time to wait for a connection if the pool is exhausted. If None, a default timeout is used.

        :return: A database connection.
        """
        start_time = time.monotonic()
        effective_timeout = timeout if timeout is not None else self._default_acquire_timeout_sec

        while True:
            with self._condition:
                self._check_open()

                # keep the pre-ejection default path allocation-free.
                if (
                    self._instance_ejection_state is None
                    and self._pool
                    and self._balance_strategy is None
                ):
                    conn = self._pool.popleft()
                    self._used.add(conn)
                    return conn

                # check if there are any available connections in the pool
                connections, ejection_deadline = self._eligible_idle_connections()
                if connections:
                    # round-robin strategy
                    if self._balance_strategy is None:
                        conn = connections[0]
                    # custom strategy
                    else:
                        try:
                            conn = self._balance_strategy(connections)
                        except Exception as e:
                            raise RuntimeError(f"balance_strategy raised an exception: {e}") from e

                        if conn not in self._pool:
                            raise RuntimeError("balance_strategy returned a connection not in pool")
                        if conn not in connections:
                            raise RuntimeError("balance_strategy returned an ejected connection")
                    self._pool.remove(conn)

                    # mark it as currently in use
                    self._used.add(conn)
                    return conn

                elapsed = time.monotonic() - start_time
                if elapsed >= effective_timeout:
                    raise TimeoutError("Timed out waiting for a free connection in the pool")

                deadline_timeout = (
                    ejection_deadline - time.monotonic()
                    if ejection_deadline is not None
                    else effective_timeout
                )
                wait_timeout = min(effective_timeout - elapsed, deadline_timeout)
                if wait_timeout <= 0:
                    continue

                # if no connections are available, wait briefly before retrying
                # this gives other threads (like `_release`) a chance to return a connection to the pool
                self._condition.wait(timeout=min(0.1, wait_timeout))

    def _pop_connection_by_instance_uuid(self, instance_uuid: str) -> Connection | None:
        for idx, conn in enumerate(self._pool):
            if getattr(conn, "instance_uuid", None) == instance_uuid:
                self._pool.rotate(-idx)
                selected = self._pool.popleft()
                self._pool.rotate(idx)
                return selected
        return None

    def _try_acquire_by_instance_uuid(
        self,
        instance_uuid: str,
        timeout: float | None = None,
    ) -> Connection | None:
        """
        Acquire a free connection for a specific Picodata instance UUID, or
        return None if none becomes free before the timeout or the requested
        instance is ejected while the pool has another usable non-ejected instance.
        """
        if not instance_uuid:
            raise ValueError("instance_uuid must be a non-empty string")

        start_time = time.monotonic()
        effective_timeout = timeout if timeout is not None else self._default_acquire_timeout_sec

        while True:
            with self._condition:
                self._check_open()

                state = self._instance_ejection_state
                ejection_deadline = (
                    state.ejected_until(instance_uuid) if state is not None else None
                )
                if (
                    state is not None
                    and ejection_deadline is not None
                    and self._has_non_ejected_usable_instance(state)
                ):
                    logger.debug(
                        "Pool: instance %s is ejected until monotonic deadline %.3f",
                        instance_uuid,
                        ejection_deadline,
                    )
                    return None
                conn = self._pop_connection_by_instance_uuid(instance_uuid)
                if conn is not None:
                    self._used.add(conn)
                    return conn

                elapsed = time.monotonic() - start_time
                if elapsed >= effective_timeout:
                    return None

                # if the pool has no connection to this node, either free or in use, waiting is
                # unnecessary: only a topology reconciliation can add one, and it runs on the
                # topology refresh interval, so we give up right away instead of holding the
                # calling thread for the whole timeout
                if not any(
                    getattr(used_conn, "instance_uuid", None) == instance_uuid
                    for used_conn in self._used
                ):
                    return None

                self._condition.wait(timeout=min(0.1, effective_timeout - elapsed))

    def _acquire_raw_by_instance_uuid(
        self,
        instance_uuid: str,
        timeout: float | None = None,
    ) -> Connection:
        """
        Acquire a free connection for a specific Picodata instance UUID.

        This path bypasses the pool balance strategy and never falls back to
        regular balancing. If the instance is ejected while another usable
        instance remains non-ejected, or no matching free connection appears before the
        timeout, `TimeoutError` is raised.
        """
        conn = self._try_acquire_by_instance_uuid(instance_uuid, timeout)
        if conn is None:
            raise TimeoutError(
                "Timed out waiting for a free connection with "
                f"instance_uuid {instance_uuid!r} in the pool"
            )
        return conn

    def connection(self, timeout: float | None = None) -> _PoolConnectionContext:
        """Return a context manager for a pooled connection.

        If no connections are available, this method will wait until one is released.
        Unlike the async pool, the sync public API intentionally does not expose
        manual acquire/release methods.

        Args:
            timeout: Maximum time to wait for a connection if the pool is exhausted.
                If None, a default timeout is used.

        Returns:
            A context manager that returns a sync `Connection` from `with`.

        Examples:
            ```python
            with pool.connection() as conn:
                cursor = conn.cursor()
                cursor.execute("UPDATE ...")
            ```
        """
        return _PoolConnectionContext(self, timeout)

    def connection_by_instance_uuid(
        self,
        instance_uuid: str,
        timeout: float | None = None,
    ) -> _PoolConnectionContext:
        """Return a context manager for a pooled connection with `instance_uuid`.

        This bypasses `balance_strategy`. If the instance is currently ejected
        while another usable instance remains non-ejected, `TimeoutError` is raised
        immediately. Otherwise, if no free connection with the given UUID
        becomes available before the timeout (see
        [Pool.routing_acquire_timeout][picopyn.synchronous.pool.Pool]), `TimeoutError` is raised.
        """
        return _PoolConnectionContext(self, timeout, instance_uuid)

    def connection_by_tier_and_bucket_id(
        self,
        tier: str,
        bucket_id: int,
        timeout: float | None = None,
    ) -> _PoolConnectionContext:
        """Acquire a pooled connection to the master owning `bucket_id` in `tier`.

        The method uses the current topology to resolve
        `tier + bucket_id` to the owning replicaset's master instance, then
        acquires a free pooled connection with that instance UUID.

        If `instance_role="replica"`, topology tracking is disabled, or the current topology cannot map
        the bucket to a master instance, `RuntimeError` is raised.
        If the master is currently ejected while another usable instance remains non-ejected,
        `TimeoutError` is raised immediately. Otherwise, if the master is known
        but no matching free connection appears before the timeout, `TimeoutError`
        is raised.

        Args:
            tier: Name of the tier the bucket belongs to.
            bucket_id: Bucket id to resolve to its owning replicaset's master instance.
            timeout: Maximum time to wait for a matching connection to become free.
                If None, a default timeout is used.

        Returns:
            A context manager that can be used with `with` to acquire a connection.

        Examples:
            ```python
            with pool.connection_by_tier_and_bucket_id("default", 42) as conn:
                conn.execute("UPDATE ...")
            ```
        """
        if not tier:
            raise ValueError("tier must be a non-empty string")

        if self._instance_role == "replica":
            raise RuntimeError(
                "Cannot acquire connection by tier and bucket_id: "
                "instance_role='replica' excludes current replicaset masters from the pool"
            )

        tracker = self._topology_tracker
        if self._topology_update_interval is None or tracker is None:
            raise RuntimeError(
                "Cannot acquire connection by tier and bucket_id: topology tracking is disabled"
            )

        master_uuid = tracker.find_bucket_master_uuid(tier, bucket_id)
        if master_uuid is None:
            raise RuntimeError(
                "Cannot acquire connection by tier and bucket_id: no known master for "
                f"tier {tier!r}, bucket_id {bucket_id!r}"
            )

        return self.connection_by_instance_uuid(master_uuid, timeout)

    def create_sharding_key_factory(self, table_name: str) -> ShardingKeyFactory:
        """Create a reusable sharding key factory for a Picodata table.

        The method reads the table's distribution schema once. Creating keys
        from the returned factory is local and does not perform database requests.

        Args:
            table_name: Exact Picodata table name.

        Returns:
            A factory configured with the table's sharding fields, tier and bucket count.

        Raises:
            ValueError: If `table_name` is empty or the table distribution
                metadata is invalid or unsupported.
            RuntimeError: If topology tracking is disabled, table metadata is
                not found, or tier metadata is unavailable.
        """
        if not table_name:
            raise ValueError("table_name must be a non-empty string")

        tracker = self._topology_tracker
        if self._topology_update_interval is None or tracker is None:
            raise RuntimeError(
                "Cannot create sharding key factory: topology tracking is disabled "
                "or the pool is not open"
            )

        row = self.fetchrow(SYNC_TABLE_SHARDING_QUERY, (table_name,))
        if row is None:
            raise RuntimeError(f"Table {table_name!r} metadata not found")

        table_info = _parse_table_sharding_info(table_name, row[0], row[1])
        bucket_count = tracker.get_bucket_count(table_info.tier)
        if bucket_count is None:
            tracker.refresh_now()
            bucket_count = tracker.get_bucket_count(table_info.tier)
        if bucket_count is None:
            raise RuntimeError(f"Tier metadata is not found for table {table_name!r}")

        return ShardingKeyFactory(table_info, bucket_count, DecimalFormat.TEXT)

    def connection_by_sharding_key(
        self,
        sharding_key: ShardingKey,
        timeout: float | None = None,
    ) -> _PoolConnectionContext:
        """Return a connection to the master selected by `sharding_key`.

        This is explicit routing: if the current topology cannot resolve the
        key or no matching connection becomes available, the method raises
        instead of falling back to a random pool connection.

        Args:
            sharding_key: Key created by a table's sharding key factory.
            timeout: Maximum time to wait for a matching connection to become free.

        Returns:
            A context manager that can be used with `with`.

        Raises:
            TypeError: If `sharding_key` is not a `ShardingKey`.
            RuntimeError: If topology tracking is disabled or the key cannot
                be resolved to a known master.
            TimeoutError: If the selected master is ejected while another usable
                instance remains non-ejected, or no matching connection becomes available before
                the timeout.
        """
        if not isinstance(sharding_key, ShardingKey):
            raise TypeError("sharding_key must be a ShardingKey")
        return self.connection_by_tier_and_bucket_id(
            sharding_key.tier,
            sharding_key.bucket_id,
            timeout,
        )

    def _record_connection_failure(
        self,
        instance_uuid: str | None,
        exc: BaseException,
        *,
        during_connect: bool,
        connection_is_broken: bool = False,
    ) -> bool:
        if not is_connection_failure(
            exc,
            during_connect=during_connect,
            connection_is_broken=connection_is_broken,
        ):
            return False

        state = self._instance_ejection_state
        if instance_uuid is not None and state is not None:
            with self._condition:
                # Candidate filtering uses the same lock, so a just-ejected
                # instance cannot be selected between filtering and removal.
                if state.record_failure(instance_uuid):
                    self._condition.notify_all()
        return True

    def _on_query_error(self, conn: Connection, exc: BaseException) -> None:
        """Record a connection-level failure and ask the topology tracker to
        refresh right now, on top of its usual timer.

        `trigger()` is synchronous and non-blocking: it just wakes up the
        tracker's own loop, so we don't spawn or track a task here.
        """
        raw_connection = getattr(conn, "conn", None)
        if not self._record_connection_failure(
            conn.instance_uuid,
            exc,
            during_connect=False,
            connection_is_broken=getattr(raw_connection, "broken", False) is True,
        ):
            return

        tracker = self._topology_tracker
        if self._topology_update_interval is not None and tracker is not None:
            tracker.trigger()

    def _acquire_routed_connection(
        self,
        query: str,
        parameters: Params,
    ) -> Connection | None:
        if self._instance_role == "replica" or self._query_metadata_cache_size is None:
            return None
        service = self._query_metadata_service
        if service is None:
            return None

        metadata = service.request_if_missing(query, parameters)
        if metadata is None or not metadata.tier or not metadata.dk_meta:
            return None

        tracker = self._topology_tracker
        if self._topology_update_interval is None or tracker is None or not tracker.has_tiers():
            return None

        if not isinstance(parameters, Sequence) or isinstance(parameters, str | bytes | bytearray):
            return None

        bucket_count = tracker.get_bucket_count(metadata.tier)
        if bucket_count is None:
            return None

        bucket_id = calculate_bucket_id(
            metadata.dk_meta,
            parameters,
            bucket_count,
            decimal_format=DecimalFormat.TEXT,
        )
        if bucket_id is None:
            return None

        master_uuid = tracker.find_bucket_master_uuid(metadata.tier, bucket_id)
        if master_uuid is None:
            return None

        state = self._instance_ejection_state
        if state is not None and state.is_ejected(master_uuid):
            return None

        return self._try_acquire_by_instance_uuid(
            master_uuid, timeout=self._routing_acquire_timeout
        )

    def execute(self, query: QueryNoTemplate, parameters: Params | None = None) -> str:
        """Executes a query that does not return rows (e.g. INSERT, UPDATE, DELETE).

        The driver supports shard-aware routing for parameterized 1-row INSERT. It is based
        on two mechanisms:
        [cache service][picopyn.synchronous.metadata_service.QueryMetadataService] for query
        meta and [topology tracking][picopyn.synchronous.topology_tracker.TopologyTracker]
        that allows us to calculate bucket id and use it to choose Picodata replicaset master.

        So, the shard-aware routing requires:

        - metadata cache service enabled (default; see
          [Pool.query_metadata_cache_size][picopyn.synchronous.pool.Pool])
        - topology tracking enabled (see [Pool.topology_update_interval][picopyn.synchronous.pool.Pool])
        - `instance_role` is not `"replica"`
        - Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
        - execute query more than once

        Read about [cache warming][picopyn.synchronous.metadata_service.QueryMetadataService.request_if_missing].

        Args:
            query: The SQL query string.
            parameters: Optional DB-API parameters for the SQL query.

        Returns:
            The result of the query execution.

        Raises:
            ProgrammingError: If the statement prepared on the connection was invalidated
                by a schema change (Picodata SQLSTATE `42999`). The connection drops its
                prepared statements and the cached query meta is evicted, so the caller
                decides how to react (execute the query again, execute another query, etc.).
                After DDL the connection raises the error only when it executes a prepared
                statement, so it depends on the `prepare_threshold` setting (see
                `prepare_threshold` in [Pool][picopyn.synchronous.pool.Pool]).
            Error: Other query failures, see [errors][picopyn.synchronous.errors.Error].
        """
        # prepare and cache query meta only for parameterized queries
        conn = None
        if parameters and isinstance(query, str):
            conn = self._acquire_routed_connection(query, parameters)

        if conn is None:
            conn = self._acquire_raw()

        try:
            cursor = conn.cursor()
            cursor.execute(query, parameters)
            status = cursor.statusmessage
            if status is None:
                raise RuntimeError("Query executed without status message")
            return status
        except Exception as e:
            # TODO retry execution to hide from user the invalidation error
            # (we have to re-calculate meta and cache it, but in the time we can execute
            # the query without routing)
            service = self._query_metadata_service
            if (
                isinstance(query, str)
                and self._query_metadata_cache_size is not None
                and service is not None
                and is_stmt_invalidated_error(e)
            ):
                service.evict(query)
            raise
        finally:
            self._release(conn)

    def fetch(
        self,
        query: QueryNoTemplate,
        parameters: Params | None = None,
    ) -> list[tuple[object, ...]]:
        """Executes a query and fetches all resulting rows.

        Args:
            query: The SQL query string.
            parameters: Optional DB-API parameters for the SQL query.

        Returns:
            A list of rows returned by the query.

        Raises:
            Error: If the query fails, see [errors][picopyn.synchronous.errors.Error].
        """
        # TODO: add cache usage when Picodata will support it for DQL
        # https://git.picodata.io/core/picodata/-/issues/2226
        with self.connection() as conn:
            try:
                cursor = conn.cursor()
                cursor.execute(query, parameters)
                return cursor.fetchall()
            except Exception as e:
                # TODO retry execution to hide from user the invalidation error
                service = self._query_metadata_service
                if (
                    isinstance(query, str)
                    and self._query_metadata_cache_size is not None
                    and service is not None
                    and is_stmt_invalidated_error(e)
                ):
                    service.evict(query)
                raise

    def fetchrow(
        self,
        query: QueryNoTemplate,
        parameters: Params | None = None,
    ) -> tuple[object, ...] | None:
        """Executes a query and fetches a single row (first row).

        Args:
            query: The SQL query string.
            parameters: Optional DB-API parameters for the SQL query.

        Returns:
            The first row returned by the query, or None.

        Raises:
            Error: If the query fails, see [errors][picopyn.synchronous.errors.Error].
        """
        # TODO: add cache usage when Picodata will support it for DQL
        # https://git.picodata.io/core/picodata/-/issues/2226
        with self.connection() as conn:
            try:
                cursor = conn.cursor()
                cursor.execute(query, parameters)
                return cursor.fetchone()
            except Exception as e:
                # TODO retry execution to hide from user the invalidation error
                service = self._query_metadata_service
                if (
                    isinstance(query, str)
                    and self._query_metadata_cache_size is not None
                    and service is not None
                    and is_stmt_invalidated_error(e)
                ):
                    service.evict(query)
                raise

    def _release(self, conn: Connection) -> None:
        """Release a previously acquired connection back to the pool.

        Args:
            conn: The connection to release.
        """
        conn_to_close = None
        with self._condition:
            if conn not in self._used:
                return

            # Connections already marked for eviction will leave the pool too.
            # Count them out before reserving another eviction during concurrent releases.
            retained_size = self._current_size()
            if retained_size > self._max_size:
                retained_size -= sum(c.to_evict for c in self._used)
            if (
                self._close_event is not None
                or conn.to_evict
                or conn.is_closed()
                or retained_size > self._max_size
            ):
                conn.to_evict = True
                # keep it in `_used` until it is really closed, so a close() waiting for
                # this connection cannot return while the connection is still closing
                conn_to_close = conn
            else:
                self._used.remove(conn)
                # TODO(#42): Consider validating connection liveness before returning it to the pool.
                # Update sync/async release docs when resolving this TODO.
                # https://git.picodata.io/core/drivers/picopyn/-/merge_requests/45#note_297586
                self._pool.append(conn)
                self._condition.notify_all()

        if conn_to_close is None:
            return

        try:
            conn_to_close.close()
        except Exception as e:
            logger.warning("Pool release: could not close evicted connection: %s", e)
        finally:
            with self._condition:
                # terminate() may have already dropped it
                self._used.discard(conn_to_close)
                self._condition.notify_all()

    @property
    def topology(self) -> Topology | None:
        """The [cluster topology][picopyn.topology] we currently know about,
        or `None` if `topology_update_interval` is disabled.

        Read-only: looking at it doesn't change the pool in any way.
        Returns a deepcopy snapshot, so repeated calls copy the whole topology.

        Pool changes its own connection membership according to the current
        topology (reconcile). So, the reconcile requires topology tracking enabled
        (see [Pool.topology_update_interval][picopyn.synchronous.pool.Pool]):
        it runs after the tracker's refresh, so disabling tracking disables it.

        Reconcile works in both modes (see [Pool.enable_discovery][picopyn.synchronous.pool.Pool]);
        the mode only decides which nodes may be added:

        - discovery mode: any node of the cluster
        - bootstrap mode: only the nodes listed in `dsn`

        In either mode, `instance_role` and the other user-connection filters further
        restrict this candidate set.

        Note:
            In bootstrap mode nodes are matched by their "host:port" string, so a
            `dsn` host must be spelled exactly as the cluster reports it in
            [picodata's system table](https://docs.picodata.io/picodata/devel/architecture/system_tables/#_pico_peer_address)
            (for example "localhost:5432" does not match "127.0.0.1:5432")

        # Reconcile

        The reconcile algorithm targets the following desired pool composition:

        - The pool is always filled up to its maximum size (if possible)
        - When creating new connections, the reconciliation process selects candidates
            to maintain a balanced distribution (if possible)
        - A full pool is balanced as well: if the topology reports a "new" node (a node
            that has returned to Online, or a new node that has joined the Picodata
            cluster), it initially has zero connections while other nodes already have
            connections. This situation must be resolved to maintain a balanced
            distribution across nodes

        Note:
            Refills happen on refresh only, so a pool left underfilled (every candidate
            node is Offline, or connecting to it failed) stays that way until the next
            refresh, i.e. up to `topology_update_interval` seconds. While the pool is
            empty, every `connection()` waits out its timeout and then raises `TimeoutError`.

        --8<-- "_reconcile_pool_with_topology.md"

        # Rebalance

        While the reconcile goal is to evict connections to dead nodes and refill the pool
        up to its max size, the rebalance goal is to even out the connection load across
        the nodes when the pool is already full. Without rebalance the pool stays skewed:

        - the cluster has the `[i1 i2 i3]` nodes and the pool max size is 6, so the pool
            initially establishes the connections `[i1 i2 i3 i1 i2 i3]`
        - the i3 node dies, so reconcile changes the pool composition: first it removes
            the connections to the dead node, `[i1 i2 i1 i2]`, and then fills the pool up
            to its max size, `[i1 i2 i1 i2 i1 i2]`
        - the i3 node comes back online, but the pool has no free slots left to connect
            to it: the pool is imbalanced, with 3 connections to i1 and i2 each and none
            to i3
        - the imbalance lasts until one of the pooled nodes dies

        So, to avoid a problem like that, a full pool is rebalanced:

        --8<-- "_rebalance_pool_after_reconcile.md"

        # Example cases
        <details><summary>Expand reconcile/rebalance cases</summary>

        --8<-- "_reconcile_example_node_outside_pool_died.md"

        --8<-- "_reconcile_example_pooled_node_died_size_eq.md"

        --8<-- "_reconcile_example_pooled_node_died_size_gt.md"

        --8<-- "_reconcile_example_pooled_node_died_size_lt.md"

        --8<-- "_reconcile_example_rebalance.md"
        </details>
        """
        if self._topology_update_interval is None:
            return None
        topology_tracker = self._topology_tracker
        return topology_tracker.snapshot() if topology_tracker is not None else None

    def get_query_metadata(self, query: str) -> PreparedStatementMetadata | None:
        """Return Picodata's [distribution key metadata][picopyn.query_metadata.is_stmt_invalidated_error]
        for `query`, blocking until it's resolved.

        Raises:
            RuntimeError: the metadata service is not running -- either it is disabled
                (see [Pool.query_metadata_cache_size][picopyn.synchronous.pool.Pool])
                or the pool is not opened.

        The pool caches query metadata -- see below for details:

        <details><summary>Expand pool initialization in cache meta context</summary>
        --8<-- "_query_metadata_pool_init.md"
        </details>

        --8<-- "_query_metadata_usage_lookup.md"
        """
        with self._condition:
            self._check_open()

        if self._query_metadata_cache_size is None:
            raise RuntimeError("query metadata service is not running")
        service = self._query_metadata_service
        if service is None:
            raise RuntimeError("query metadata service is not running")

        return service.get_query_metadata(query)

    def close(self, timeout: float | None = None) -> None:
        """
        Close the pool: close free connections and wait for the used ones to come back.

        Args:
            timeout: Maximum graceful shutdown time for the call that starts pool
                shutdown. If None, the default shutdown timeout of 1 hour is used.

        Only the first close() call starts shutdown and determines the shutdown timeout.
        Concurrent close() calls wait for the same shutdown operation and do not change
        the first timeout. If that timeout expires, any remaining connections are closed
        anyway, even the ones still checked out.

        The timeout bounds the waiting shutdown does, not the blocking calls it makes:
        closing a connection cannot be interrupted, so a close that hangs on an
        unresponsive node still runs to completion.

        Note:
            This should be called during application shutdown to clean up resources.
            Do not call it from a thread that still holds a pooled connection: shutdown
            waits for that connection to be released, so it would block until the timeout.
            Use [terminate()][picopyn.synchronous.pool.Pool.terminate] to close the pool
            without waiting.
        """
        with self._condition:
            close_event = self._close_event
            first_call = close_event is None
            if close_event is None:
                close_event = Event()
                self._close_event = close_event
                # let acquire waiters fail fast instead of polling until their timeout
                self._condition.notify_all()

        if not first_call:
            # somebody else already started shutdown, wait for the same operation
            close_event.wait()
            return

        effective_timeout = _DEFAULT_CLOSE_TIMEOUT_SEC if timeout is None else timeout
        try:
            self._close_gracefully(time.monotonic() + effective_timeout)
        finally:
            self.terminate()

    def _close_gracefully(self, deadline: float) -> None:
        """Close free connections and wait until the used ones are released.

        Blocking `close()` calls cannot be interrupted, so `deadline` only bounds the
        waiting: the join of the topology update loop and the wait for connections that
        are still in-use.
        """
        if not self._wait_for_service_shutdowns(deadline):
            return

        # `terminate()` must stay able to cut the wait below short, so the lock covers
        # only the closing, and is released before the pool waits on `_used`
        with self._shutdown_lock:
            # stop the tracker before closing pool connections, so a refresh cannot
            # change pool membership mid-shutdown
            topology_tracker = self._topology_tracker
            if topology_tracker is not None:
                try:
                    topology_tracker.stop(timeout=max(0.0, deadline - time.monotonic()))
                except Exception as e:
                    logger.warning("Pool close: could not stop topology tracker: %s", e)

            with self._condition:
                total = self._current_size()
                conns_to_close = list(self._pool)
                self._pool.clear()
                # a connection checked out during shutdown never comes back to the pool,
                # so mark it the way any evicted connection is marked
                for conn in self._used:
                    conn.to_evict = True

            self._close_connections(conns_to_close)

        # connections that are still in-use are closed by `_release()`, because the
        # pool is no longer open, so just wait until their owners hand them back
        with self._condition:
            released = self._condition.wait_for(
                lambda: not self._used,
                timeout=max(0.0, deadline - time.monotonic()),
            )
            still_used = len(self._used)

        if not released:
            logger.warning(
                "Pool close timed out; closing %d connection(s) still in use", still_used
            )
            # the deadline is spent, so the query-metadata service is left to the
            # terminate() that close() runs next: closing it gracefully would join a
            # background prepare
            return

        if self._query_metadata_service is not None:
            try:
                self._query_metadata_service.close()
            except Exception as e:
                logger.warning("Pool close: could not close query metadata service: %s", e)

        logger.info("Pool closed (%d connection(s))", total)

    def _wait_for_service_shutdowns(self, deadline: float) -> bool:
        for service_shutdown in (self._topology_disable, self._metadata_disable):
            if service_shutdown is not None:
                try:
                    service_shutdown.result(timeout=max(0.0, deadline - time.monotonic()))
                except FutureTimeoutError:
                    return False
                except Exception as e:
                    logger.warning("Pool close: service shutdown failed: %s", e)
        return True

    def terminate(self) -> None:
        """Close all connections owned by the pool without graceful close.

        Used connections are closed right away.
        Calling it while a `close()` is running finishes that shutdown instead
        of starting another one. On an already closed pool it does nothing.

        Concurrent terminate() calls run one after another: the first one does the
        work, the rest return once it is done.
        """
        # the whole body is serialized, so `_close_event` cannot be set by another
        # shutdown while this one still has connections to close
        with self._shutdown_lock:
            with self._condition:
                close_event = self._close_event
                if close_event is not None and close_event.is_set():
                    return
                if close_event is None:
                    close_event = Event()
                    self._close_event = close_event

                conns = [*self._pool, *self._used]
                self._pool.clear()
                # dropping `_used` also releases a close() waiting for these connections,
                # the way async terminate() cancels the running close task
                self._used.clear()
                self._condition.notify_all()

            tracker = self._topology_tracker
            if tracker is not None:
                try:
                    tracker.terminate()
                except Exception as e:
                    logger.warning("Pool terminate: could not terminate topology tracker: %s", e)

            service = self._query_metadata_service
            if service is not None:
                try:
                    service.terminate()
                except Exception as e:
                    logger.warning(
                        "Pool terminate: could not terminate query-metadata service: %s", e
                    )

            self._close_connections(conns)

            with self._condition:
                close_event.set()
                self._condition.notify_all()

    def _close_connections(self, conns: Iterable[Connection]) -> None:
        for conn in conns:
            try:
                conn.close()
            except Exception as e:
                logger.warning("Pool close: could not close connection: %s", e)

`balance_strategy` `property` `writable`

A custom strategy function to select a connection from the pool. If None, round-robin strategy is used.

Can be changed: a new value applies to the next ordinary acquire.

Raises:

Type	Description
`ValueError`	If the new value is neither callable nor `None`.
`RuntimeError`	If the pool is closing or closed.

`instance_ejection_duration` `property` `writable`

Number of seconds an ejected instance stays out of pool selection.

None if instance ejection is disabled.

While instance ejection is enabled, a new value applies to ejections started after the change; an ejection that is already active keeps its original deadline.

Raises:

Type	Description
`ValueError`	If the new value is not a positive finite number.
`RuntimeError`	If the pool is closing or closed, or instance ejection is disabled.

`instance_ejection_enabled` `property`

Whether error-based instance ejection is enabled.

Instance ejection temporarily excludes an instance from pool selection after repeated connection failures. It is enabled only if all three instance_ejection_* settings are passed to the constructor.

When an instance reaches instance_ejection_failure_threshold connection failures within instance_ejection_failure_window seconds, the instance is excluded from pool selection for instance_ejection_duration seconds. Its connections stay open, but the pool stops handing them out (see the exceptions below).

Connection failures are failed or timed-out connection attempts and connections lost while in use; SQL errors are not counted.

Instance ejection cannot be enabled after construction. Use disable_instance_ejection() to disable it.

Instance Ejection and Connection Selection

This table describes connection selection when all instance_ejection_* options are set. With instance ejection disabled, the pool uses its usual selection behavior.

Call	Conditions	Behavior
`pool.acquire()` (async) / `pool.connection()` (sync), or `pool.execute()` without routing	Instance ejection applies	A custom `balance_strategy` or the default round-robin strategy receives only eligible idle connections; ejected connections are excluded.
`pool.acquire()` (async) / `pool.connection()` (sync), or `pool.execute()` without routing	Every usable instance is ejected	The pool fails open and may use an ejected idle connection rather than become unavailable.
`pool.execute()` with metadata routing	The calculated bucket master is ejected	The pool immediately falls back to ordinary pool selection.
Explicit `acquire_by_()` (async) / `connection_by_()` (sync)	The target instance is ejected while another usable instance remains non-ejected	The call immediately raises `TimeoutError`; it does not fall back to another instance.
Explicit `acquire_by_()` (async) / `connection_by_()` (sync)	The target instance is not ejected but has no free connection	The call waits until its timeout, then raises `TimeoutError`.
Explicit `acquire_by_()` (async) / `connection_by_()` (sync)	Every usable instance is ejected	The pool fails open and may acquire a free connection to the target instance.

Note

Ejection only affects which connections are handed out. Connections to an ejected instance stay open and still count toward max_size, and pool refill and rebalancing keep treating the instance as a regular pool member. While an instance is ejected, the pool therefore serves requests with fewer connections: with two instances and one of them ejected, only about half of max_size is available, and further requests wait for a connection to a non-ejected instance.

`instance_ejection_failure_threshold` `property` `writable`

Number of connection failures from one instance required for ejection.

None if instance ejection is disabled.

While instance ejection is enabled, a new value applies to the next recorded failure; an ejection that is already active keeps its original deadline.

Raises:

Type	Description
`ValueError`	If the new value is not a positive integer.
`RuntimeError`	If the pool is closing or closed, or instance ejection is disabled.

`instance_ejection_failure_window` `property` `writable`

Sliding window in seconds used to count failures for instance ejection.

None if instance ejection is disabled.

While instance ejection is enabled, a new value applies to the next recorded failure; an ejection that is already active keeps its original deadline.

Raises:

Type	Description
`ValueError`	If the new value is not a positive finite number.
`RuntimeError`	If the pool is closing or closed, or instance ejection is disabled.

`max_size` `property` `writable`

Maximum number of connections in the pool.

Can be changed before or after the pool is opened:

increasing the limit requires topology tracking and triggers a topology refresh to refill the pool. The setter does not wait for new connections.
decreasing the limit closes excess connections as they are released. With topology tracking it also triggers a topology refresh without waiting for it: the refresh closes excess free connections and marks excess in-use ones to close on release, starting from the most loaded nodes. Without topology tracking free connections stay until acquired and released. In-use connections stay until released in any case, so the actual size can temporarily exceed this limit.

Raises:

Type	Description
`ValueError`	If the new value is less than 1.
`RuntimeError`	If the pool is closing or closed, or if the value is increased while topology tracking is disabled.

`query_metadata_cache_size` `property` `writable`

Maximum number of distinct queries to keep Picodata's distribution key metadata for, on the pool's dedicated metadata-discovery connection (see query meta usage). Oldest entries are evicted once exceeded. Must be at least 1 if set. Pass None at initialization to disable the metadata service: its technical connection is never opened and get_query_metadata raises RuntimeError.

Shrinking immediately evicts excess LRU entries, including entries for queries without metadata. The pending prepare limit changes too; already queued prepares are allowed to finish. Before open(), the value is saved for startup.

Use disable_query_metadata() to stop it. Assigning the current value is a no-op.

Raises:

Type	Description
`ValueError`	If the new cache size is less than 1.
`RuntimeError`	If the pool is closing or closed, or if the new value enables or disables the metadata service.

`rebalance_pool_divisor` `property` `writable`

How much of the pool a single rebalance loop may rebalance: at most max(1, max_size // rebalance_pool_divisor) connections are migrated from overloaded nodes to underloaded ones per reconcile, so even a pool smaller than the divisor still migrates one connection per refresh. A smaller value converges faster at the cost of more reconnects per refresh. Must be at least 1.

Can be changed: a new value applies to the next rebalance loop.

Raises:

Type	Description
`ValueError`	If the new value is less than 1.
`RuntimeError`	If the pool is closing or closed.

`routing_acquire_timeout` `property` `writable`

Time in seconds that shard-aware routing waits for a free connection to the bucket master before falling back to an ordinary pool connection. 0 means a single attempt with no waiting. The value cannot be negative.

The timeout is only spent when the pool does hold a connection to the bucket master and every such connection is busy. If the pool has no connection to that node at all, routing falls back to an ordinary pool connection at once: only a topology reconciliation can add one, and it runs on the topology refresh interval, not within the acquire timeout.

Choose the value based on your workload. If the pool is overloaded and queries frequently wait for a free connection, set this value to 0 to avoid additional waiting. If queries rarely wait for a free connection, consider setting it to a positive value appropriate for your setup. Keep in mind that, unlike the async pool, waiting here occupies the calling thread.

Can be changed: a new value applies to new routing attempts. An acquire already waiting keeps its original timeout.

Raises:

Type	Description
`ValueError`	If the new value is negative.
`RuntimeError`	If the pool is closing or closed.

`topology` `property`

The cluster topology we currently know about, or None if topology_update_interval is disabled.

Read-only: looking at it doesn't change the pool in any way. Returns a deepcopy snapshot, so repeated calls copy the whole topology.

Pool changes its own connection membership according to the current topology (reconcile). So, the reconcile requires topology tracking enabled (see Pool.topology_update_interval): it runs after the tracker's refresh, so disabling tracking disables it.

Reconcile works in both modes (see Pool.enable_discovery); the mode only decides which nodes may be added:

discovery mode: any node of the cluster
bootstrap mode: only the nodes listed in dsn

In either mode, instance_role and the other user-connection filters further restrict this candidate set.

Note

In bootstrap mode nodes are matched by their "host:port" string, so a dsn host must be spelled exactly as the cluster reports it in picodata's system table (for example "localhost:5432" does not match "127.0.0.1:5432")

Reconcile

The reconcile algorithm targets the following desired pool composition:

The pool is always filled up to its maximum size (if possible)
When creating new connections, the reconciliation process selects candidates to maintain a balanced distribution (if possible)
A full pool is balanced as well: if the topology reports a "new" node (a node that has returned to Online, or a new node that has joined the Picodata cluster), it initially has zero connections while other nodes already have connections. This situation must be resolved to maintain a balanced distribution across nodes

Note

Refills happen on refresh only, so a pool left underfilled (every candidate node is Offline, or connecting to it failed) stays that way until the next refresh, i.e. up to topology_update_interval seconds. While the pool is empty, every connection() waits out its timeout and then raises TimeoutError.

---
title: Reconcile pool according to topology
---
flowchart TD

classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555
classDef success fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef error fill:#ffebee,stroke:#f44336,color:#b71c1c

subgraph Pool["Pool"]
    FreeConns["free conns=[...]"]
    UsedConns["used conns=[...]"]
    Topology[/"Topology"/]
end

TOPOLOGY_APPLY["Apply change to topology"] -.-> Topology

EVENT_REMOVE[/"Event<br>delete instance A_uuid"/]:::neutral -.-> TOPOLOGY_APPLY
EVENT_REPLACE[/"Event<br>replace instance B_uuid"/]:::neutral -.-> TOPOLOGY_APPLY

TOPOLOGY_DETAILS_NOTE>"Topology details"]:::note
click TOPOLOGY_DETAILS_NOTE "../topology/#picopyn.topology.Topology" "Topology details"
TOPOLOGY_DETAILS_NOTE -.-> TOPOLOGY_APPLY

TOPOLOGY_APPLY --> COMPLETE{"topology has both<br>instances and replicasets?"}
COMPLETE -->|no| NOT_TRUSTED(["snapshot is not trustworthy:<br>pool is left as is<br>(exit)"]):::error
COMPLETE -->|yes| GET_LIST["build candidate_list from Topology:<br>- uuids with allowed tiers<br>- online uuids<br>- uuids matching instance role<br>- (bootstrap mode) only<br>DSN node addresses"]
GET_LIST --> REMOVE["close all connections whose<br>uuid not in candidate_list"] --> CHECK_SIZE_START

REMOVE -.-> FreeConns
REMOVE -.-> UsedConns

CHECK_SIZE_START{"current pool size < max size?"}
CHECK_SIZE_START -->|no| REBALANCE
CHECK_SIZE_START -->|yes| MIN["choose uuid from candidate_list<br>with min connections count"]

MIN -->|got candidate| ADD_CONN["try to connect to candidate<br>and add connection to pool..."]
MIN -->|no candidate| NO_CANDIDATE2(["pool is temporarily not<br>full until next event<br>(exit)"]):::error

ADD_CONN -->|connection failed| EXCLUDE["exclude the candidate uuid<br/>from list in this loop"] --> CHECK_SIZE_LOOP
ADD_CONN-->|connected| CHECK_SIZE_LOOP{"current size < max size?"}

CHECK_SIZE_LOOP -->|no| REBALANCE
CHECK_SIZE_LOOP -->|yes| MIN

REBALANCE["rebalance the pool:<br>move connections from overloaded<br>nodes to underloaded ones"]
REBALANCE --> POOL_OK(["Pool okay<br>(exit)"]):::success

REBALANCE_DETAILS_NOTE>"Rebalance details"]:::note
click REBALANCE_DETAILS_NOTE "#picopyn.synchronous.Pool.topology--rebalance" "Rebalance details"
REBALANCE_DETAILS_NOTE -.-> REBALANCE

Rebalance

While the reconcile goal is to evict connections to dead nodes and refill the pool up to its max size, the rebalance goal is to even out the connection load across the nodes when the pool is already full. Without rebalance the pool stays skewed:

the cluster has the [i1 i2 i3] nodes and the pool max size is 6, so the pool initially establishes the connections [i1 i2 i3 i1 i2 i3]
the i3 node dies, so reconcile changes the pool composition: first it removes the connections to the dead node, [i1 i2 i1 i2], and then fills the pool up to its max size, [i1 i2 i1 i2 i1 i2]
the i3 node comes back online, but the pool has no free slots left to connect to it: the pool is imbalanced, with 3 connections to i1 and i2 each and none to i3
the imbalance lasts until one of the pooled nodes dies

So, to avoid a problem like that, a full pool is rebalanced:

---
title: Rebalance pool after reconcile
---
flowchart TD

classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555
classDef success fill:#e8f5e9,stroke:#4caf50,color:#1b5e20

START(["START"]):::neutral --> TRIM

TRIM["trim connections above max size<br>(left after max size was decreased),<br>from the most loaded uuids first"] --> ABOVE

TRIM_DETAILS_NOTE>"free connections are closed now,<br>in-use ones are marked to evict<br>and closed on release"]:::note
TRIM_DETAILS_NOTE -.-> TRIM

ABOVE{"current pool size > max size?"}
ABOVE -->|yes| POOL_ABOVE(["wait for release<br>(exit)"]):::success
ABOVE -->|no| COUNT

COUNT_DETAILS_NOTE>"connections already marked<br>to evict are not counted:<br>they are closed on<br>release and refilled elsewhere"]:::note
COUNT_DETAILS_NOTE -.-> COUNT

COUNT_DETAILS_NOTE2>"candidate nodes are<br>eligible online<br>Picodata nodes"]:::note
COUNT_DETAILS_NOTE2 -.-> COUNT

COUNT["count free and in-use pool<br>connections per candidate uuid"] --> GAP

GAP_DETAILS_NOTE>"a gap of one<br>is not an imbalance"]:::note
GAP_DETAILS_NOTE -.-> GAP

GAP{"gap between<br>max count and min count<br>(most overloaded and<br>most underloaded nodes)<br>>= 2?"}
GAP -->|no| POOL_OK(["pool is balanced<br>(exit)"]):::success
GAP -->|yes| FREE{"is there a free connection<br>to the overloaded uuid?"}

FREE -->|no| EVICT_USED["mark one in-use connection<br>to the overloaded uuid to evict"]
FREE -->|yes| MIGRATE["migrate one connection:<br>take the free one out, connect<br>to the underloaded uuid,<br>close the old one"]

EVICT_DETAILS_NOTE>"release closes it, and the next<br>refresh refills the free slot to<br>the underloaded uuid"]:::note
EVICT_DETAILS_NOTE -.-> EVICT_USED

EVICT_USED --> POOL_OK3(["(exit)"]):::success

MIGRATE -->|migrated| LIMIT
MIGRATE -->|"could not connect to<br>the underloaded uuid"| EXCLUDE["put the connection back and<br>exclude the underloaded uuid<br>from the candidate list<br>in this reconcile"]
EXCLUDE --> LIMIT

LIMIT{"migration limit for<br>this reconcile reached?"}
LIMIT -->|no| GAP
LIMIT -->|yes| POOL_OK4(["(exit)"]):::success

LIMIT_DETAILS_NOTE>"the limit is proportional to<br>the pool size, see<br>Pool.rebalance_pool_divisor"]:::note
click LIMIT_DETAILS_NOTE "#picopyn.synchronous.Pool" "Pool.rebalance_pool_divisor"
LIMIT_DETAILS_NOTE -.-> LIMIT

LIMIT_DETAILS_NOTE2>"the rest is left to<br>the next refresh"]:::note
LIMIT_DETAILS_NOTE2 -.-> POOL_OK4

Example cases

Expand reconcile/rebalance cases

---
title: A node outside the pool died
---

flowchart TD

classDef offline fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef online fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

subgraph before["Before"]
    subgraph Picodata_before["Picodata"]
        p1[("ins 1")]:::online
        p2[("ins 2")]:::online
        p3[("ins 3")]:::online
        p4[("ins 4")]:::online
    end

    subgraph Pool_before["Pool: max_size=3"]
        c1["conn"] --> p1
        c2["conn"] --> p2
        c3["conn"] --> p3
    end
end
EVENT1[/"Event<br>Instance 4 dead"/]:::neutral -.-> REFILL{{"reconcile pool"}}
before -.-> REFILL

NOTE>"Node outside the pool died<br>=> pool stays unchanged"]:::note
NOTE-.->REFILL

REFILL --> after

subgraph after["After"]
    subgraph Picodata_after["Picodata"]
        p21[("ins 1")]:::online
        p22[("ins 2")]:::online
        p23[("ins 3")]:::online
        p24[("ins 4")]:::offline
    end

    subgraph Pool_after["Pool: max_size=3"]
        c21["conn"] --> p21
        c22["conn"] --> p22
        c23["conn"] --> p23
    end
end

---
title: A pooled node died (pool size = cluster size)
---

flowchart TD

classDef offline fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef online fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

subgraph before["Before"]
    subgraph Picodata_before["Picodata"]
        p1[("ins 1")]:::online
        p2[("ins 2")]:::online
        p3[("ins 3")]:::online
        p4[("ins 4")]:::online
    end

    subgraph Pool_before["Pool: max_size=4"]
        c1["conn"] --> p1
        c2["conn"] --> p2
        c3["conn"] --> p3
        c4["conn"] --> p4
    end
end
EVENT1[/"Event<br>Instance 3 dead"/]:::neutral -.-> REFILL{{"reconcile pool"}}
before -.-> REFILL
NOTE>"Add connection to instance<br>with the min (=1) conn count.<br>There are several nodes<br>with count=1<br>=> choose the first one (ins 1)"]:::note
NOTE-.->REFILL
REFILL --> after

subgraph after["After"]
    subgraph Picodata_after["Picodata"]
        p21[("ins 1")]:::online
        p22[("ins 2")]:::online
        p23[("ins 3")]:::offline
        p24[("ins 4")]:::online
    end

    subgraph Pool_after["Pool: max_size=4"]
        c21["conn"] --> p21
        c22["conn"] --> p22
        c24["conn"] --> p24
        c25["conn"] --> p21
    end
end

---
title: A pooled node died (pool size > cluster size)
---

flowchart TD

classDef offline fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef online fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

subgraph before["Before"]
    subgraph Picodata_before["Picodata"]
        p1[("ins 1")]:::online
        p2[("ins 2")]:::online
        p3[("ins 3")]:::online
        p4[("ins 4")]:::online
    end

    subgraph Pool_before["Pool: max_size=6"]
        c1["conn"] --> p1
        c2["conn"] --> p2
        c3["conn"] --> p3
        c4["conn"] --> p4
        c5["conn"] --> p1
        c6["conn"] --> p2
    end
end
EVENT1[/"Event<br>Instance 3 dead"/]:::neutral -.-> REFILL{{"reconcile pool"}}
before -.-> REFILL
NOTE>"Add connection to instance<br>with the min (=1) conn count<br>=> ins 4"]:::note
NOTE-.->REFILL
REFILL --> after

subgraph after["After"]
    subgraph Picodata_after["Picodata"]
        p21[("ins 1")]:::online
        p22[("ins 2")]:::online
        p23[("ins 3")]:::offline
        p24[("ins 4")]:::online
    end

    subgraph Pool_after["Pool: max_size=6"]
        c21["conn"] --> p21
        c22["conn"] --> p22
        c24["conn"] --> p24
        c25["conn"] --> p21
        c26["conn"] --> p22
        c27["conn"] --> p24
    end
end

---
title: A pooled node died (pool size < cluster size)
---

flowchart TD

classDef offline fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef online fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

subgraph before["Before"]
    subgraph Picodata_before["Picodata"]
        p1[("ins 1")]:::online
        p2[("ins 2")]:::online
        p3[("ins 3")]:::online
        p4[("ins 4")]:::online
    end

    subgraph Pool_before["Pool: max_size=3"]
        c1["conn"] --> p1
        c2["conn"] --> p2
        c3["conn"] --> p3
    end
end
EVENT1[/"Event<br>Instance 3 dead"/]:::neutral -.-> REFILL{{"reconcile pool"}}
before -.-> REFILL
NOTE>"Add connection to instance<br>with the min (=0) conn count<br>=> ins 4"]:::note
NOTE -.-> REFILL
REFILL --> after

subgraph after["After"]
    subgraph Picodata_after["Picodata"]
        p21[("ins 1")]:::online
        p22[("ins 2")]:::online
        p23[("ins 3")]:::offline
        p24[("ins 4")]:::online
    end

    subgraph Pool_after["Pool: max_size=3"]
        c21["conn"] --> p21
        c22["conn"] --> p22
        c24["conn"] --> p24
    end
end

---
title: (Rebalance) a pooled node returned to Online (pool is already full)
---

flowchart TD

classDef offline fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef online fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

subgraph before["Before"]
    subgraph Picodata_before["Picodata"]
        p1[("ins 1")]:::online
        p2[("ins 2")]:::online
        p3[("ins 3")]:::offline
    end

    subgraph Pool_before["Pool: max_size=6"]
        c1["conn x3"] --> p1
        c2["conn x3"] --> p2
    end
end
EVENT1[/"Event<br>Instance 3 back Online"/]:::neutral -.-> REBALANCE{{"reconcile pool"}}
before -.-> REBALANCE
NOTE>"Pool is full, so refill<br>does nothing.<br>Initial connection counts:<br>[3/3/0], gap=3<br>=> migrate: [2/3/1], gap=2<br>=> migrate: [2/2/2], gap=0<br>Pool rebalanced in 2 refreshes"]:::note
NOTE -.-> REBALANCE
REBALANCE --> after

subgraph after["After (2 refreshes)"]
    subgraph Picodata_after["Picodata"]
        p21[("ins 1")]:::online
        p22[("ins 2")]:::online
        p23[("ins 3")]:::online
    end

    subgraph Pool_after["Pool: max_size=6"]
        c21["conn x2"] --> p21
        c22["conn x2"] --> p22
        c23["conn x2"] --> p23
    end
end

`topology_update_interval` `property` `writable`

Interval in seconds between topology refresh cycles for Topology Tracker. Pass None at initialization to disable periodic updates. With updates disabled topology is None and reconcile never runs.

A new interval must be at least 1s. After open(), changing it triggers an immediate topology refresh, so the new interval applies without waiting for the previous timeout. A refresh already in progress is allowed to finish. Before open(), the value is saved for startup.

Use disable_topology_tracking() to stop it. Assigning the current value is a no-op.

Raises:

Type	Description
`ValueError`	If the new interval is less than 1s.
`RuntimeError`	If the pool is closing or closed, or if the new value enables or disables topology tracking.

`close(timeout=None)`

Close the pool: close free connections and wait for the used ones to come back.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Maximum graceful shutdown time for the call that starts pool shutdown. If None, the default shutdown timeout of 1 hour is used.	`None`

Only the first close() call starts shutdown and determines the shutdown timeout. Concurrent close() calls wait for the same shutdown operation and do not change the first timeout. If that timeout expires, any remaining connections are closed anyway, even the ones still checked out.

The timeout bounds the waiting shutdown does, not the blocking calls it makes: closing a connection cannot be interrupted, so a close that hangs on an unresponsive node still runs to completion.

Note

This should be called during application shutdown to clean up resources. Do not call it from a thread that still holds a pooled connection: shutdown waits for that connection to be released, so it would block until the timeout. Use terminate() to close the pool without waiting.

Source code in picopyn/synchronous/pool.py

def close(self, timeout: float | None = None) -> None:
    """
    Close the pool: close free connections and wait for the used ones to come back.

    Args:
        timeout: Maximum graceful shutdown time for the call that starts pool
            shutdown. If None, the default shutdown timeout of 1 hour is used.

    Only the first close() call starts shutdown and determines the shutdown timeout.
    Concurrent close() calls wait for the same shutdown operation and do not change
    the first timeout. If that timeout expires, any remaining connections are closed
    anyway, even the ones still checked out.

    The timeout bounds the waiting shutdown does, not the blocking calls it makes:
    closing a connection cannot be interrupted, so a close that hangs on an
    unresponsive node still runs to completion.

    Note:
        This should be called during application shutdown to clean up resources.
        Do not call it from a thread that still holds a pooled connection: shutdown
        waits for that connection to be released, so it would block until the timeout.
        Use [terminate()][picopyn.synchronous.pool.Pool.terminate] to close the pool
        without waiting.
    """
    with self._condition:
        close_event = self._close_event
        first_call = close_event is None
        if close_event is None:
            close_event = Event()
            self._close_event = close_event
            # let acquire waiters fail fast instead of polling until their timeout
            self._condition.notify_all()

    if not first_call:
        # somebody else already started shutdown, wait for the same operation
        close_event.wait()
        return

    effective_timeout = _DEFAULT_CLOSE_TIMEOUT_SEC if timeout is None else timeout
    try:
        self._close_gracefully(time.monotonic() + effective_timeout)
    finally:
        self.terminate()

`connection(timeout=None)`

Return a context manager for a pooled connection.

If no connections are available, this method will wait until one is released. Unlike the async pool, the sync public API intentionally does not expose manual acquire/release methods.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Maximum time to wait for a connection if the pool is exhausted. If None, a default timeout is used.	`None`

Returns:

Type	Description
`_PoolConnectionContext`	A context manager that returns a sync `Connection` from `with`.

Examples:

with pool.connection() as conn:
    cursor = conn.cursor()
    cursor.execute("UPDATE ...")

Source code in picopyn/synchronous/pool.py

def connection(self, timeout: float | None = None) -> _PoolConnectionContext:
    """Return a context manager for a pooled connection.

    If no connections are available, this method will wait until one is released.
    Unlike the async pool, the sync public API intentionally does not expose
    manual acquire/release methods.

    Args:
        timeout: Maximum time to wait for a connection if the pool is exhausted.
            If None, a default timeout is used.

    Returns:
        A context manager that returns a sync `Connection` from `with`.

    Examples:
        ```python
        with pool.connection() as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE ...")
        ```
    """
    return _PoolConnectionContext(self, timeout)

`connection_by_instance_uuid(instance_uuid, timeout=None)`

Return a context manager for a pooled connection with instance_uuid.

This bypasses balance_strategy. If the instance is currently ejected while another usable instance remains non-ejected, TimeoutError is raised immediately. Otherwise, if no free connection with the given UUID becomes available before the timeout (see Pool.routing_acquire_timeout), TimeoutError is raised.

Source code in picopyn/synchronous/pool.py

def connection_by_instance_uuid(
    self,
    instance_uuid: str,
    timeout: float | None = None,
) -> _PoolConnectionContext:
    """Return a context manager for a pooled connection with `instance_uuid`.

    This bypasses `balance_strategy`. If the instance is currently ejected
    while another usable instance remains non-ejected, `TimeoutError` is raised
    immediately. Otherwise, if no free connection with the given UUID
    becomes available before the timeout (see
    [Pool.routing_acquire_timeout][picopyn.synchronous.pool.Pool]), `TimeoutError` is raised.
    """
    return _PoolConnectionContext(self, timeout, instance_uuid)

`connection_by_sharding_key(sharding_key, timeout=None)`

Return a connection to the master selected by sharding_key.

This is explicit routing: if the current topology cannot resolve the key or no matching connection becomes available, the method raises instead of falling back to a random pool connection.

Parameters:

Name	Type	Description	Default
`sharding_key`	`ShardingKey`	Key created by a table's sharding key factory.	required
`timeout`	`float \| None`	Maximum time to wait for a matching connection to become free.	`None`

Returns:

Type	Description
`_PoolConnectionContext`	A context manager that can be used with `with`.

Raises:

Type	Description
`TypeError`	If `sharding_key` is not a `ShardingKey`.
`RuntimeError`	If topology tracking is disabled or the key cannot be resolved to a known master.
`TimeoutError`	If the selected master is ejected while another usable instance remains non-ejected, or no matching connection becomes available before the timeout.

Source code in picopyn/synchronous/pool.py

def connection_by_sharding_key(
    self,
    sharding_key: ShardingKey,
    timeout: float | None = None,
) -> _PoolConnectionContext:
    """Return a connection to the master selected by `sharding_key`.

    This is explicit routing: if the current topology cannot resolve the
    key or no matching connection becomes available, the method raises
    instead of falling back to a random pool connection.

    Args:
        sharding_key: Key created by a table's sharding key factory.
        timeout: Maximum time to wait for a matching connection to become free.

    Returns:
        A context manager that can be used with `with`.

    Raises:
        TypeError: If `sharding_key` is not a `ShardingKey`.
        RuntimeError: If topology tracking is disabled or the key cannot
            be resolved to a known master.
        TimeoutError: If the selected master is ejected while another usable
            instance remains non-ejected, or no matching connection becomes available before
            the timeout.
    """
    if not isinstance(sharding_key, ShardingKey):
        raise TypeError("sharding_key must be a ShardingKey")
    return self.connection_by_tier_and_bucket_id(
        sharding_key.tier,
        sharding_key.bucket_id,
        timeout,
    )

`connection_by_tier_and_bucket_id(tier, bucket_id, timeout=None)`

Acquire a pooled connection to the master owning bucket_id in tier.

The method uses the current topology to resolve tier + bucket_id to the owning replicaset's master instance, then acquires a free pooled connection with that instance UUID.

If instance_role="replica", topology tracking is disabled, or the current topology cannot map the bucket to a master instance, RuntimeError is raised. If the master is currently ejected while another usable instance remains non-ejected, TimeoutError is raised immediately. Otherwise, if the master is known but no matching free connection appears before the timeout, TimeoutError is raised.

Parameters:

Name	Type	Description	Default
`tier`	`str`	Name of the tier the bucket belongs to.	required
`bucket_id`	`int`	Bucket id to resolve to its owning replicaset's master instance.	required
`timeout`	`float \| None`	Maximum time to wait for a matching connection to become free. If None, a default timeout is used.	`None`

Returns:

Type	Description
`_PoolConnectionContext`	A context manager that can be used with `with` to acquire a connection.

Examples:

with pool.connection_by_tier_and_bucket_id("default", 42) as conn:
    conn.execute("UPDATE ...")

Source code in picopyn/synchronous/pool.py

def connection_by_tier_and_bucket_id(
    self,
    tier: str,
    bucket_id: int,
    timeout: float | None = None,
) -> _PoolConnectionContext:
    """Acquire a pooled connection to the master owning `bucket_id` in `tier`.

    The method uses the current topology to resolve
    `tier + bucket_id` to the owning replicaset's master instance, then
    acquires a free pooled connection with that instance UUID.

    If `instance_role="replica"`, topology tracking is disabled, or the current topology cannot map
    the bucket to a master instance, `RuntimeError` is raised.
    If the master is currently ejected while another usable instance remains non-ejected,
    `TimeoutError` is raised immediately. Otherwise, if the master is known
    but no matching free connection appears before the timeout, `TimeoutError`
    is raised.

    Args:
        tier: Name of the tier the bucket belongs to.
        bucket_id: Bucket id to resolve to its owning replicaset's master instance.
        timeout: Maximum time to wait for a matching connection to become free.
            If None, a default timeout is used.

    Returns:
        A context manager that can be used with `with` to acquire a connection.

    Examples:
        ```python
        with pool.connection_by_tier_and_bucket_id("default", 42) as conn:
            conn.execute("UPDATE ...")
        ```
    """
    if not tier:
        raise ValueError("tier must be a non-empty string")

    if self._instance_role == "replica":
        raise RuntimeError(
            "Cannot acquire connection by tier and bucket_id: "
            "instance_role='replica' excludes current replicaset masters from the pool"
        )

    tracker = self._topology_tracker
    if self._topology_update_interval is None or tracker is None:
        raise RuntimeError(
            "Cannot acquire connection by tier and bucket_id: topology tracking is disabled"
        )

    master_uuid = tracker.find_bucket_master_uuid(tier, bucket_id)
    if master_uuid is None:
        raise RuntimeError(
            "Cannot acquire connection by tier and bucket_id: no known master for "
            f"tier {tier!r}, bucket_id {bucket_id!r}"
        )

    return self.connection_by_instance_uuid(master_uuid, timeout)

`create_sharding_key_factory(table_name)`

Create a reusable sharding key factory for a Picodata table.

The method reads the table's distribution schema once. Creating keys from the returned factory is local and does not perform database requests.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Exact Picodata table name.	required

Returns:

Type	Description
`ShardingKeyFactory`	A factory configured with the table's sharding fields, tier and bucket count.

Raises:

Type	Description
`ValueError`	If `table_name` is empty or the table distribution metadata is invalid or unsupported.
`RuntimeError`	If topology tracking is disabled, table metadata is not found, or tier metadata is unavailable.

Source code in picopyn/synchronous/pool.py

def create_sharding_key_factory(self, table_name: str) -> ShardingKeyFactory:
    """Create a reusable sharding key factory for a Picodata table.

    The method reads the table's distribution schema once. Creating keys
    from the returned factory is local and does not perform database requests.

    Args:
        table_name: Exact Picodata table name.

    Returns:
        A factory configured with the table's sharding fields, tier and bucket count.

    Raises:
        ValueError: If `table_name` is empty or the table distribution
            metadata is invalid or unsupported.
        RuntimeError: If topology tracking is disabled, table metadata is
            not found, or tier metadata is unavailable.
    """
    if not table_name:
        raise ValueError("table_name must be a non-empty string")

    tracker = self._topology_tracker
    if self._topology_update_interval is None or tracker is None:
        raise RuntimeError(
            "Cannot create sharding key factory: topology tracking is disabled "
            "or the pool is not open"
        )

    row = self.fetchrow(SYNC_TABLE_SHARDING_QUERY, (table_name,))
    if row is None:
        raise RuntimeError(f"Table {table_name!r} metadata not found")

    table_info = _parse_table_sharding_info(table_name, row[0], row[1])
    bucket_count = tracker.get_bucket_count(table_info.tier)
    if bucket_count is None:
        tracker.refresh_now()
        bucket_count = tracker.get_bucket_count(table_info.tier)
    if bucket_count is None:
        raise RuntimeError(f"Tier metadata is not found for table {table_name!r}")

    return ShardingKeyFactory(table_info, bucket_count, DecimalFormat.TEXT)

`disable_instance_ejection()`

Disable error-based instance ejection.

Existing ejections and recorded failures are discarded. Connections stay open and immediately return to pool selection. The operation is idempotent. See instance ejection.

Raises:

Type	Description
`RuntimeError`	If the pool is closing or closed.

Source code in picopyn/synchronous/pool.py

def disable_instance_ejection(self) -> None:
    """Disable error-based instance ejection.

    Existing ejections and recorded failures are discarded. Connections stay
    open and immediately return to pool selection. The operation is
    idempotent. See
    [instance ejection][picopyn.synchronous.pool.Pool.instance_ejection_enabled].

    Raises:
        RuntimeError: If the pool is closing or closed.
    """
    with self._condition:
        self._check_open()
        if self._instance_ejection_state is None:
            return
        self._instance_ejection_state = None
        self._condition.notify_all()

`disable_query_metadata()`

Disable query metadata and wait for its technical connection to close.

Sets query_metadata_cache_size to None and cancels pending metadata requests. New execute() calls use ordinary pool selection, and get_query_metadata() raises RuntimeError. Explicit routing by sharding key remains available while topology tracking is enabled.

Can be called before open(). Repeated calls wait for the same shutdown. This call blocks until the running prepare and connection close finish.

Raises:

Type	Description
`RuntimeError`	If the pool is closing or closed.

Source code in picopyn/synchronous/pool.py

def disable_query_metadata(self) -> None:
    """Disable query metadata and wait for its technical connection to close.

    Sets `query_metadata_cache_size` to None and cancels pending metadata
    requests. New execute() calls use ordinary pool selection, and
    get_query_metadata() raises RuntimeError. Explicit routing by sharding
    key remains available while topology tracking is enabled.

    Can be called before open(). Repeated calls wait for the same shutdown.
    This call blocks until the running prepare and connection close finish.

    Raises:
        RuntimeError: If the pool is closing or closed.
    """
    with self._condition:
        self._check_open()
        if self._query_metadata_cache_size is None:
            shutdown = self._metadata_disable
            if shutdown is None:
                return
            first_call = False
        else:
            self._query_metadata_cache_size = None
            service = self._query_metadata_service
            if service is None:
                return
            shutdown = Future()
            self._metadata_disable = shutdown
            first_call = True
    assert shutdown is not None
    if first_call:
        assert service is not None
        try:
            self._stop_query_metadata(service)
        except BaseException as e:
            shutdown.set_exception(e)
        else:
            shutdown.set_result(None)
    try:
        shutdown.result()
    finally:
        with self._condition:
            if self._metadata_disable is shutdown and shutdown.done():
                self._metadata_disable = None

`disable_topology_tracking()`

Disable topology tracking and wait for its technical connection to close.

Sets topology_update_interval and topology to None. New refreshes, reconcile and topology-based routing stop; user connections are kept. Previously scheduled evictions still take effect on release.

Can be called before open(). Repeated calls wait for the same shutdown. This call blocks until the running refresh and source close finish.

Raises:

Type	Description
`RuntimeError`	If the pool is closing or closed.

Source code in picopyn/synchronous/pool.py

def disable_topology_tracking(self) -> None:
    """Disable topology tracking and wait for its technical connection to close.

    Sets `topology_update_interval` and `topology` to None. New refreshes,
    reconcile and topology-based routing stop; user connections are kept.
    Previously scheduled evictions still take effect on release.

    Can be called before open(). Repeated calls wait for the same shutdown.
    This call blocks until the running refresh and source close finish.

    Raises:
        RuntimeError: If the pool is closing or closed.
    """
    with self._condition:
        self._check_open()
        if self._topology_update_interval is None:
            shutdown = self._topology_disable
            if shutdown is None:
                return
            first_call = False
        else:
            self._topology_update_interval = None
            tracker = self._topology_tracker
            if tracker is None:
                return
            shutdown = Future()
            self._topology_disable = shutdown
            first_call = True
    assert shutdown is not None
    if first_call:
        assert tracker is not None
        try:
            self._stop_topology_tracking(tracker)
        except BaseException as e:
            shutdown.set_exception(e)
        else:
            shutdown.set_result(None)
    try:
        shutdown.result()
    finally:
        with self._condition:
            if self._topology_disable is shutdown and shutdown.done():
                self._topology_disable = None

`execute(query, parameters=None)`

Executes a query that does not return rows (e.g. INSERT, UPDATE, DELETE).

The driver supports shard-aware routing for parameterized 1-row INSERT. It is based on two mechanisms: cache service for query meta and topology tracking that allows us to calculate bucket id and use it to choose Picodata replicaset master.

So, the shard-aware routing requires:

metadata cache service enabled (default; see Pool.query_metadata_cache_size)
topology tracking enabled (see Pool.topology_update_interval)
instance_role is not "replica"
Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
execute query more than once

Read about cache warming.

Parameters:

Name	Type	Description	Default
`query`	`QueryNoTemplate`	The SQL query string.	required
`parameters`	`Params \| None`	Optional DB-API parameters for the SQL query.	`None`

Returns:

Type	Description
`str`	The result of the query execution.

Raises:

Type Description

ProgrammingError

If the statement prepared on the connection was invalidated by a schema change (Picodata SQLSTATE 42999). The connection drops its prepared statements and the cached query meta is evicted, so the caller decides how to react (execute the query again, execute another query, etc.). After DDL the connection raises the error only when it executes a prepared statement, so it depends on the prepare_threshold setting (see prepare_threshold in Pool).

Error

Other query failures, see errors.

Source code in picopyn/synchronous/pool.py

def execute(self, query: QueryNoTemplate, parameters: Params | None = None) -> str:
    """Executes a query that does not return rows (e.g. INSERT, UPDATE, DELETE).

    The driver supports shard-aware routing for parameterized 1-row INSERT. It is based
    on two mechanisms:
    [cache service][picopyn.synchronous.metadata_service.QueryMetadataService] for query
    meta and [topology tracking][picopyn.synchronous.topology_tracker.TopologyTracker]
    that allows us to calculate bucket id and use it to choose Picodata replicaset master.

    So, the shard-aware routing requires:

    - metadata cache service enabled (default; see
      [Pool.query_metadata_cache_size][picopyn.synchronous.pool.Pool])
    - topology tracking enabled (see [Pool.topology_update_interval][picopyn.synchronous.pool.Pool])
    - `instance_role` is not `"replica"`
    - Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
    - execute query more than once

    Read about [cache warming][picopyn.synchronous.metadata_service.QueryMetadataService.request_if_missing].

    Args:
        query: The SQL query string.
        parameters: Optional DB-API parameters for the SQL query.

    Returns:
        The result of the query execution.

    Raises:
        ProgrammingError: If the statement prepared on the connection was invalidated
            by a schema change (Picodata SQLSTATE `42999`). The connection drops its
            prepared statements and the cached query meta is evicted, so the caller
            decides how to react (execute the query again, execute another query, etc.).
            After DDL the connection raises the error only when it executes a prepared
            statement, so it depends on the `prepare_threshold` setting (see
            `prepare_threshold` in [Pool][picopyn.synchronous.pool.Pool]).
        Error: Other query failures, see [errors][picopyn.synchronous.errors.Error].
    """
    # prepare and cache query meta only for parameterized queries
    conn = None
    if parameters and isinstance(query, str):
        conn = self._acquire_routed_connection(query, parameters)

    if conn is None:
        conn = self._acquire_raw()

    try:
        cursor = conn.cursor()
        cursor.execute(query, parameters)
        status = cursor.statusmessage
        if status is None:
            raise RuntimeError("Query executed without status message")
        return status
    except Exception as e:
        # TODO retry execution to hide from user the invalidation error
        # (we have to re-calculate meta and cache it, but in the time we can execute
        # the query without routing)
        service = self._query_metadata_service
        if (
            isinstance(query, str)
            and self._query_metadata_cache_size is not None
            and service is not None
            and is_stmt_invalidated_error(e)
        ):
            service.evict(query)
        raise
    finally:
        self._release(conn)

`fetch(query, parameters=None)`

Executes a query and fetches all resulting rows.

Parameters:

Name	Type	Description	Default
`query`	`QueryNoTemplate`	The SQL query string.	required
`parameters`	`Params \| None`	Optional DB-API parameters for the SQL query.	`None`

Returns:

Type	Description
`list[tuple[object, ...]]`	A list of rows returned by the query.

Raises:

Type	Description
`Error`	If the query fails, see errors.

Source code in picopyn/synchronous/pool.py

def fetch(
    self,
    query: QueryNoTemplate,
    parameters: Params | None = None,
) -> list[tuple[object, ...]]:
    """Executes a query and fetches all resulting rows.

    Args:
        query: The SQL query string.
        parameters: Optional DB-API parameters for the SQL query.

    Returns:
        A list of rows returned by the query.

    Raises:
        Error: If the query fails, see [errors][picopyn.synchronous.errors.Error].
    """
    # TODO: add cache usage when Picodata will support it for DQL
    # https://git.picodata.io/core/picodata/-/issues/2226
    with self.connection() as conn:
        try:
            cursor = conn.cursor()
            cursor.execute(query, parameters)
            return cursor.fetchall()
        except Exception as e:
            # TODO retry execution to hide from user the invalidation error
            service = self._query_metadata_service
            if (
                isinstance(query, str)
                and self._query_metadata_cache_size is not None
                and service is not None
                and is_stmt_invalidated_error(e)
            ):
                service.evict(query)
            raise

`fetchrow(query, parameters=None)`

Executes a query and fetches a single row (first row).

Parameters:

Name	Type	Description	Default
`query`	`QueryNoTemplate`	The SQL query string.	required
`parameters`	`Params \| None`	Optional DB-API parameters for the SQL query.	`None`

Returns:

Type	Description
`tuple[object, ...] \| None`	The first row returned by the query, or None.

Raises:

Type	Description
`Error`	If the query fails, see errors.

Source code in picopyn/synchronous/pool.py

def fetchrow(
    self,
    query: QueryNoTemplate,
    parameters: Params | None = None,
) -> tuple[object, ...] | None:
    """Executes a query and fetches a single row (first row).

    Args:
        query: The SQL query string.
        parameters: Optional DB-API parameters for the SQL query.

    Returns:
        The first row returned by the query, or None.

    Raises:
        Error: If the query fails, see [errors][picopyn.synchronous.errors.Error].
    """
    # TODO: add cache usage when Picodata will support it for DQL
    # https://git.picodata.io/core/picodata/-/issues/2226
    with self.connection() as conn:
        try:
            cursor = conn.cursor()
            cursor.execute(query, parameters)
            return cursor.fetchone()
        except Exception as e:
            # TODO retry execution to hide from user the invalidation error
            service = self._query_metadata_service
            if (
                isinstance(query, str)
                and self._query_metadata_cache_size is not None
                and service is not None
                and is_stmt_invalidated_error(e)
            ):
                service.evict(query)
            raise

`get_query_metadata(query)`

Return Picodata's distribution key metadata for query, blocking until it's resolved.

Raises:

Type	Description
`RuntimeError`	the metadata service is not running -- either it is disabled (see Pool.query_metadata_cache_size) or the pool is not opened.

The pool caches query metadata -- see below for details:

Expand pool initialization in cache meta context

---
title: Query metadata -- pool initialization in cache meta context
---

flowchart TD
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555


INIT(["Pool init"]):::neutral

INIT --> CONNECT["Fill the pool with ordinary<br>connections to picodata"] --> Pool1

NOTE_CONN>"connection options require<br>statement invalidation<br>signal only"]:::note
NOTE_CONN -.-> CONNECT

CONNECT_DETAILS_NOTE>"Open algorithm details"]:::note
click CONNECT_DETAILS_NOTE "#picopyn.synchronous.Pool.open" "Open algorithm details"
CONNECT_DETAILS_NOTE -.-> CONNECT

subgraph Pool1["Pool"]
    direction TB

    CONN11["conn to inst1"] --> PICO11[(Picodata)]
    CONN12["..."] --> PICO12[(Picodata)]
    CONN1N["conn to instN"] --> PICO1N[(Picodata)]
end

Pool1 --> CONNECT_SERV["create empty query meta cache<br>and service connection<br>for meta listener"]

NOTE_DISABLED>"skipped entirely when<br>service disabled: no cache,<br>no service connection"]:::note
NOTE_DISABLED -.-> CONNECT_SERV

NOTE_CONN_SERV>"connection options require<br>statement invalidation signal<br>and distribution key metadata"]:::note
NOTE_CONN_SERV -.-> CONNECT_SERV

CONNECT_SERV --> LIST["add picodata notice listener"]

LIST --> Pool

subgraph Pool["Pool"]
    direction TB

    CONN1["conn1 for<br>user usage"] --> PICO1[(Picodata1)]
    CONN2["..."] --> PICO2[(...)]
    CONNN["connN for<br>user usage"] --> PICON[(PicodataN)]

    PIC(["can be any of"]) -.- CONNSERV["service connection<br>with Notice listener"]
    PICO1 -.- PIC
    PICO2 -.- PIC
    PICON -.- PIC

    CACHE[/"Query meta cache"/]
end

---
title: Query metadata -- usage lookup
---

sequenceDiagram
    participant App as Application
    box rgba(128,128,128,0.15) Picopyn
        participant Driver as Core
        participant Pool as Connection pool
        participant Meta as Service connection
    end
    participant Server as Picodata

    App->>Driver: pool.execute(query, params)

    alt First call: cache miss
        Driver->>Pool: Any free connection
    else Repeated call: cache hit
        Driver->>Driver: query meta + params -> bucket_id -> node
        Driver->>Pool: Connection to that node
    end
    Pool-->>Driver: Connection

    Driver->>Server: Bind + Execute
    Server-->>Driver: Result
    Driver-->>App: Result

    rect rgba(128,128,128,0.15)
        Note over Driver,Server: In the background, only after a cache miss
        Driver->>Meta: query
        Meta->>Server: Parse(query)
        Server-->>Meta: NoticeResponse
        Meta-->>Driver: query meta
        Driver->>Driver: LRU cache
    end

Source code in picopyn/synchronous/pool.py

def get_query_metadata(self, query: str) -> PreparedStatementMetadata | None:
    """Return Picodata's [distribution key metadata][picopyn.query_metadata.is_stmt_invalidated_error]
    for `query`, blocking until it's resolved.

    Raises:
        RuntimeError: the metadata service is not running -- either it is disabled
            (see [Pool.query_metadata_cache_size][picopyn.synchronous.pool.Pool])
            or the pool is not opened.

    The pool caches query metadata -- see below for details:

    <details><summary>Expand pool initialization in cache meta context</summary>
    --8<-- "_query_metadata_pool_init.md"
    </details>

    --8<-- "_query_metadata_usage_lookup.md"
    """
    with self._condition:
        self._check_open()

    if self._query_metadata_cache_size is None:
        raise RuntimeError("query metadata service is not running")
    service = self._query_metadata_service
    if service is None:
        raise RuntimeError("query metadata service is not running")

    return service.get_query_metadata(query)

`open()`

Open the pool by creating up to max_size connections.

This should be called before using the pool to ensure connections are available.

---
title: Pool initialization
---

flowchart TD
%% === Initialization ===
classDef success fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef error fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

START([Start]):::neutral
DONE([Pool ready]):::success
ERROR([Initialization failed]):::error
CFG[/"Pool settings<br>· DSN string<br>· pool size<br>· forbidden tiers<br>· cluster discovery mode"/]

USAGE_DETAILS_NOTE>"Return to pool lifecycle details"]:::note
click USAGE_DETAILS_NOTE "#picopyn.synchronous.Pool" "Return to pool lifecycle details"

%% === Data objects ===
POOL[[Connection pool]]
B_DSN[[DSN node list]]
B_SYS[(Picodata system tables)]
D_DSN[[DSN node list]]
D_SYS[(Picodata system tables)]
CAND[[Candidate node list]]

%% === Actions ===

START --> CFG --> PARSE[Parse DSN string to DSN node list] --> DISCOVERY{Cluster discovery enabled?}

DISCOVERY -->|yes| D_FILTER[Use DSN nodes to connect to<br>Picodata cluster and retrieve<br>eligible online nodes<br>as a candidate node list]
DISCOVERY -->|no| B_FILTER[Filter DSN nodes: use only<br>eligible online nodes]

subgraph DiscoverMode["Cluster discovery mode"]
    D_FILTER --> D_FILL_POOL[Try to fill pool with connections<br>to candidates up to pool size]

    D_DETAILS_NOTE>"Discovery mode details"]:::note
    click D_DETAILS_NOTE "#pool-init-discovery-enabled" "Discovery mode details"

    D_FILTER <-. write .-> CAND
    CAND <-. read .-> D_FILL_POOL
end

subgraph BootstrapMode["Bootstrap mode"]
    B_FILTER --> B_FILL_POOL[Try to fill pool with connections<br>to DSN nodes up to pool size]

    B_DETAILS_NOTE>"Bootstrap mode details"]:::note
    click B_DETAILS_NOTE "#pool-init-discovery-disabled" "Bootstrap mode details"
end

B_DSN <-. read .-> B_FILTER
B_SYS <-. read .-> B_FILTER
B_FILL_POOL <-. write .-> POOL

D_DSN <-. read .-> D_FILTER
D_SYS <-. read .-> D_FILTER
D_FILL_POOL <-. write .-> POOL

POOL <-.-> POOL_CHECK{Pool full?}
POOL_CHECK -->|no| ERROR
POOL_CHECK -->|yes| DONE

See open details in discovery mode

---
title: Pool initialization with cluster discovery enabled
---

flowchart TD

%% === Initialization ===
classDef success fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef error fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

START([Start]):::neutral
DONE([Pool ready]):::success
D_ERROR_CAND([Initialization failed]):::error
ERROR([Initialization failed]):::error
CFG[/"Pool settings<br>· DSN string<br>· max_size<br>· forbidden tiers<br>· cluster discovery enabled"/]

GENERAL_DETAILS_NOTE>"Return to general algorithm"]:::note
click GENERAL_DETAILS_NOTE "#picopyn.synchronous.Pool.open" "Return to general algorithm"

%% === Data objects ===
D_DSN[[DSN node list]]
CAND[[Candidate node list]]
DISCOVERY_POOL[[Connection pool]]
SYS[(Picodata system tables)]

%% === Discovery mode ===
START --> CFG --> PARSE[Parse DSN string to<br>DSN node list] --> DiscoverNodes
PARSE <-. write .-> D_DSN

D_DiscoverNodes_NOTE>"iterate over DSN nodes<br>to establish a temporary<br>connection and retrieve the<br>list of eligible online nodes"]:::note
D_DiscoverNodes_NOTE -.-> DiscoverNodes
D_DSN <-. read .-> D_RECEIVE[Try to connect and receive cluster nodes]
subgraph DiscoverNodes["Discover cluster nodes"]
    SYS <-. read .-> D_RECEIVE
    D_RECEIVE -->|could not select candidates| D_ERROR_CAND
    D_RECEIVE <-. write .-> CAND
end
DiscoverNodes --> D_PoolFilling

D_PoolFilling_NOTE>"iterate over candidate nodes<br>until either:<br>- the list is empty<br>- max_size connections<br>have been established"]:::note
D_PoolFilling_NOTE -.-> D_PoolFilling
CAND <-. read .-> D_CONNECT_NODE[Connect to candidate node]
subgraph D_PoolFilling["Pool filling"]
    D_CONNECT_NODE -->|connection error| D_EXCLUDE[Exclude candidate node]
    D_CONNECT_NODE -->|connected| D_SAVE[Save opened connection to pool]
    D_SAVE <-. write .-> DISCOVERY_POOL
end

D_PoolFilling --> POOL_CHECK{Pool full?}
POOL_CHECK -->|no| ERROR
POOL_CHECK -->|yes| DONE

See open details in bootstrap mode

---
title: Pool initialization with cluster discovery disabled
---

flowchart TD

%% === Initialization ===
classDef success fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
classDef error fill:#ffebee,stroke:#f44336,color:#b71c1c
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

START([Start]):::neutral
DONE([Pool ready]):::success
ERROR([Initialization failed]):::error
CFG[/"Pool settings<br>· DSN string<br>· pool size<br>· forbidden tiers<br>· cluster discovery disabled"/]

GENERAL_DETAILS_NOTE>"Return to general algorithm"]:::note
click GENERAL_DETAILS_NOTE "#picopyn.synchronous.Pool.open" "Return to general algorithm"

%% === Data objects ===
B_DSN[[DSN node list]]
BOOTSTRAP_POOL[[Connection pool]]
SYS[(Picodata system tables)]

%% === Bootstrap mode ===
START --> CFG --> PARSE[Parse DSN string to DSN node list] --> B_PoolFilling
PARSE <-. write .-> B_DSN
B_DSN <-. read .-> B_CONNECT_NODE[Connect to DSN node]

B_PoolFilling_NOTE>"iterate over DSN nodes until either:<br>- the list is empty<br>- max size connections<br>have been established"]:::note
B_PoolFilling_NOTE -.-> B_PoolFilling
subgraph B_PoolFilling["Pool filling"]
    direction TB
    SYS <-. read .-> B_CHECK_NODE_TIER[Check node state, tier, and role]
    B_CONNECT_NODE --> B_CHECK_NODE_TIER
    B_CHECK_NODE_TIER -->|node offline, tier forbidden,<br>or does not match instance role| B_EXCLUDE[Exclude DSN node]
    B_CHECK_NODE_TIER -->|node online, tier allowed,<br>and matches instance role| B_SAVE[Save connection to pool]
    B_SAVE <-. write .-> BOOTSTRAP_POOL
end

B_PoolFilling --> POOL_CHECK{Pool full?}
POOL_CHECK -->|no| ERROR
POOL_CHECK -->|yes| DONE

Raises:

Type	Description
`RuntimeError`	If the pool is closing or already closed, if discovery fails, or if fewer than `max_size` connections could be established.

Source code in picopyn/synchronous/pool.py

def open(self) -> None:
    """
    Open the pool by creating up to `max_size` connections.

    This should be called before using the pool to ensure connections are available.

    --8<-- "_pool_init.md"

    <details id="pool-init-discovery-enabled"><summary>See open details in discovery mode</summary>
    --8<-- "_pool_init_discovery_enabled.md"
    </details>

    <details id="pool-init-discovery-disabled"><summary>See open details in bootstrap mode</summary>
    --8<-- "_pool_init_discovery_disabled.md"
    </details>

    Raises:
        RuntimeError: If the pool is closing or already closed, if discovery fails,
            or if fewer than `max_size` connections could be established.
    """
    with self._condition:
        self._check_open()

        if self._current_size() >= self._max_size:
            return

        # if node discovery is enabled, then connect to all alive picodata instances
        # (if they fit within the max_size limit)
        if self._enable_discovery:
            self._fill_pool_from_discovery()
        else:
            self._fill_pool_from_bootstrap_dsns()

        conn_count = self._current_size()
        if conn_count < self._max_size:
            while self._pool:
                conn = self._pool.popleft()
                try:
                    conn.close()
                except Exception as e:
                    logger.warning("Pool connect cleanup: could not close connection: %s", e)
            raise RuntimeError(
                f"Failed to initialize connection pool: only {conn_count} "
                f"out of {self._max_size} connections established for DSN "
                f"{self._redacted_dsn}"
            )

        # rotate the pool to randomize the order of connections.
        # this helps to distribute the initial load more evenly across nodes
        # when using round-robin or when multiple clients start simultaneously.
        shift = random.randint(0, len(self._pool) - 1)
        self._pool.rotate(shift)

        logger.info("Pool initialized with %d connections", conn_count)
        if self._topology_update_interval is not None and self._topology_tracker is None:
            self._topology_tracker = TopologyTracker(
                PollingTopologySource(
                    lambda topology: self._get_technical_connection(topology=topology)
                ),
                self._topology_update_interval,
            )
            # register the pool's reconcile callback so connections are updated
            # to match the topology after each refresh
            self._topology_tracker.start(on_refresh=self._reconcile_pool_with_topology)
            logger.debug(
                "Pool topology tracker started with update interval %ss",
                self._topology_update_interval,
            )

        if self._query_metadata_cache_size is not None and self._query_metadata_service is None:
            # technical connection + cache for query meta discovery, decoupled from the
            # connections below that actually run queries
            self._query_metadata_service = QueryMetadataService(
                connect=lambda on_query_metadata: self._get_technical_connection(
                    on_query_metadata
                ),
                cache_size=self._query_metadata_cache_size,
            )

            self._query_metadata_service.start()
            logger.debug(
                "Pool query-metadata service started with cache size %s",
                self._query_metadata_cache_size,
            )
            if self._instance_role == "replica" and self._topology_update_interval is not None:
                logger.warning(
                    "Automatic shard-aware routing is disabled for instance_role='replica': "
                    "current replicaset masters are excluded from user pool connections"
                )
    return

`terminate()`

Close all connections owned by the pool without graceful close.

Used connections are closed right away. Calling it while a close() is running finishes that shutdown instead of starting another one. On an already closed pool it does nothing.

Concurrent terminate() calls run one after another: the first one does the work, the rest return once it is done.

Source code in picopyn/synchronous/pool.py

def terminate(self) -> None:
    """Close all connections owned by the pool without graceful close.

    Used connections are closed right away.
    Calling it while a `close()` is running finishes that shutdown instead
    of starting another one. On an already closed pool it does nothing.

    Concurrent terminate() calls run one after another: the first one does the
    work, the rest return once it is done.
    """
    # the whole body is serialized, so `_close_event` cannot be set by another
    # shutdown while this one still has connections to close
    with self._shutdown_lock:
        with self._condition:
            close_event = self._close_event
            if close_event is not None and close_event.is_set():
                return
            if close_event is None:
                close_event = Event()
                self._close_event = close_event

            conns = [*self._pool, *self._used]
            self._pool.clear()
            # dropping `_used` also releases a close() waiting for these connections,
            # the way async terminate() cancels the running close task
            self._used.clear()
            self._condition.notify_all()

        tracker = self._topology_tracker
        if tracker is not None:
            try:
                tracker.terminate()
            except Exception as e:
                logger.warning("Pool terminate: could not terminate topology tracker: %s", e)

        service = self._query_metadata_service
        if service is not None:
            try:
                service.terminate()
            except Exception as e:
                logger.warning(
                    "Pool terminate: could not terminate query-metadata service: %s", e
                )

        self._close_connections(conns)

        with self._condition:
            close_event.set()
            self._condition.notify_all()

Errors

`Warning`

Bases: Exception

Exception raised for important warnings.

Source code in picopyn/synchronous/errors.py

class Warning(Exception):  # noqa: N818 - required by PEP 249
    """Exception raised for important warnings."""

`Error`

Bases: Exception

Base class for all synchronous DB-API errors.

Source code in picopyn/synchronous/errors.py

class Error(Exception):
    """Base class for all synchronous DB-API errors."""

`InterfaceError`

Bases: Error

Error related to the database interface rather than the database itself.

Source code in picopyn/synchronous/errors.py

class InterfaceError(Error):
    """Error related to the database interface rather than the database itself."""

`DatabaseError`

Bases: Error

Error related to the database.

Source code in picopyn/synchronous/errors.py

class DatabaseError(Error):
    """Error related to the database."""

`DataError`

Bases: DatabaseError

Error due to problems with processed data.

Source code in picopyn/synchronous/errors.py

class DataError(DatabaseError):
    """Error due to problems with processed data."""

`OperationalError`

Bases: DatabaseError

Error related to database operation.

Source code in picopyn/synchronous/errors.py

class OperationalError(DatabaseError):
    """Error related to database operation."""

`IntegrityError`

Bases: DatabaseError

Error related to relational integrity.

Source code in picopyn/synchronous/errors.py

class IntegrityError(DatabaseError):
    """Error related to relational integrity."""

`InternalError`

Bases: DatabaseError

Internal database error.

Source code in picopyn/synchronous/errors.py

class InternalError(DatabaseError):
    """Internal database error."""

`ProgrammingError`

Bases: DatabaseError

Programming error, such as invalid SQL or invalid API usage.

Source code in picopyn/synchronous/errors.py

class ProgrammingError(DatabaseError):
    """Programming error, such as invalid SQL or invalid API usage."""

`NotSupportedError`

Bases: DatabaseError

Error raised when an unsupported DB-API feature is requested.

Source code in picopyn/synchronous/errors.py

class NotSupportedError(DatabaseError):
    """Error raised when an unsupported DB-API feature is requested."""

Sync

picopyn.synchronous

apilevel = '2.0' module-attribute

connect = Connection.connect module-attribute

paramstyle = 'format' module-attribute

threadsafety = 1 module-attribute

Connection

autocommit property writable

cancel()

close()

commit()

connect(dsn, on_query_metadata=None, **connect_kwargs) classmethod

cursor()

is_closed()

prepare(query, parameters=None)

rollback()

Cursor

arraysize property writable

description property

rowcount property

statusmessage property

__iter__()

close()

execute(operation, parameters=None)

executemany(operation, seq_of_parameters)

fetchall()

fetchmany(size=None)

fetchone()

Pool

balance_strategy property writable

instance_ejection_duration property writable

instance_ejection_enabled property

Instance Ejection and Connection Selection

instance_ejection_failure_threshold property writable

instance_ejection_failure_window property writable

max_size property writable

query_metadata_cache_size property writable

rebalance_pool_divisor property writable

routing_acquire_timeout property writable

topology property

Reconcile

Rebalance

Example cases

topology_update_interval property writable

close(timeout=None)

connection(timeout=None)

connection_by_instance_uuid(instance_uuid, timeout=None)

connection_by_sharding_key(sharding_key, timeout=None)

connection_by_tier_and_bucket_id(tier, bucket_id, timeout=None)

create_sharding_key_factory(table_name)

disable_instance_ejection()

disable_query_metadata()

disable_topology_tracking()

execute(query, parameters=None)

fetch(query, parameters=None)

fetchrow(query, parameters=None)

get_query_metadata(query)

open()

terminate()

Errors

Warning

Error

InterfaceError

DatabaseError

DataError

OperationalError

IntegrityError

InternalError

ProgrammingError

NotSupportedError

`picopyn.synchronous`

`apilevel = '2.0'` `module-attribute`

`connect = Connection.connect` `module-attribute`

`paramstyle = 'format'` `module-attribute`

`threadsafety = 1` `module-attribute`

`Connection`

`autocommit` `property` `writable`

`cancel()`

`close()`

`commit()`

`connect(dsn, on_query_metadata=None, **connect_kwargs)` `classmethod`

`cursor()`

`is_closed()`

`prepare(query, parameters=None)`

`rollback()`

`Cursor`

`arraysize` `property` `writable`

`description` `property`

`rowcount` `property`

`statusmessage` `property`

`iter()`

`close()`

`execute(operation, parameters=None)`

`executemany(operation, seq_of_parameters)`

`fetchall()`

`fetchmany(size=None)`

`fetchone()`

`Pool`

`balance_strategy` `property` `writable`

`instance_ejection_duration` `property` `writable`

`instance_ejection_enabled` `property`

`instance_ejection_failure_threshold` `property` `writable`

`instance_ejection_failure_window` `property` `writable`

`max_size` `property` `writable`

`query_metadata_cache_size` `property` `writable`

`rebalance_pool_divisor` `property` `writable`

`routing_acquire_timeout` `property` `writable`

`topology` `property`

`topology_update_interval` `property` `writable`

`close(timeout=None)`

`connection(timeout=None)`

`connection_by_instance_uuid(instance_uuid, timeout=None)`

`connection_by_sharding_key(sharding_key, timeout=None)`

`connection_by_tier_and_bucket_id(tier, bucket_id, timeout=None)`

`create_sharding_key_factory(table_name)`

`disable_instance_ejection()`

`disable_query_metadata()`

`disable_topology_tracking()`

`execute(query, parameters=None)`

`fetch(query, parameters=None)`

`fetchrow(query, parameters=None)`

`get_query_metadata(query)`

`open()`

`terminate()`

`Warning`

`Error`

`InterfaceError`

`DatabaseError`

`DataError`

`OperationalError`

`IntegrityError`

`InternalError`

`ProgrammingError`

`NotSupportedError`