Skip to content

Async

picopyn.asynchronous

Async API based on asyncpg for connection management and pools.

Connection

A representation of a database session.

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 asyncpg.connect() (e.g., ssl).

 {}
Note

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

Every connection always requests the statements invalidation option (so execute/fetch/prepare/etc. can detect and recover from stale plans), but the Notice listener is registered only when on_query_metadata is set. See picopyn.query_metadata for more on the underlying protocol.

Source code in picopyn/asynchronous/connection.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
class Connection:
    """A representation of a database session.

    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.asynchronous.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 asyncpg.connect() (e.g., ssl).

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

    Every connection always requests the
    [statements invalidation option][picopyn.query_metadata.STMT_INVALIDATION_OPTION]
    (so `execute`/`fetch`/`prepare`/etc. can detect and recover from stale plans), but
    the Notice listener is registered only when `on_query_metadata` is set. See
    [`picopyn.query_metadata`][] for more on the underlying protocol.
    """

    def __init__(
        self,
        dsn: str,
        on_query_metadata: Callable[[PreparedStatementMetadata], None] | None = None,
        **_connect_kwargs: Any,
    ) -> None:
        if dsn is None:
            raise ValueError("dsn can not be None")

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

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

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

    async def connect(self) -> None:
        """Create a new connection to Picodata.

        Register the Notice listener if needed.

        Closes the existing connection first, if already connected.

        Raises:
            RuntimeError: If the connection attempt fails.
        """
        if self.conn and not self.is_closed():
            await self.close()

        connect_kwargs = _with_picodata_startup_options(
            self._connect_kwargs,
            self._connect_dsn,
            request_query_metadata=self._on_query_metadata is not None,
        )
        try:
            self.conn = await asyncpg.connect(self._connect_dsn, **connect_kwargs)
        except Exception as e:
            raise RuntimeError(
                f"Failed to connect to picodata instance using DSN {self.dsn}: {e}"
            ) from e

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

    def _on_query_metadata_notice(
        self, _connection: asyncpg.Connection, message: PostgresLogMessage
    ) -> None:
        """Forward the [parsed][picopyn.query_metadata.parse_query_metadata_notice]
        query metadata to the `on_query_metadata` callback, if set.

        Currently, that means handing it to the pool's
        [cache service][picopyn.asynchronous.metadata_service.QueryMetadataService],
        which stores it.

        Args:
            _connection: The connection the Notice arrived on -- part of
                asyncpg's log listener signature, unused here since `self`
                already has the connection.
            message: The raw Notice, as asyncpg parsed it.
        """
        # Note: must stay a plain function, not a coroutine. asyncpg delivers
        # Notices before it resolves the `await conn.prepare(...)` that's
        # waiting on them. A plain function runs to completion immediately,
        # so the cache is guaranteed to be filled by the time `prepare()`
        # returns. A coroutine callback wouldn't give that guarantee -- it
        # could still be running (or not even started) when `prepare()` returns.
        if self._on_query_metadata is None:
            return
        metadata = parse_query_metadata_notice(message.message)
        if metadata is not None:
            self._on_query_metadata(metadata)

    async def _clear_stmt_cache_if_stale(self, exc: BaseException) -> None:
        """If the exception is Picodata's stale-plan signal, clear asyncpg's own
        statement cache so it stops reusing the now-invalid plan.

        asyncpg keeps its own cache of prepared statements and reuses them
        without talking to the server again. It doesn't know about Picodata's
        stale-plan signal, so it won't clear a cached entry on its own -- left
        alone, the next call with the same query text would just get the same
        broken statement and fail again, every time.

        asyncpg has no way to drop just one cached statement, so this clears
        its entire cache. That's fine here: the signal means the schema changed,
        so other cached queries touching it are likely stale too.

        Args:
            exc: The exception raised by the failed operation. A no-op if
                this isn't Picodata's stale-plan signal.
        """
        if not is_stmt_invalidated_error(exc):
            return
        if self.conn is not None:
            with contextlib.suppress(Exception):
                await self.conn.reload_schema_state()

    async def prepare(self, query: str, **kwargs: Any) -> asyncpg.prepared_stmt.PreparedStatement:
        """Create a prepared statement for `query`.

        Call this directly if you want the prepared statement handle itself
        (e.g. to bind and execute it multiple times), or if you specifically
        want to force a `Parse` now -- this is what
        [query metadata service][picopyn.asynchronous.metadata_service.QueryMetadataService]
        uses on its dedicated connection to trigger Picodata's metadata Notice.

        Args:
            query: The SQL query text to prepare.
            **kwargs: Additional keyword arguments forwarded to
                `conn.prepare()` (e.g. `timeout`).

        Returns:
            The prepared statement handle.

        Raises:
            OSError: If there is no active connection.
            RuntimeError: If preparing the statement fails.

        Examples:
            ```python
            stmt = await conn.prepare("SELECT * FROM warehouse WHERE id = $1")
            row = await stmt.fetchrow(1)
            rows = await stmt.fetch(2)
            ```
        """
        if not self.conn:
            raise OSError("No active connection. Try to call .connect() before.")

        try:
            return await self.conn.prepare(query, **kwargs)
        except Exception as e:
            await self._clear_stmt_cache_if_stale(e)
            raise RuntimeError(f"Failed to prepare SQL query: {e}. Query: {query}") from e

    async def execute(self, *args: Any, **kwargs: Any) -> str:
        """Execute an SQL command.

        Args:
            *args: The SQL command text, followed by any bind parameters.
            **kwargs: Additional keyword arguments forwarded to
                `conn.execute()` (e.g. `timeout`).

        Returns:
            The command's status tag, as returned by the server (e.g. "INSERT 0 1").

        Raises:
            OSError: If there is no active connection.
            RuntimeError: If executing the command fails.
        """

        if not self.conn:
            raise OSError("No active connection. Try to call .connect() before.")

        try:
            return await self.conn.execute(*args, **kwargs)
        except Exception as e:
            await self._clear_stmt_cache_if_stale(e)
            raise RuntimeError(f"Failed to execute SQL query: {e}. Query: {args}") from e

    async def fetchrow(self, *args: Any, **kwargs: Any) -> asyncpg.Record | None:
        """Run a query and return the first row.

        Args:
            *args: The SQL query text, followed by any bind parameters.
            **kwargs: Additional keyword arguments forwarded to
                `conn.fetchrow()` (e.g. `timeout`).

        Returns:
            The first row, or None if the query returned no rows.

        Raises:
            OSError: If there is no active connection.
            RuntimeError: If the query fails.
        """

        if not self.conn:
            raise OSError("No active connection. Try to call .connect() before")

        try:
            return await self.conn.fetchrow(*args, **kwargs)
        except Exception as e:
            await self._clear_stmt_cache_if_stale(e)
            raise RuntimeError(
                f"Failed to execute SQL query and fetch row: {e}. Query: {args}"
            ) from e

    async def fetch(self, *args: Any, **kwargs: Any) -> list[asyncpg.Record]:
        """Run a query and return the results as a list.

        Args:
            *args: The SQL query text, followed by any bind parameters.
            **kwargs: Additional keyword arguments forwarded to
                `conn.fetch()` (e.g. `timeout`).

        Returns:
            All matching rows.

        Raises:
            OSError: If there is no active connection.
            RuntimeError: If the query fails.
        """

        if not self.conn:
            raise OSError("No active connection. Try to call .connect() before")

        try:
            return await self.conn.fetch(*args, **kwargs)
        except Exception as e:
            await self._clear_stmt_cache_if_stale(e)
            raise RuntimeError(
                f"Failed to execute SQL query and fetch result: {e}. Query: {args}"
            ) from e

    async def explain(
        self, query: str, *args: Any, raw: bool = False
    ) -> ExplainPlan | ExplainRawPlan:
        """Run EXPLAIN for a query and return a structured plan.

        Args:
            query: The SQL query string without EXPLAIN prefix.
            *args: Optional parameters for the SQL query.
            raw: If True, uses EXPLAIN (RAW).

        Returns:
            ExplainPlan for plain EXPLAIN or ExplainRawPlan for RAW mode.

        Raises:
            RuntimeError: If executing EXPLAIN fails or its output cannot be parsed. See
                [`parse_plain_explain_lines()`][picopyn.explain.parse_plain_explain_lines]
                and [`parse_raw_explain_lines()`][picopyn.explain.parse_raw_explain_lines]
                for the underlying parsing functions and their errors.

        Examples:
            ```python
            plan = await conn.explain(
                'SELECT * FROM "warehouse" WHERE id = $1',
                1,
            )
            raw_plan = await conn.explain("SELECT * FROM warehouse", raw=True)
            ```
        """

        explain_query = build_explain_query(query=query, raw=raw)
        try:
            rows = await self.fetch(explain_query, *args)
            lines = rows_to_lines(rows)
            if raw:
                return parse_raw_explain_lines(lines)
            return parse_plain_explain_lines(lines)
        except ValueError as e:
            raise RuntimeError(
                f"Failed to parse EXPLAIN output: {e}. Query: {explain_query}"
            ) from e

    async def close(self, *args: Any, **kwargs: Any) -> None:
        """Close the connection gracefully.

        Args:
            *args: Forwarded to `conn.close()`.
            **kwargs: Forwarded to `conn.close()` (e.g. `timeout`).

        Raises:
            RuntimeError: If closing the connection fails.
        """
        if self.conn:
            try:
                return await self.conn.close(*args, **kwargs)
            except Exception as e:
                raise RuntimeError(
                    f"Failed to disconnect from picodata instance {self.dsn}: {e}"
                ) from e

    def terminate(self) -> None:
        """Terminate the connection without waiting for graceful shutdown."""
        if self.conn:
            self.conn.terminate()

close(*args, **kwargs) async

Close the connection gracefully.

Parameters:

Name Type Description Default
*args Any

Forwarded to conn.close().

 ()
**kwargs Any

Forwarded to conn.close() (e.g. timeout).

 {}

Raises:

Type Description
RuntimeError

If closing the connection fails.
Source code in picopyn/asynchronous/connection.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
async def close(self, *args: Any, **kwargs: Any) -> None:
    """Close the connection gracefully.

    Args:
        *args: Forwarded to `conn.close()`.
        **kwargs: Forwarded to `conn.close()` (e.g. `timeout`).

    Raises:
        RuntimeError: If closing the connection fails.
    """
    if self.conn:
        try:
            return await self.conn.close(*args, **kwargs)
        except Exception as e:
            raise RuntimeError(
                f"Failed to disconnect from picodata instance {self.dsn}: {e}"
            ) from e

connect() async

Create a new connection to Picodata.

Register the Notice listener if needed.

Closes the existing connection first, if already connected.

Raises:

Type Description
RuntimeError

If the connection attempt fails.
Source code in picopyn/asynchronous/connection.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def connect(self) -> None:
    """Create a new connection to Picodata.

    Register the Notice listener if needed.

    Closes the existing connection first, if already connected.

    Raises:
        RuntimeError: If the connection attempt fails.
    """
    if self.conn and not self.is_closed():
        await self.close()

    connect_kwargs = _with_picodata_startup_options(
        self._connect_kwargs,
        self._connect_dsn,
        request_query_metadata=self._on_query_metadata is not None,
    )
    try:
        self.conn = await asyncpg.connect(self._connect_dsn, **connect_kwargs)
    except Exception as e:
        raise RuntimeError(
            f"Failed to connect to picodata instance using DSN {self.dsn}: {e}"
        ) from e

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

execute(*args, **kwargs) async

Execute an SQL command.

Parameters:

Name Type Description Default
*args Any

The SQL command text, followed by any bind parameters.

 ()
**kwargs Any

Additional keyword arguments forwarded to conn.execute() (e.g. timeout).

 {}

Returns:

Type Description
str

The command's status tag, as returned by the server (e.g. "INSERT 0 1").

Raises:

Type Description
OSError

If there is no active connection.
RuntimeError

If executing the command fails.
Source code in picopyn/asynchronous/connection.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def execute(self, *args: Any, **kwargs: Any) -> str:
    """Execute an SQL command.

    Args:
        *args: The SQL command text, followed by any bind parameters.
        **kwargs: Additional keyword arguments forwarded to
            `conn.execute()` (e.g. `timeout`).

    Returns:
        The command's status tag, as returned by the server (e.g. "INSERT 0 1").

    Raises:
        OSError: If there is no active connection.
        RuntimeError: If executing the command fails.
    """

    if not self.conn:
        raise OSError("No active connection. Try to call .connect() before.")

    try:
        return await self.conn.execute(*args, **kwargs)
    except Exception as e:
        await self._clear_stmt_cache_if_stale(e)
        raise RuntimeError(f"Failed to execute SQL query: {e}. Query: {args}") from e

explain(query, *args, raw=False) async

Run EXPLAIN for a query and return a structured plan.

Parameters:

Name Type Description Default
query str

The SQL query string without EXPLAIN prefix.

 required
*args Any

Optional parameters for the SQL query.

 ()
raw bool

If True, uses EXPLAIN (RAW).

 False

Returns:

Type Description
ExplainPlan | ExplainRawPlan

ExplainPlan for plain EXPLAIN or ExplainRawPlan for RAW mode.

Raises:

Type Description
RuntimeError

If executing EXPLAIN fails or its output cannot be parsed. See parse_plain_explain_lines() and parse_raw_explain_lines() for the underlying parsing functions and their errors.

Examples:

plan = await conn.explain(
    'SELECT * FROM "warehouse" WHERE id = $1',
    1,
)
raw_plan = await conn.explain("SELECT * FROM warehouse", raw=True)
Source code in picopyn/asynchronous/connection.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
async def explain(
    self, query: str, *args: Any, raw: bool = False
) -> ExplainPlan | ExplainRawPlan:
    """Run EXPLAIN for a query and return a structured plan.

    Args:
        query: The SQL query string without EXPLAIN prefix.
        *args: Optional parameters for the SQL query.
        raw: If True, uses EXPLAIN (RAW).

    Returns:
        ExplainPlan for plain EXPLAIN or ExplainRawPlan for RAW mode.

    Raises:
        RuntimeError: If executing EXPLAIN fails or its output cannot be parsed. See
            [`parse_plain_explain_lines()`][picopyn.explain.parse_plain_explain_lines]
            and [`parse_raw_explain_lines()`][picopyn.explain.parse_raw_explain_lines]
            for the underlying parsing functions and their errors.

    Examples:
        ```python
        plan = await conn.explain(
            'SELECT * FROM "warehouse" WHERE id = $1',
            1,
        )
        raw_plan = await conn.explain("SELECT * FROM warehouse", raw=True)
        ```
    """

    explain_query = build_explain_query(query=query, raw=raw)
    try:
        rows = await self.fetch(explain_query, *args)
        lines = rows_to_lines(rows)
        if raw:
            return parse_raw_explain_lines(lines)
        return parse_plain_explain_lines(lines)
    except ValueError as e:
        raise RuntimeError(
            f"Failed to parse EXPLAIN output: {e}. Query: {explain_query}"
        ) from e

fetch(*args, **kwargs) async

Run a query and return the results as a list.

Parameters:

Name Type Description Default
*args Any

The SQL query text, followed by any bind parameters.

 ()
**kwargs Any

Additional keyword arguments forwarded to conn.fetch() (e.g. timeout).

 {}

Returns:

Type Description
list[Record]

All matching rows.

Raises:

Type Description
OSError

If there is no active connection.
RuntimeError

If the query fails.
Source code in picopyn/asynchronous/connection.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
async def fetch(self, *args: Any, **kwargs: Any) -> list[asyncpg.Record]:
    """Run a query and return the results as a list.

    Args:
        *args: The SQL query text, followed by any bind parameters.
        **kwargs: Additional keyword arguments forwarded to
            `conn.fetch()` (e.g. `timeout`).

    Returns:
        All matching rows.

    Raises:
        OSError: If there is no active connection.
        RuntimeError: If the query fails.
    """

    if not self.conn:
        raise OSError("No active connection. Try to call .connect() before")

    try:
        return await self.conn.fetch(*args, **kwargs)
    except Exception as e:
        await self._clear_stmt_cache_if_stale(e)
        raise RuntimeError(
            f"Failed to execute SQL query and fetch result: {e}. Query: {args}"
        ) from e

fetchrow(*args, **kwargs) async

Run a query and return the first row.

Parameters:

Name Type Description Default
*args Any

The SQL query text, followed by any bind parameters.

 ()
**kwargs Any

Additional keyword arguments forwarded to conn.fetchrow() (e.g. timeout).

 {}

Returns:

Type Description
Record | None

The first row, or None if the query returned no rows.

Raises:

Type Description
OSError

If there is no active connection.
RuntimeError

If the query fails.
Source code in picopyn/asynchronous/connection.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
async def fetchrow(self, *args: Any, **kwargs: Any) -> asyncpg.Record | None:
    """Run a query and return the first row.

    Args:
        *args: The SQL query text, followed by any bind parameters.
        **kwargs: Additional keyword arguments forwarded to
            `conn.fetchrow()` (e.g. `timeout`).

    Returns:
        The first row, or None if the query returned no rows.

    Raises:
        OSError: If there is no active connection.
        RuntimeError: If the query fails.
    """

    if not self.conn:
        raise OSError("No active connection. Try to call .connect() before")

    try:
        return await self.conn.fetchrow(*args, **kwargs)
    except Exception as e:
        await self._clear_stmt_cache_if_stale(e)
        raise RuntimeError(
            f"Failed to execute SQL query and fetch row: {e}. Query: {args}"
        ) from e

is_closed()

Returns:

Type Description
bool

True if there is no active underlying connection.
Source code in picopyn/asynchronous/connection.py 
122
123
124
125
126
127
def is_closed(self) -> bool:
    """
    Returns:
        True if there is no active underlying connection.
    """
    return self.conn is None or self.conn.is_closed()

prepare(query, **kwargs) async

Create a prepared statement for query.

Call this directly if you want the prepared statement handle itself (e.g. to bind and execute it multiple times), or 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 str

The SQL query text to prepare.

 required
**kwargs Any

Additional keyword arguments forwarded to conn.prepare() (e.g. timeout).

 {}

Returns:

Type Description
PreparedStatement

The prepared statement handle.

Raises:

Type Description
OSError

If there is no active connection.
RuntimeError

If preparing the statement fails.

Examples:

stmt = await conn.prepare("SELECT * FROM warehouse WHERE id = $1")
row = await stmt.fetchrow(1)
rows = await stmt.fetch(2)
Source code in picopyn/asynchronous/connection.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
async def prepare(self, query: str, **kwargs: Any) -> asyncpg.prepared_stmt.PreparedStatement:
    """Create a prepared statement for `query`.

    Call this directly if you want the prepared statement handle itself
    (e.g. to bind and execute it multiple times), or if you specifically
    want to force a `Parse` now -- this is what
    [query metadata service][picopyn.asynchronous.metadata_service.QueryMetadataService]
    uses on its dedicated connection to trigger Picodata's metadata Notice.

    Args:
        query: The SQL query text to prepare.
        **kwargs: Additional keyword arguments forwarded to
            `conn.prepare()` (e.g. `timeout`).

    Returns:
        The prepared statement handle.

    Raises:
        OSError: If there is no active connection.
        RuntimeError: If preparing the statement fails.

    Examples:
        ```python
        stmt = await conn.prepare("SELECT * FROM warehouse WHERE id = $1")
        row = await stmt.fetchrow(1)
        rows = await stmt.fetch(2)
        ```
    """
    if not self.conn:
        raise OSError("No active connection. Try to call .connect() before.")

    try:
        return await self.conn.prepare(query, **kwargs)
    except Exception as e:
        await self._clear_stmt_cache_if_stale(e)
        raise RuntimeError(f"Failed to prepare SQL query: {e}. Query: {query}") from e

terminate()

Terminate the connection without waiting for graceful shutdown.

Source code in picopyn/asynchronous/connection.py 
391
392
393
394
def terminate(self) -> None:
    """Terminate the connection without waiting for graceful shutdown."""
    if self.conn:
        self.conn.terminate()

Pool

A connection pool.

Connection pool can be used to manage a set of connections to the database. Connections are first acquired from the pool, then used, and then released back to the pool.

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.

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.

---
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.asynchronous.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: any cluster node,<br>bootstrap mode: DSN nodes only)"]:::note
click RECONCILE_NOTE "#picopyn.asynchronous.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["pool.acquire() -> conn"] --> USE["Execute queries<br>· conn.fetchrow()<br>· conn.fetch()<br>· conn.execute()"] --> RELEASE["pool.release(conn)"]
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. Must be at least 1.

 10
enable_discovery bool

If True, the pool will automatically discover available picodata instances. If False, only the given dsn will be used for user-usage connections.

 False
balance_strategy Callable[[list[Connection]], Connection] | None

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

 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. The filter applies to the pool's user-facing connections only. Technical connections can ignore it.

 None
topology_update_interval float | None

Interval in seconds between topology refresh cycles for Topology Tracker. Set to None to disable periodic updates. With updates disabled topology is None, reconcile never runs, and shard-aware routing does not work because it needs a topology snapshot.

 60.0
query_metadata_cache_size 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). Oldest entries are evicted once exceeded. Must be at least 1 if set. Set to None to disable the metadata service: the service's technical connection is never opened, get_query_metadata raises RuntimeError, and shard-aware routing in execute always falls back to an ordinary pool connection.

 100
rebalance_pool_divisor int

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.

 10
routing_acquire_timeout float

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.

 0.0
**connect_kwargs Any

Additional keyword arguments to pass to asyncpg.connect() (e.g., ssl).

 {}

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. Nodes are iterated in round-robin until a connection succeeds, enabling cluster discovery:

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

SSL (kwargs-based):

import ssl

ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
ctx.load_cert_chain(certfile="/path/to/client.crt", keyfile="/path/to/client.key")
pool = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    ssl=ctx,
)

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/asynchronous/pool.py 
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
class Pool:
    """A connection pool.

    Connection pool can be used to manage a set of connections to the database.
    Connections are first acquired from the pool, then used, and then released back to the pool.

    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.

    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.asynchronous.topology_tracker.TopologyTracker] and
        [metadata cache service][picopyn.asynchronous.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.

    --8<-- "_pool_lifecycle_async.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. Must be at least 1.
        enable_discovery: If True, the pool will automatically discover available picodata
            instances. If False, only the given `dsn` will be used for user-usage connections.
        balance_strategy: A custom strategy function to select a connection from the pool.
            If None, round-robin strategy is used.
        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. The filter applies to the pool's user-facing connections only. Technical connections
            can ignore it.
        topology_update_interval: Interval in seconds between topology refresh cycles for
            [Topology Tracker][picopyn.asynchronous.topology_tracker.TopologyTracker].
            Set to None to disable periodic updates. With updates disabled
            [topology][picopyn.asynchronous.pool.Pool.topology] is None, reconcile
            never runs, and [shard-aware routing][picopyn.asynchronous.pool.Pool.acquire_by_tier_and_bucket_id]
            does not work because it needs a topology snapshot.
        query_metadata_cache_size: 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.asynchronous.pool.Pool.get_query_metadata]).
            Oldest entries are evicted once exceeded. Must be at least 1 if set.
            Set to None to disable the metadata service: the service's technical connection is never
            opened, [get_query_metadata][picopyn.asynchronous.pool.Pool.get_query_metadata] raises
            RuntimeError, and shard-aware routing in [execute][picopyn.asynchronous.pool.Pool.execute]
            always falls back to an ordinary pool connection.
        rebalance_pool_divisor: How much of the pool a single
            [rebalance][picopyn.asynchronous.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.
        routing_acquire_timeout: Time in seconds that
            [shard-aware routing][picopyn.asynchronous.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.
        **connect_kwargs: Additional keyword arguments to pass to asyncpg.connect() (e.g., ssl).

    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. Nodes are iterated in round-robin until a connection
        succeeds, enabling cluster discovery:

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

        SSL (kwargs-based):

        ```python
        import ssl

        ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
        ctx.load_cert_chain(certfile="/path/to/client.crt", keyfile="/path/to/client.key")
        pool = Pool(
            dsn="postgresql://admin:pass@host1:5432,host2:5432",
            ssl=ctx,
        )
        ```

        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,
        **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")

        # 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._max_size = max_size
        self._rebalance_pool_divisor = rebalance_pool_divisor
        self._routing_acquire_timeout = routing_acquire_timeout
        # 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
        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()
        )
        # guards all pool state
        self._lock: asyncio.Lock = asyncio.Lock()
        # wake waiters that can acquire any available connection
        # only one waiter can acquire a connection after it is returned to the pool,
        # so notifying all waiters at once is unnecessary and may be expensive
        self._any_free: asyncio.Condition = asyncio.Condition(self._lock)
        # wake waiters that are waiting for a connection from a specific node, one
        # condition per instance uuid so that freeing a connection to one node does not
        # wake the waiters of every other node
        self._uuid_free: dict[str, asyncio.Condition] = {}
        self._default_acquire_timeout_sec = 5
        # Pool shutdown state: None -> open; pending -> closing; done -> closed.
        # close() stores a Task, terminate() may store a done Future.
        self._close_task: asyncio.Future[None] | None = None
        # completed only after release cleanup finishes
        self._in_use: dict[Connection, asyncio.Future[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
        # 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

    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 _check_open(self) -> None:
        if self._close_task is None:
            return
        if self._close_task.done():
            raise RuntimeError("pool is closed")
        raise RuntimeError("pool is closing")

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

    def _mark_as_used(self, conn: Connection) -> None:
        self._used.add(conn)
        self._in_use[conn] = asyncio.get_running_loop().create_future()

    def _finish_release(self, conn: Connection) -> None:
        self._used.discard(conn)
        waiter = self._in_use.pop(conn, None)
        if waiter is not None and not waiter.done():
            waiter.set_result(None)

    async 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>
        """
        async with self._lock:
            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)
            # TODO: maybe we should use `_reconcile_pool_with_topology` instead
            if self._enable_discovery:
                await self._fill_pool_from_discovery()
            else:
                await 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:
                        await conn.close()
                    except Exception as e:
                        logger.warning("Pool open 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", len(self._pool))

            if self._topology_update_interval is not None and self._topology_tracker is None:
                # TODO: the very first technical connection always goes to the first
                # reachable DSN node: picking a node by load needs the topology, and
                # reading the topology needs a connection, so at startup there is no
                # choice but the DSN. maybe we should reopen it right after the first
                # topology refresh, so the least loaded node is picked from the start
                self._topology_tracker = TopologyTracker(
                    PollingTopologySource(self._get_technical_connection),
                    self._topology_update_interval,
                )
                # register the pool's reconcile callback so connections are updated
                # to match the topology after each refresh
                await 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:
                # 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,
                )

                await self._query_metadata_service.start()
                logger.debug(
                    "Pool query-metadata service started with cache size %s",
                    self._query_metadata_cache_size,
                )
            return

    async def _fill_pool_from_discovery(self) -> None:
        """
        Fill the pool with online nodes from Picodata cluster if they pass filters.
        Current filter is `forbidden_tiers`.
        """
        try:
            instance_addrs = await 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)
            try:
                conn = Connection(dsn, **self._connect_kwargs)
                await conn.connect()
                conn.instance_uuid = self._instance_uuid_by_address.get(address)
                self._pool.append(conn)
                logger.debug("Discovery: connected to node %s", address)
            except Exception as e:
                logger.warning("Discovery: could not connect to node %s for pool: %s", address, e)
                addrs.popleft()
                continue

            addrs.rotate(-1)

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

        Current filter is `forbidden_tiers`.
        """
        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
            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(candidate, **self._connect_kwargs)
                await conn.connect()

                if instance_info_by_address is None:
                    instance_info_by_address = await self._fetch_bootstrap_instance_info(conn)
                    instance_info = instance_info_by_address.get(address)
                    if not self._is_accepted_bootstrap_candidate(instance_info, address):
                        await 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:
                if conn is not None:
                    try:
                        await 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]]:
        return is_accepted_bootstrap_candidate(instance_info, address, self._forbidden_tiers)

    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].
        """
        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].
        """
        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)

    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].
        """
        return technical_candidates(topology, self._forbidden_tiers, (*self._pool, *self._used))

    async def _get_technical_connection(
        self, on_query_metadata: Callable[[PreparedStatementMetadata], None] | 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 topology snapshot:
        - 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 = self.topology
        candidates = self._technical_candidates(topology) if topology 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(dsn, on_query_metadata=on_query_metadata, **self._connect_kwargs)
                await conn.connect()
            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 await self._get_dsn_connection(on_query_metadata)

    async def _restore_rebalance_old_connection(self, old_conn: Connection) -> bool:
        if self._close_task is not None:
            return False

        async with self._lock:
            if self._close_task is not None or self._current_size() >= self._max_size:
                return False
            self._insert_randomly(old_conn)
            self._wake_waiters(old_conn)
            return True

    def _terminate_rebalance_connection(self, conn: Connection, error_message: str) -> None:
        try:
            conn.terminate()
        except Exception as e:
            logger.warning(error_message, e)

    async 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 terminated if the pool
            is already closing or the slot has already been filled.
        """
        dsn = _replace_dsn_host(self._connect_dsns[0], new_node_address)
        new_conn: Connection | None = None
        migrated = False
        try:
            new_conn = Connection(dsn, **self._connect_kwargs)
            await new_conn.connect()
            new_conn.instance_uuid = new_node_uuid
            async with self._lock:
                self._check_open()
                if self._current_size() < self._max_size:
                    self._insert_randomly(new_conn)
                    self._wake_waiters(new_conn)
                    migrated = True
        except Exception as e:
            logger.warning("Rebalance: could not migrate connection to %s: %s", new_node_address, e)
        finally:
            # both connections belong to nobody while the replacement is being opened,
            # and opening it can be interrupted by cancellation as well as by an error
            # (pool shutdown cancels the reconcile task)
            if not migrated:
                # the old connection goes back first: it is the one the pool is missing
                # until it returns, while the replacement is only a left to drop
                if not await self._restore_rebalance_old_connection(old_conn):
                    # the pool is closing or the free slot was filled already, so there is
                    # nothing to await on here: terminate instead of handing back
                    self._terminate_rebalance_connection(
                        old_conn,
                        "Rebalance: could not drop the old connection: %s",
                    )
                # the replacement may be established already (cancelled while waiting for
                # the lock, pool closing): nothing owns it, and a cancelled coroutine
                # cannot await a close, so drop it right away
                if new_conn is not None:
                    # this runs in a `finally`, so a raise here would replace the
                    # cancellation that brought us in with an unrelated error
                    self._terminate_rebalance_connection(
                        new_conn,
                        "Rebalance: could not drop the replacement: %s",
                    )

        if migrated:
            try:
                await old_conn.close()
            except Exception as e:
                logger.warning(
                    "Rebalance: could not close migrated connection to %s: %s",
                    old_conn.instance_uuid,
                    e,
                )

        return migrated

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

        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 closing or 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.
        """
        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

            async with self._lock:
                try:
                    self._check_open()
                except RuntimeError:
                    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
                async with self._lock:
                    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 await 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,
            )

    async 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.
        """
        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
        async with self._lock:
            # remove connections to dead nodes
            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)

        # close stale connections outside the lock --
        # this must not block other coroutines waiting to acquire/release
        for conn in stale_conns:
            try:
                await 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:
            async with self._lock:
                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(dsn, **self._connect_kwargs)
                await conn.connect()
            except Exception as e:
                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
            async with self._lock:
                try:
                    self._check_open()
                except RuntimeError:
                    conn_to_close = conn
                else:
                    if self._current_size() >= self._max_size:
                        conn_to_close = conn
                    else:
                        self._insert_randomly(conn)
                        self._wake_waiters(conn)

            if conn_to_close is not None:
                try:
                    await 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
        await 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)

    async def _fetch_bootstrap_instance_info(
        self,
        conn: Connection,
    ) -> dict[str, InstanceInfo]:
        try:
            rows = await conn.fetch(BOOTSTRAP_INSTANCE_INFO_QUERY)
            instance_info_by_address, malformed_rows = _parse_instance_info_by_address(
                cast(Iterable[InstanceInfoRow], 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

    async 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(dsn, **self._connect_kwargs)
                await candidate.connect()
                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 = await self._fetch_discovery_rows(temp_conn)
            online_addresses, self._instance_uuid_by_address = _collect_online_nodes(
                cast(Iterable[DiscoveryRow], rows)
            )
            if not online_addresses:
                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:
            await temp_conn.close()

    async def _fetch_discovery_rows(self, conn: Connection) -> list[asyncpg.Record]:
        # all instance addresses excluding forbidden tiers
        if self._forbidden_tiers:
            placeholders = ", ".join(f"${i + 1}" for i in range(len(self._forbidden_tiers)))
            query = DISCOVERY_INSTANCE_ADDRESSES_EXCLUDING_TIERS_QUERY_TEMPLATE.format(
                placeholders=placeholders
            )
            return await conn.fetch(query, *list(self._forbidden_tiers))

        # all instance addresses
        return await conn.fetch(DISCOVERY_INSTANCE_ADDRESSES_QUERY)

    async 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:
                candidate = Connection(
                    dsn, on_query_metadata=on_query_metadata, **self._connect_kwargs
                )
                await candidate.connect()
                return candidate
            except Exception as e:
                last_error = e

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

    def _wake_waiters(self, conn: Connection) -> None:
        """Wake waiters that may be able to acquire the free connection

        Must be called with `self._lock` held.
        """

        # only one waiter can acquire the newly available connection, so waking
        # one waiter from each relevant group is enough. waking all waiters would
        # make them unnecessarily compete for the lock.

        # wake a waiter waiting specifically for this node
        node_free = self._uuid_free.get(conn.instance_uuid) if conn.instance_uuid else None
        if node_free is not None:
            node_free.notify(1)

        # also wake a waiter waiting for any available connection, regardless of
        # the node UUID
        self._any_free.notify(1)

    async def _wait_for_free_connection(self, condition: asyncio.Condition, timeout: float) -> None:
        """Wait until a connection becomes available or `timeout` expires.

        Must be called with `self._lock` held; the lock is still held on return.
        """
        with contextlib.suppress(asyncio.TimeoutError):
            # `terminate()` is not a coroutine, so it cannot notify the condition
            # wake up periodically to detect a terminated pool instead of waiting
            # until the full acquire timeout expires
            await asyncio.wait_for(condition.wait(), timeout=min(0.1, timeout))

    async def _acquire_raw(self, timeout: float | None = None) -> Connection:
        """
        Acquire a connection from the pool.

        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:
            async with self._lock:
                self._check_open()
                # check if there are any available connections in the pool
                if self._pool:
                    # round-robin strategy
                    if self._balance_strategy is None:
                        conn = self._pool.popleft()
                    # custom strategy
                    else:
                        try:
                            conn = self._balance_strategy(list(self._pool))
                        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")
                        self._pool.remove(conn)

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

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

                # if no connections are available, wait until something (release/reconcile/..)
                # puts one back and notifies us
                await self._wait_for_free_connection(self._any_free, effective_timeout - elapsed)

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

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

    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

    async 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.
        """
        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:
            async with self._lock:
                self._check_open()
                conn = self._pop_connection_by_instance_uuid(instance_uuid)
                if conn is not None:
                    self._mark_as_used(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 fall back immediately instead of waiting for the acquire timeout
                if not any(
                    getattr(used_conn, "instance_uuid", None) == instance_uuid
                    for used_conn in self._used
                ):
                    return None

                node_free = self._uuid_free.get(instance_uuid)
                if node_free is None:
                    node_free = asyncio.Condition(self._lock)
                    self._uuid_free[instance_uuid] = node_free
                await self._wait_for_free_connection(node_free, effective_timeout - elapsed)

    async 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 no matching free connection appears before the
        timeout, `TimeoutError` is raised.
        """
        conn = await 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 acquire(self, timeout: float | None = None) -> _PoolAcquireContext:
        """Acquire a connection from the pool.

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

        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 can be used with `async with` or `await` to acquire a connection.

        Note:
            If you use acquire/release manually, prepared statements may be dropped when
            the connection is returned to the pool. To preserve prepared statements, use
            the connection within a single context block. This is also why Pool has no
            `prepare()` API: `Pool.prepare()` would have to release the connection before
            returning the statement to user, making it unsafe to use. So, use
            [conn.prepare()][picopyn.asynchronous.connection.Connection.prepare] and call it
            inside an `acquire()` block instead (see example below).

        Examples:
            Context manager (recommended -- safely returns the connection):

            ```python
            async with pool.acquire() as conn:
                await conn.execute("UPDATE ...")
            ```

            Explicit acquisition (e.g., for compatibility with prepared statements):

            ```python
            ctx = pool.acquire()
            conn = await ctx
            try:
                await conn.execute(...)
            finally:
                await pool.release(conn)
            ```

            Prepared statement usage with connection from pool:

            ```python
            async with pool.acquire() as conn:
                stmt = await conn.prepare('SELECT * FROM "warehouse" WHERE id = $1')
                row = await stmt.fetchrow(1)
                rows = await stmt.fetch(2)
            ```
        """
        return _PoolAcquireContext(self, timeout)

    def acquire_by_instance_uuid(
        self,
        instance_uuid: str,
        timeout: float | None = None,
    ) -> _PoolAcquireContext:
        """Acquire a pooled connection with `instance_uuid`.

        This bypasses `balance_strategy`. If no free connection with the given
        UUID becomes available before the timeout (see
        [Pool.routing_acquire_timeout][picopyn.asynchronous.pool.Pool]), `TimeoutError` is raised.

        Args:
            instance_uuid: UUID of the picodata instance to acquire a connection to.
            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 `async with` or `await` to acquire a connection.

        Examples:
            ```python
            async with pool.acquire_by_instance_uuid("f0b83347-6409-44dd-89e1-a9d967d0f4e6") as conn:
                await conn.execute("UPDATE ...")
            ```
        """
        return _PoolAcquireContext(self, timeout, instance_uuid)

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

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

        If topology tracking is disabled, or the current topology snapshot
        cannot map the bucket to a master instance, `RuntimeError` is raised.
        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 `async with` or `await` to acquire a connection.

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

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

        master = topology.find_bucket_master(tier, bucket_id)
        if master 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.acquire_by_instance_uuid(master.uuid, timeout)

    async 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 tracker is None:
            raise RuntimeError(
                "Cannot create sharding key factory: topology tracking is disabled "
                "or the pool is not open"
            )

        row = await self.fetchrow(ASYNC_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["distribution"],
            row["format"],
        )
        bucket_count = tracker.topology.tiers.get(table_info.tier)
        if bucket_count is None:
            await tracker.refresh_now()
            bucket_count = tracker.topology.tiers.get(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.BINARY)

    def acquire_by_sharding_key(
        self,
        sharding_key: ShardingKey,
        timeout: float | None = None,
    ) -> _PoolAcquireContext:
        """Acquire 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 `async with` or `await`.

        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 no matching connection becomes available before
                the timeout.
        """
        if not isinstance(sharding_key, ShardingKey):
            raise TypeError("sharding_key must be a ShardingKey")
        return self.acquire_by_tier_and_bucket_id(
            sharding_key.tier,
            sharding_key.bucket_id,
            timeout,
        )

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

        Args:
            conn: The connection to release.
        """
        await asyncio.shield(self._release(conn))

    async def _release(self, conn: Connection) -> None:
        conn_to_close: Connection | None = None
        async with self._lock:
            if conn not in self._used:
                return
            if self._close_task is not None or conn.to_evict or conn.is_closed():
                conn_to_close = conn
            else:
                self._pool.append(conn)
                self._finish_release(conn)
                self._wake_waiters(conn)

        if conn_to_close is not None:
            try:
                await conn_to_close.close()
            except Exception as e:
                logger.warning("Pool release: could not close evicted connection: %s", e)
            finally:
                async with self._lock:
                    self._finish_release(conn)

    async def close(self, timeout: float | None = None) -> None:
        """
        Closes all connections in the pool.

        Args:
            timeout: Maximum graceful shutdown time for the call that starts
                pool shutdown. If None, a default shutdown timeout 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 first timeout.
        If that timeout expires, any remaining connections are terminated.

        Note:
            This should be called during application shutdown to clean up resources.
        """
        close_task = self._close_task
        if close_task is None:
            close_task = asyncio.create_task(self._close_with_timeout(timeout))
            self._close_task = close_task

        if close_task.done():
            return

        await asyncio.shield(close_task)

    async def _close_with_timeout(self, timeout: float | None) -> None:
        effective_timeout = _DEFAULT_CLOSE_TIMEOUT_SEC if timeout is None else timeout
        try:
            await asyncio.wait_for(self._close_gracefully(), timeout=effective_timeout)
        except asyncio.TimeoutError:
            logger.warning(
                "Pool close timed out after %.1f second(s); terminating remaining connections",
                effective_timeout,
            )
            self.terminate()
        except asyncio.CancelledError:
            # terminate() cancels the shared close task to finish shutdown,
            # but concurrent close() callers should still see a closed pool.
            self.terminate()
        except Exception:
            self.terminate()
            raise

    async def _close_gracefully(self) -> None:
        # stop the tracker (and cancel any in-flight reconcile) before closing
        # pool connections, so reconcile can't refill the pool mid-shutdown
        if self._topology_tracker is not None:
            await self._topology_tracker.stop()

        shutdown = await self._start_user_connections_shutdown()
        if shutdown is None:
            return
        total, conns_to_close, wait_until_released = shutdown

        await self._close_connections(conns_to_close)

        if wait_until_released:
            await asyncio.gather(*wait_until_released)

        if self._query_metadata_service is not None:
            await self._query_metadata_service.close()

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

    async def _start_user_connections_shutdown(
        self,
    ) -> tuple[int, list[Connection], list[asyncio.Future[None]]] | None:
        conns_to_close: list[Connection] = []
        wait_until_released: list[asyncio.Future[None]] = []
        async with self._lock:
            if self._close_task is not None and self._close_task.done():
                return None

            total = self._current_size()
            conns_to_close.extend(self._pool)
            for conn in self._used:
                conn.to_evict = True
                waiter = self._in_use.get(conn)
                if waiter is None:
                    conns_to_close.append(conn)
                else:
                    wait_until_released.append(waiter)

            # the pool is closing, every waiter must stop waiting -- not just one
            self._any_free.notify_all()
            for node_free in self._uuid_free.values():
                node_free.notify_all()
        return total, conns_to_close, wait_until_released

    async def _close_connections(self, conns: Iterable[Connection]) -> None:
        for conn in conns:
            try:
                await conn.close()
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.warning("Pool close: could not close connection: %s", e)
            async with self._lock:
                if conn in self._pool:
                    self._pool.remove(conn)
                self._finish_release(conn)

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

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

    def terminate(self) -> None:
        """Terminate all connections owned by the pool without graceful close."""
        close_task = self._close_task
        if close_task is not None and close_task.done():
            return

        done = asyncio.get_running_loop().create_future()
        done.set_result(None)
        self._close_task = done

        try:
            current_task = asyncio.current_task()
        except RuntimeError:
            current_task = None
        if close_task is not None and close_task is not current_task:
            close_task.cancel()

        self._terminate_services()

        conns = [*self._pool, *self._used]
        self._pool.clear()

        for conn in conns:
            try:
                conn.terminate()
            except Exception as e:
                logger.warning("Pool terminate: could not terminate connection: %s", e)

        for conn in list(self._used):
            self._finish_release(conn)

    @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.

        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.asynchronous.pool.Pool]):
        it runs after the tracker's refresh, so disabling tracking disables it.

        Reconcile works in both modes (see [Pool.enable_discovery][picopyn.asynchronous.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`

        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 `acquire()` 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_node_outside_pool_died.md"

        --8<-- "_reconcile_pooled_node_died_size_eq.md"

        --8<-- "_reconcile_pooled_node_died_size_gt.md"

        --8<-- "_reconcile_pooled_node_died_size_lt.md"

        --8<-- "_reconcile_rebalance.md"
        </details>
        """
        return self._topology_tracker.topology if self._topology_tracker is not None else None

    def _on_query_error(self, exc: BaseException) -> None:
        """If exception looks like a connection error, ask the topology tracker to
        refresh right now, on top of its usual timer.

        Sync and non-blocking: `trigger()` just wakes up the tracker's own
        loop, so we don't spawn or track a task here.
        """

        def _is_connection_refused(exc: BaseException) -> bool:
            return "connection is closed" in str(exc).lower()

        if self._topology_tracker is not None and _is_connection_refused(exc):
            self._topology_tracker.trigger()

    async def _acquire_routed_connection(
        self,
        query: str,
        args: tuple[Any, ...],
    ) -> Connection | None:
        if self._query_metadata_service is None:
            return None

        metadata = self._query_metadata_service.request_if_missing(query)
        if metadata is None or not metadata.tier or not metadata.dk_meta:
            return None

        topology = self.topology
        if topology is None or not topology.tiers:
            return None

        bucket_count = topology.tiers.get(metadata.tier)
        if bucket_count is None:
            return None

        bucket_id = calculate_bucket_id(metadata.dk_meta, args, bucket_count)
        if bucket_id is None:
            return None

        master = topology.find_bucket_master(metadata.tier, bucket_id)
        if master is None:
            return None

        return await self._try_acquire_by_instance_uuid(
            master.uuid, timeout=self._routing_acquire_timeout
        )

    async def execute(self, query: str, *args: Any) -> 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.asynchronous.metadata_service.QueryMetadataService] for query
        meta and [topology tracking][picopyn.asynchronous.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.asynchronous.pool.Pool])
        - topology tracking enabled (see [Pool.topology_update_interval][picopyn.asynchronous.pool.Pool])
        - Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
        - execute query more than once

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

        Args:
            query: The SQL query string.
            *args: Optional parameters for the SQL query.

        Returns:
            The result of the query execution.
        """
        conn: Connection | None = None
        # prepare and cache query meta only for parameterized queries
        if args:
            conn = await self._acquire_routed_connection(query, args)

        if conn is None:
            conn = await self.acquire()

        try:
            return await conn.execute(query, *args)
        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)
            if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
                self._query_metadata_service.evict(query)
            self._on_query_error(e)
            raise
        finally:
            await self.release(conn)

    async def fetch(self, query: str, *args: Any) -> list[asyncpg.Record]:
        """Executes a query and fetches all resulting rows.

        Args:
            query: The SQL query string.
            *args: Optional parameters for the SQL query.

        Returns:
            A list of rows returned by the query.
        """
        # TODO: add cache usage when Picodata will support it for DQL
        # https://git.picodata.io/core/picodata/-/issues/2226
        async with self.acquire() as conn:
            try:
                return await conn.fetch(query, *args)
            except Exception as e:
                # TODO retry execution to hide from user the invalidation error
                if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
                    self._query_metadata_service.evict(query)
                self._on_query_error(e)
                raise

    async def fetchrow(self, query: str, *args: Any) -> asyncpg.Record | None:
        """Executes a query and fetches a single row (first row).

        Args:
            query: The SQL query string.
            *args: Optional parameters for the SQL query.

        Returns:
            A single row returned by the query.
        """
        # TODO: add cache usage when Picodata will support it for DQL
        # https://git.picodata.io/core/picodata/-/issues/2226
        async with self.acquire() as conn:
            try:
                return await conn.fetchrow(query, *args)
            except Exception as e:
                # TODO retry execution to hide from user the invalidation error
                if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
                    self._query_metadata_service.evict(query)
                self._on_query_error(e)
                raise

    async def explain(
        self, query: str, *args: Any, raw: bool = False
    ) -> ExplainPlan | ExplainRawPlan:
        """Executes EXPLAIN for a query and returns a structured plan.

        See [`Connection.explain`][picopyn.asynchronous.connection.Connection.explain]
        for full documentation.
        """
        async with self.acquire() as conn:
            return await conn.explain(query, *args, raw=raw)

    async 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.asynchronous.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"
        """
        self._check_open()

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

        return await self._query_metadata_service.get_query_metadata(query)

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.

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
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 acquire() 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>- (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.asynchronous.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 --> 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>online Picodata nodes<br>from allowed tiers"]:::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.asynchronous.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

acquire(timeout=None)

Acquire a connection from the pool.

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

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
_PoolAcquireContext

A context manager that can be used with async with or await to acquire a connection.
Note

If you use acquire/release manually, prepared statements may be dropped when the connection is returned to the pool. To preserve prepared statements, use the connection within a single context block. This is also why Pool has no prepare() API: Pool.prepare() would have to release the connection before returning the statement to user, making it unsafe to use. So, use conn.prepare() and call it inside an acquire() block instead (see example below).

Examples:

Context manager (recommended -- safely returns the connection):

async with pool.acquire() as conn:
    await conn.execute("UPDATE ...")

Explicit acquisition (e.g., for compatibility with prepared statements):

ctx = pool.acquire()
conn = await ctx
try:
    await conn.execute(...)
finally:
    await pool.release(conn)

Prepared statement usage with connection from pool:

async with pool.acquire() as conn:
    stmt = await conn.prepare('SELECT * FROM "warehouse" WHERE id = $1')
    row = await stmt.fetchrow(1)
    rows = await stmt.fetch(2)
Source code in picopyn/asynchronous/pool.py 
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
def acquire(self, timeout: float | None = None) -> _PoolAcquireContext:
    """Acquire a connection from the pool.

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

    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 can be used with `async with` or `await` to acquire a connection.

    Note:
        If you use acquire/release manually, prepared statements may be dropped when
        the connection is returned to the pool. To preserve prepared statements, use
        the connection within a single context block. This is also why Pool has no
        `prepare()` API: `Pool.prepare()` would have to release the connection before
        returning the statement to user, making it unsafe to use. So, use
        [conn.prepare()][picopyn.asynchronous.connection.Connection.prepare] and call it
        inside an `acquire()` block instead (see example below).

    Examples:
        Context manager (recommended -- safely returns the connection):

        ```python
        async with pool.acquire() as conn:
            await conn.execute("UPDATE ...")
        ```

        Explicit acquisition (e.g., for compatibility with prepared statements):

        ```python
        ctx = pool.acquire()
        conn = await ctx
        try:
            await conn.execute(...)
        finally:
            await pool.release(conn)
        ```

        Prepared statement usage with connection from pool:

        ```python
        async with pool.acquire() as conn:
            stmt = await conn.prepare('SELECT * FROM "warehouse" WHERE id = $1')
            row = await stmt.fetchrow(1)
            rows = await stmt.fetch(2)
        ```
    """
    return _PoolAcquireContext(self, timeout)

acquire_by_instance_uuid(instance_uuid, timeout=None)

Acquire a pooled connection with instance_uuid.

This bypasses balance_strategy. If no free connection with the given UUID becomes available before the timeout (see Pool.routing_acquire_timeout), TimeoutError is raised.

Parameters:

Name Type Description Default
instance_uuid str

UUID of the picodata instance to acquire a connection to.

 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
_PoolAcquireContext

A context manager that can be used with async with or await to acquire a connection.

Examples:

async with pool.acquire_by_instance_uuid("f0b83347-6409-44dd-89e1-a9d967d0f4e6") as conn:
    await conn.execute("UPDATE ...")
Source code in picopyn/asynchronous/pool.py 
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
def acquire_by_instance_uuid(
    self,
    instance_uuid: str,
    timeout: float | None = None,
) -> _PoolAcquireContext:
    """Acquire a pooled connection with `instance_uuid`.

    This bypasses `balance_strategy`. If no free connection with the given
    UUID becomes available before the timeout (see
    [Pool.routing_acquire_timeout][picopyn.asynchronous.pool.Pool]), `TimeoutError` is raised.

    Args:
        instance_uuid: UUID of the picodata instance to acquire a connection to.
        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 `async with` or `await` to acquire a connection.

    Examples:
        ```python
        async with pool.acquire_by_instance_uuid("f0b83347-6409-44dd-89e1-a9d967d0f4e6") as conn:
            await conn.execute("UPDATE ...")
        ```
    """
    return _PoolAcquireContext(self, timeout, instance_uuid)

acquire_by_sharding_key(sharding_key, timeout=None)

Acquire 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
_PoolAcquireContext

A context manager that can be used with async with or await.

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 no matching connection becomes available before the timeout.
Source code in picopyn/asynchronous/pool.py 
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
def acquire_by_sharding_key(
    self,
    sharding_key: ShardingKey,
    timeout: float | None = None,
) -> _PoolAcquireContext:
    """Acquire 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 `async with` or `await`.

    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 no matching connection becomes available before
            the timeout.
    """
    if not isinstance(sharding_key, ShardingKey):
        raise TypeError("sharding_key must be a ShardingKey")
    return self.acquire_by_tier_and_bucket_id(
        sharding_key.tier,
        sharding_key.bucket_id,
        timeout,
    )

acquire_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 snapshot to resolve tier + bucket_id to the owning replicaset's master instance, then acquires a free pooled connection with that instance UUID.

If topology tracking is disabled, or the current topology snapshot cannot map the bucket to a master instance, RuntimeError is raised. 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
_PoolAcquireContext

A context manager that can be used with async with or await to acquire a connection.

Examples:

async with pool.acquire_by_tier_and_bucket_id("default", 42) as conn:
    await conn.execute("UPDATE ...")
Source code in picopyn/asynchronous/pool.py 
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
def acquire_by_tier_and_bucket_id(
    self,
    tier: str,
    bucket_id: int,
    timeout: float | None = None,
) -> _PoolAcquireContext:
    """Acquire a pooled connection to the master owning `bucket_id` in `tier`.

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

    If topology tracking is disabled, or the current topology snapshot
    cannot map the bucket to a master instance, `RuntimeError` is raised.
    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 `async with` or `await` to acquire a connection.

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

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

    master = topology.find_bucket_master(tier, bucket_id)
    if master 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.acquire_by_instance_uuid(master.uuid, timeout)

close(timeout=None) async

Closes all connections in the pool.

Parameters:

Name Type Description Default
timeout float | None

Maximum graceful shutdown time for the call that starts pool shutdown. If None, a default shutdown timeout 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 first timeout. If that timeout expires, any remaining connections are terminated.

Note

This should be called during application shutdown to clean up resources.

Source code in picopyn/asynchronous/pool.py 
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
async def close(self, timeout: float | None = None) -> None:
    """
    Closes all connections in the pool.

    Args:
        timeout: Maximum graceful shutdown time for the call that starts
            pool shutdown. If None, a default shutdown timeout 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 first timeout.
    If that timeout expires, any remaining connections are terminated.

    Note:
        This should be called during application shutdown to clean up resources.
    """
    close_task = self._close_task
    if close_task is None:
        close_task = asyncio.create_task(self._close_with_timeout(timeout))
        self._close_task = close_task

    if close_task.done():
        return

    await asyncio.shield(close_task)

create_sharding_key_factory(table_name) async

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/asynchronous/pool.py 
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
async 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 tracker is None:
        raise RuntimeError(
            "Cannot create sharding key factory: topology tracking is disabled "
            "or the pool is not open"
        )

    row = await self.fetchrow(ASYNC_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["distribution"],
        row["format"],
    )
    bucket_count = tracker.topology.tiers.get(table_info.tier)
    if bucket_count is None:
        await tracker.refresh_now()
        bucket_count = tracker.topology.tiers.get(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.BINARY)

execute(query, *args) async

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:

Read about cache warming.

Parameters:

Name Type Description Default
query str

The SQL query string.

 required
*args Any

Optional parameters for the SQL query.

 ()

Returns:

Type Description
str

The result of the query execution.
Source code in picopyn/asynchronous/pool.py 
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
async def execute(self, query: str, *args: Any) -> 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.asynchronous.metadata_service.QueryMetadataService] for query
    meta and [topology tracking][picopyn.asynchronous.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.asynchronous.pool.Pool])
    - topology tracking enabled (see [Pool.topology_update_interval][picopyn.asynchronous.pool.Pool])
    - Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
    - execute query more than once

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

    Args:
        query: The SQL query string.
        *args: Optional parameters for the SQL query.

    Returns:
        The result of the query execution.
    """
    conn: Connection | None = None
    # prepare and cache query meta only for parameterized queries
    if args:
        conn = await self._acquire_routed_connection(query, args)

    if conn is None:
        conn = await self.acquire()

    try:
        return await conn.execute(query, *args)
    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)
        if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
            self._query_metadata_service.evict(query)
        self._on_query_error(e)
        raise
    finally:
        await self.release(conn)

explain(query, *args, raw=False) async

Executes EXPLAIN for a query and returns a structured plan.

See Connection.explain for full documentation.

Source code in picopyn/asynchronous/pool.py 
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
async def explain(
    self, query: str, *args: Any, raw: bool = False
) -> ExplainPlan | ExplainRawPlan:
    """Executes EXPLAIN for a query and returns a structured plan.

    See [`Connection.explain`][picopyn.asynchronous.connection.Connection.explain]
    for full documentation.
    """
    async with self.acquire() as conn:
        return await conn.explain(query, *args, raw=raw)

fetch(query, *args) async

Executes a query and fetches all resulting rows.

Parameters:

Name Type Description Default
query str

The SQL query string.

 required
*args Any

Optional parameters for the SQL query.

 ()

Returns:

Type Description
list[Record]

A list of rows returned by the query.
Source code in picopyn/asynchronous/pool.py 
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
async def fetch(self, query: str, *args: Any) -> list[asyncpg.Record]:
    """Executes a query and fetches all resulting rows.

    Args:
        query: The SQL query string.
        *args: Optional parameters for the SQL query.

    Returns:
        A list of rows returned by the query.
    """
    # TODO: add cache usage when Picodata will support it for DQL
    # https://git.picodata.io/core/picodata/-/issues/2226
    async with self.acquire() as conn:
        try:
            return await conn.fetch(query, *args)
        except Exception as e:
            # TODO retry execution to hide from user the invalidation error
            if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
                self._query_metadata_service.evict(query)
            self._on_query_error(e)
            raise

fetchrow(query, *args) async

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

Parameters:

Name Type Description Default
query str

The SQL query string.

 required
*args Any

Optional parameters for the SQL query.

 ()

Returns:

Type Description
Record | None

A single row returned by the query.
Source code in picopyn/asynchronous/pool.py 
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
async def fetchrow(self, query: str, *args: Any) -> asyncpg.Record | None:
    """Executes a query and fetches a single row (first row).

    Args:
        query: The SQL query string.
        *args: Optional parameters for the SQL query.

    Returns:
        A single row returned by the query.
    """
    # TODO: add cache usage when Picodata will support it for DQL
    # https://git.picodata.io/core/picodata/-/issues/2226
    async with self.acquire() as conn:
        try:
            return await conn.fetchrow(query, *args)
        except Exception as e:
            # TODO retry execution to hide from user the invalidation error
            if self._query_metadata_service is not None and is_stmt_invalidated_error(e):
                self._query_metadata_service.evict(query)
            self._on_query_error(e)
            raise

get_query_metadata(query) async

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.asynchronous.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/asynchronous/pool.py 
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
async 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.asynchronous.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"
    """
    self._check_open()

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

    return await self._query_metadata_service.get_query_metadata(query)

open() async

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.asynchronous.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>online nodes with allowed tiers<br>as a candidate node list]
DISCOVERY -->|no| B_FILTER[Filter DSN nodes: use only online<br>nodes with allowed tiers]

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.asynchronous.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 online nodes with<br>allowed tiers"]:::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.asynchronous.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 tier]
    B_CONNECT_NODE --> B_CHECK_NODE_TIER
    B_CHECK_NODE_TIER -->|node offline or tier forbidden| B_EXCLUDE[Exclude DSN node]
    B_CHECK_NODE_TIER -->|node online and tier allowed| 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
Source code in picopyn/asynchronous/pool.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
async 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>
    """
    async with self._lock:
        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)
        # TODO: maybe we should use `_reconcile_pool_with_topology` instead
        if self._enable_discovery:
            await self._fill_pool_from_discovery()
        else:
            await 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:
                    await conn.close()
                except Exception as e:
                    logger.warning("Pool open 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", len(self._pool))

        if self._topology_update_interval is not None and self._topology_tracker is None:
            # TODO: the very first technical connection always goes to the first
            # reachable DSN node: picking a node by load needs the topology, and
            # reading the topology needs a connection, so at startup there is no
            # choice but the DSN. maybe we should reopen it right after the first
            # topology refresh, so the least loaded node is picked from the start
            self._topology_tracker = TopologyTracker(
                PollingTopologySource(self._get_technical_connection),
                self._topology_update_interval,
            )
            # register the pool's reconcile callback so connections are updated
            # to match the topology after each refresh
            await 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:
            # 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,
            )

            await self._query_metadata_service.start()
            logger.debug(
                "Pool query-metadata service started with cache size %s",
                self._query_metadata_cache_size,
            )
        return

release(conn) async

Release a previously acquired connection back to the pool.

Parameters:

Name Type Description Default
conn Connection

The connection to release.

 required
Source code in picopyn/asynchronous/pool.py 
1374
1375
1376
1377
1378
1379
1380
async def release(self, conn: Connection) -> None:
    """Release a previously acquired connection back to the pool.

    Args:
        conn: The connection to release.
    """
    await asyncio.shield(self._release(conn))

terminate()

Terminate all connections owned by the pool without graceful close.

Source code in picopyn/asynchronous/pool.py 
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
def terminate(self) -> None:
    """Terminate all connections owned by the pool without graceful close."""
    close_task = self._close_task
    if close_task is not None and close_task.done():
        return

    done = asyncio.get_running_loop().create_future()
    done.set_result(None)
    self._close_task = done

    try:
        current_task = asyncio.current_task()
    except RuntimeError:
        current_task = None
    if close_task is not None and close_task is not current_task:
        close_task.cancel()

    self._terminate_services()

    conns = [*self._pool, *self._used]
    self._pool.clear()

    for conn in conns:
        try:
            conn.terminate()
        except Exception as e:
            logger.warning("Pool terminate: could not terminate connection: %s", e)

    for conn in list(self._used):
        self._finish_release(conn)