Skip to content

Async

picopyn.asynchronous

Async API based on asyncpg for connection management, pools, and client classes.

Client

Async client for managing connections to a picodata cluster using a connection pool.

This client handles connection pooling, automatic cluster node discovery, and supports load balancing strategies for query distribution. See details about Pool.

Parameters:

Name Type Description Default
dsn str

The data source name (e.g., "postgresql://user:pass@host:port") for the cluster.

 required
pool_size int | None

Maximum number of connections in the pool. Must be at least 1. Default is 10.

 None
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 connection is forbidden (e.g., "arbiter,readonly"). If None, connections to all tiers are allowed.

 None
topology_update_interval float | None

Set to None to disable periodic topology updates. Default is 60 seconds. See details at Pool.topology_update_interval.

 60.0
query_metadata_cache_size int

Default is 100. See details at Pool.query_metadata_cache_size.

 100
**connect_kwargs Any

Additional keyword arguments passed to each connection.

 {}

Examples:

Client with random balance strategy:

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

client = Client(
    dsn="postgresql://admin:pass@localhost:5432",
    balance_strategy=random_strategy,
)

Client with strategy always choosing first connection:

def custom_strategy(pool_conns):
    return pool_conns[0]

client = Client(
    dsn="postgresql://admin:pass@localhost:5432",
    balance_strategy=custom_strategy,
)

Client with multi-host DSN string:

client = Client(dsn="postgresql://admin:pass@host1:5432,host2:5432")
Source code in picopyn/asynchronous/client.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 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
class Client:
    """Async client for managing connections to a picodata cluster using a connection pool.

    This client handles connection pooling, automatic cluster node discovery,
    and supports load balancing strategies for query distribution. See details about [Pool][picopyn.asynchronous.Pool].

    Args:
        dsn: The data source name (e.g., "postgresql://user:pass@host:port") for the cluster.
        pool_size: Maximum number of connections in the pool. Must be at least 1. Default is 10.
        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 connection
            is forbidden (e.g., "arbiter,readonly"). If None, connections to all tiers are allowed.
        topology_update_interval: Set to None to disable periodic topology updates. Default is 60
            seconds. See details at [Pool.topology_update_interval][picopyn.asynchronous.Pool].
        query_metadata_cache_size: Default is 100. See details at
            [Pool.query_metadata_cache_size][picopyn.asynchronous.Pool].
        **connect_kwargs: Additional keyword arguments passed to each connection.

    Examples:
        Client with random balance strategy:

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

        client = Client(
            dsn="postgresql://admin:pass@localhost:5432",
            balance_strategy=random_strategy,
        )
        ```

        Client with strategy always choosing first connection:

        ```python
        def custom_strategy(pool_conns):
            return pool_conns[0]

        client = Client(
            dsn="postgresql://admin:pass@localhost:5432",
            balance_strategy=custom_strategy,
        )
        ```

        Client with multi-host DSN string:

        ```python
        client = Client(dsn="postgresql://admin:pass@host1:5432,host2:5432")
        ```
    """

    def __init__(
        self,
        dsn: str,
        pool_size: int | None = None,
        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 = 100,
        **connect_kwargs: Any,
    ) -> None:
        self._pool = Pool(
            dsn=dsn,
            max_size=pool_size or 10,
            enable_discovery=True,
            balance_strategy=balance_strategy,
            forbidden_tiers=forbidden_tiers,
            topology_update_interval=topology_update_interval,
            query_metadata_cache_size=query_metadata_cache_size,
            **connect_kwargs,
        )

    async def connect(self) -> None:
        """Prepares the client by connecting the connection pool.

        Note:
            This should be called before using the client to ensure connections are available.

        Examples:
            ```python
            client = Client(dsn="postgresql://admin:pass@localhost:5432")
            await client.connect()
            ```
        """
        await self._pool.connect()

    async def execute(self, query: str, *args: Any) -> str:
        """Executes a query that does not return rows (e.g. INSERT, UPDATE, DELETE).

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

        Returns:
            The result of the query execution.

        Examples:
            DDL example:

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

            DML example:

            ```python
            dml = 'INSERT INTO "warehouse" VALUES ($1::int, $2::varchar)'
            await client.execute(dml, 1, "test")
            # 'INSERT 0 1'
            ```
        """
        return await self._pool.execute(query, *args)

    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.

        Examples:
            ```python
            rows = await client.fetch('SELECT * FROM "warehouse";')
            # [<Record id=1 item='test'>]
            ```
        """
        return await self._pool.fetch(query, *args)

    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.

        Examples:
            ```python
            row = await client.fetchrow('SELECT * FROM "warehouse";')
            # <Record id=1 item='test'>
            ```
        """
        return await self._pool.fetchrow(query, *args)

    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.
        """
        return await self._pool.explain(query, *args, raw=raw)

    async def close(self) -> None:
        """
        Closes all connections in the pool.

        This should be called during application shutdown to clean up resources.
        """
        await self._pool.close()

close() async

Closes all connections in the pool.

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

Source code in picopyn/asynchronous/client.py 
174
175
176
177
178
179
180
async def close(self) -> None:
    """
    Closes all connections in the pool.

    This should be called during application shutdown to clean up resources.
    """
    await self._pool.close()

connect() async

Prepares the client by connecting the connection pool.

Note

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

Examples:

client = Client(dsn="postgresql://admin:pass@localhost:5432")
await client.connect()
Source code in picopyn/asynchronous/client.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
async def connect(self) -> None:
    """Prepares the client by connecting the connection pool.

    Note:
        This should be called before using the client to ensure connections are available.

    Examples:
        ```python
        client = Client(dsn="postgresql://admin:pass@localhost:5432")
        await client.connect()
        ```
    """
    await self._pool.connect()

execute(query, *args) async

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

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.

Examples:

DDL example:

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

DML example:

dml = 'INSERT INTO "warehouse" VALUES ($1::int, $2::varchar)'
await client.execute(dml, 1, "test")
# 'INSERT 0 1'
Source code in picopyn/asynchronous/client.py 
 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
async def execute(self, query: str, *args: Any) -> str:
    """Executes a query that does not return rows (e.g. INSERT, UPDATE, DELETE).

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

    Returns:
        The result of the query execution.

    Examples:
        DDL example:

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

        DML example:

        ```python
        dml = 'INSERT INTO "warehouse" VALUES ($1::int, $2::varchar)'
        await client.execute(dml, 1, "test")
        # 'INSERT 0 1'
        ```
    """
    return await self._pool.execute(query, *args)

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/client.py 
164
165
166
167
168
169
170
171
172
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.
    """
    return await self._pool.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.

Examples:

rows = await client.fetch('SELECT * FROM "warehouse";')
# [<Record id=1 item='test'>]
Source code in picopyn/asynchronous/client.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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.

    Examples:
        ```python
        rows = await client.fetch('SELECT * FROM "warehouse";')
        # [<Record id=1 item='test'>]
        ```
    """
    return await self._pool.fetch(query, *args)

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.

Examples:

row = await client.fetchrow('SELECT * FROM "warehouse";')
# <Record id=1 item='test'>
Source code in picopyn/asynchronous/client.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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.

    Examples:
        ```python
        row = await client.fetchrow('SELECT * FROM "warehouse";')
        # <Record id=1 item='test'>
        ```
    """
    return await self._pool.fetchrow(query, *args)

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 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 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
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, 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.

        Examples:
            ```python
            plan = await conn.explain("SELECT * FROM warehouse")
            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 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
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 
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
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, 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 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
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.

Examples:

plan = await conn.explain("SELECT * FROM warehouse")
raw_plan = await conn.explain("SELECT * FROM warehouse", raw=True)
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
331
332
333
334
335
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.

    Examples:
        ```python
        plan = await conn.explain("SELECT * FROM warehouse")
        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 
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 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 
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
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 
 97
 98
 99
100
101
102
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 
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
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 
355
356
357
358
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.

---
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.connect()"] --> UsePool
CONNECT <-. write .-> POOL

CONNECT_DETAILS_NOTE>"Connect algorithm details"]:::note
click CONNECT_DETAILS_NOTE "#picopyn.asynchronous.Pool.connect" "Connect algorithm details"
CONNECT_DETAILS_NOTE -.-> CONNECT

RECONCILE_NOTE>"(discovery mode only):<br>pool membership changes<br>in the background:<br>connections are opened/closed<br>to match topology changes"]:::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()"] --> USE["Execute queries<br>· pool.fetchrow()<br>· pool.fetch()<br>· pool.execute()"] --> RELEASE["pool.release(conn)"]
end

ACQUIRE <-. delete .-> POOL
RELEASE <-. write .-> 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. Default is 10.

 10
enable_discovery bool

If True, the pool will automatically discover available picodata instances. If False, only the given dsn will be used.

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

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

 None
forbidden_tiers str | None

A comma-separated list of Picodata node tiers for which connection is forbidden (e.g., "arbiter,readonly"). If None, connections to all tiers are allowed.

 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. Default is 60 seconds.

 60.0
query_metadata_cache_size int

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. Default is 100.

 100
**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 
 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
 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
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].

    --8<-- "_pool_lifecycle.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. Default is 10.
        enable_discovery: If True, the pool will automatically discover available picodata
            instances. If False, only the given `dsn` will be used.
        balance_strategy: A custom strategy function to select a connection from the pool.
            If None, round-robin strategy is used.
        forbidden_tiers: A comma-separated list of Picodata node tiers for which connection
            is forbidden (e.g., "arbiter,readonly"). If None, connections to all tiers are allowed.
        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.
            Default is 60 seconds.
        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. Default is 100.
        **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 = 100,
        **connect_kwargs: Any,
    ) -> None:
        if max_size < 1:
            raise ValueError("max_size 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")

        # 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
        # Dedicated 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_dsn_connection(on_query_metadata),
            cache_size=query_metadata_cache_size,
        )
        self._pool: deque[Connection] = deque()
        self._used: set[Connection] = set()
        self._instance_uuid_by_address: dict[str, str] = {}
        self._forbidden_tiers = (
            set(t.strip() for t in forbidden_tiers.split(",") if t.strip())
            if forbidden_tiers
            else set()
        )
        self._lock: asyncio.Lock = asyncio.Lock()
        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, but only changes pool membership when
        # discovery mode enabled -- see _reconcile_pool_with_topology, wired up as the
        # topology tracker's refresh callback below. With discovery disabled, tracking
        # stays purely observational
        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 _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 connect(self) -> None:
        """
        Prepares the pool by opening 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 connect details in discovery mode</summary>
        --8<-- "_pool_init_discovery_enabled.md"
        </details>

        <details id="pool-init-discovery-disabled"><summary>See connect details in bootstrap mode</summary>
        --8<-- "_pool_init_discovery_disabled.md"
        </details>
        """
        async with self._lock:
            self._check_open()
            if len(self._pool) == 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 = len(self._pool)
            if conn_count < self._max_size:
                while self._pool:
                    conn = self._pool.popleft()
                    try:
                        await conn.close()
                    except Exception as e:
                        logger.warning("Pool connect cleanup: could not close connection: %s", e)
                raise RuntimeError(
                    f"Failed to initialize connection pool: only {conn_count} "
                    f"out of {self._max_size} connections established for DSN "
                    f"{self._redacted_dsn}"
                )

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

            logger.info("Pool initialized with %d connections", len(self._pool))

            if self._topology_update_interval is not None:
                self._topology_tracker = TopologyTracker(
                    PollingTopologySource(self._get_dsn_connection),
                    self._topology_update_interval,
                )
                # if discovery mode enabled, 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 if self._enable_discovery else None
                    )
                )
                logger.debug("Pool topology tracker started")

            await self._query_metadata_service.start()
            logger.debug("Pool query-metadata service started")
            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

        addr_index = 0
        # 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 will be skipped and removed from the list.
        # the loop will exit early if no nodes remain to avoid an infinite loop.
        while len(self._pool) < self._max_size and instance_addrs:
            address = instance_addrs[addr_index % len(instance_addrs)]
            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)
                instance_addrs.remove(address)
                if not instance_addrs:
                    break
                continue

            addr_index += 1

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

        Current filter is `forbidden_tiers`.
        """
        available = list(self._connect_dsns)
        instance_info_by_address: dict[str, InstanceInfo] | None = None
        idx = 0
        while len(self._pool) < self._max_size and available:
            candidate = available[idx % len(available)]
            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.remove(candidate)
                        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.remove(candidate)
                        continue
                    _, instance_uuid = instance_info

                conn.instance_uuid = instance_uuid
                self._pool.append(conn)
                idx += 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.remove(candidate)

    def _is_accepted_bootstrap_candidate(
        self,
        instance_info: InstanceInfo | None,
        address: str,
    ) -> TypeGuard[tuple[str | None, str]]:
        if instance_info is None:
            logger.warning(
                "Bootstrap by DSN: could not find metadata for %s in bootstrap instance info",
                address,
            )
            return False

        tier, instance_uuid = instance_info
        if instance_uuid is None:
            logger.warning(
                "Bootstrap by DSN: could not determine instance uuid for %s",
                address,
            )
            return False

        if self._forbidden_tiers and tier and tier in self._forbidden_tiers:
            logger.debug(
                "Bootstrap by DSN: skipping %s: tier %r is forbidden",
                address,
                tier,
            )
            return False

        return True

    def _pick_refill_candidate(self, candidate_list: list[str]) -> str | None:
        """Pick the uuid from candidate nodes with the fewest current connections.
        For example, if the pool contains connections to (i1, i1, i2), the choice
        from candidates (i1, i2, i3) will be i3, and from (i1, i2) will be i2.
        This keeps connections balanced across nodes.

        Counts existing connections (free + in-use) per uuid, then returns the
        uuid with the minimum count, or, in case of a tie, the first one in
        `candidate_list`'s order.
        Returns `None` if `candidate_list` is empty.
        """
        if not candidate_list:
            return None
        candidate_set = set(candidate_list)
        counts = Counter(
            conn.instance_uuid
            for conn in (*self._pool, *self._used)
            if conn.instance_uuid in candidate_set
        )
        return min(candidate_list, key=lambda uuid: counts[uuid])

    async def _reconcile_pool_with_topology(self, topology: Topology) -> None:  # noqa: C901
        """
        Change the pool's connection list according to the current topology.
        """
        try:
            self._check_open()
        except RuntimeError:
            logger.debug("Reconcile: pool is closing/closed, skipping")
            return

        # collect candidate node list for pool connections
        candidate_list = [
            instance_uuid
            for instance_uuid, instance in topology.instances.items()
            if instance.current_state == "Online"
            and instance.replicaset_uuid in topology.replicasets
            and topology.replicasets[instance.replicaset_uuid].tier not in self._forbidden_tiers
        ]
        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",
            len(self._pool) + len(self._used),
            self._max_size,
        )
        while len(self._pool) + len(self._used) < self._max_size and candidate_list:
            # 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
            async with self._lock:
                self._pool.append(conn)
        logger.debug(
            "Reconcile: done with %s/%s pool connections",
            len(self._pool) + len(self._used),
            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 _fetch_current_instance_info(self, conn: Connection) -> tuple[str | None, str | None]:
        row = await conn.fetchrow(CURRENT_INSTANCE_INFO_QUERY)
        if row is None:
            return None, None

        instance_uuid = str(row["instance_uuid"]) if row["instance_uuid"] is not None else None
        tier = row["tier"] if isinstance(row["tier"], str) else None
        return tier, instance_uuid

    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:
            alive_instances_info = await self._fetch_discovery_rows(temp_conn)
            online_addresses, malformed_rows = _parse_alive_instance_addresses(
                cast(Iterable[DiscoveryRow], alive_instances_info)
            )
            online_address_set = set(online_addresses)
            self._instance_uuid_by_address = {}
            for row in cast(Iterable[DiscoveryRow], alive_instances_info):
                address = row.get("address")
                instance_uuid = row.get("instance_uuid")
                if (
                    isinstance(address, str)
                    and address in online_address_set
                    and instance_uuid is not None
                ):
                    self._instance_uuid_by_address[address] = str(instance_uuid)

            for row in malformed_rows:
                logger.warning("Failed to decode discovery row of picodata instance %s", row)

            logger.debug("Discovered %d instance(s): %s", len(online_addresses), online_addresses)

            if not online_addresses:
                if self._forbidden_tiers:
                    raise ValueError(
                        f"No online nodes available after applying forbidden_tiers filter: {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

    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

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

            # if no connections are available, wait briefly before retrying
            # this gives other coroutines (like `release`) a chance to return a connection to the pool
            await asyncio.sleep(0.1)

    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 _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.
        """
        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

            if (time.monotonic() - start_time) >= effective_timeout:
                raise TimeoutError(
                    "Timed out waiting for a free connection with "
                    f"instance_uuid {instance_uuid!r} in the pool"
                )

            await asyncio.sleep(0.1)

    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, `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 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)

        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)

        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 = len(self._pool) + len(self._used)
            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)
        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(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()

        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)
        try:
            self._query_metadata_service.terminate()
        except Exception as e:
            logger.warning("Pool terminate: could not terminate query-metadata service: %s", e)

        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:

        - discovery mode (see [Pool.enable_discovery][picopyn.asynchronous.pool.Pool])
        - topology tracking enabled (see [Pool.topology_update_interval][picopyn.asynchronous.pool.Pool])

        Reconcile runs on the tracker's refresh, so disabling either one disables
        it. Otherwise the topology is tracked for observation and
        [shard-aware routing][picopyn.asynchronous.pool.Pool.acquire_by_tier_and_bucket_id]
        only -- pool membership stays as it was built from `dsn`.

        The reconcile algorithm targets the following desired pool composition:

        - The pool is always filled up to its max size (if possible)
        - The reconciliation results in a balanced connection list (if possible)

        --8<-- "_reconcile_pool_with_topology.md"

        <details><summary>Example reconcile 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"
        </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:
        topology = self.topology
        if topology is None or not topology.tiers:
            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

        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

        try:
            return await self.acquire_by_instance_uuid(master.uuid, timeout=0)
        except TimeoutError:
            return None

    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:

        - cache enabled (always)
        - 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:
            if 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:
                if 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:
                if 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.

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

        --8<-- "_query_metadata_pool_init.md"

        --8<-- "_query_metadata_usage_lookup.md"
        """
        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:

Reconcile runs on the tracker's refresh, so disabling either one disables it. Otherwise the topology is tracked for observation and shard-aware routing only -- pool membership stays as it was built from dsn.

The reconcile algorithm targets the following desired pool composition:

  • The pool is always filled up to its max size (if possible)
  • The reconciliation results in a balanced connection list (if possible)
---
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"]
    Settings[/"forbidden tiers=[...]<br>max size=N"/]
    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 --> GET_LIST["build candidate_list from Topology:<br>- uuids with allowed tiers<br>- online uuids"]
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| POOL_OK(["Pool okay<br>(exit)"]):::success
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| POOL_OK2(["Pool okay<br>(exit)"]):::success
CHECK_SIZE_LOOP -->|yes| MIN
Example reconcile 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

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 
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
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, 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 
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
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, `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_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 
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
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 
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
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)

connect() async

Prepares the pool by opening 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 connect 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>· pool_size<br>· forbidden tiers<br>· cluster discovery enabled"/]

GENERAL_DETAILS_NOTE>"Return to general algorithm"]:::note
click GENERAL_DETAILS_NOTE "#picopyn.asynchronous.Pool.connect" "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>- pool_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 connect 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.connect" "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>- pool_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 
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
async def connect(self) -> None:
    """
    Prepares the pool by opening 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 connect details in discovery mode</summary>
    --8<-- "_pool_init_discovery_enabled.md"
    </details>

    <details id="pool-init-discovery-disabled"><summary>See connect details in bootstrap mode</summary>
    --8<-- "_pool_init_discovery_disabled.md"
    </details>
    """
    async with self._lock:
        self._check_open()
        if len(self._pool) == 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 = len(self._pool)
        if conn_count < self._max_size:
            while self._pool:
                conn = self._pool.popleft()
                try:
                    await conn.close()
                except Exception as e:
                    logger.warning("Pool connect cleanup: could not close connection: %s", e)
            raise RuntimeError(
                f"Failed to initialize connection pool: only {conn_count} "
                f"out of {self._max_size} connections established for DSN "
                f"{self._redacted_dsn}"
            )

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

        logger.info("Pool initialized with %d connections", len(self._pool))

        if self._topology_update_interval is not None:
            self._topology_tracker = TopologyTracker(
                PollingTopologySource(self._get_dsn_connection),
                self._topology_update_interval,
            )
            # if discovery mode enabled, 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 if self._enable_discovery else None
                )
            )
            logger.debug("Pool topology tracker started")

        await self._query_metadata_service.start()
        logger.debug("Pool query-metadata service started")
        return

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:

  • cache enabled (always)
  • topology tracking enabled (see Pool.topology_update_interval)
  • Picodata can compute distribution key metadata for executed query (supported: parameterized 1-row INSERT)
  • execute query more than once

Read about cache warming.

Parameters:

Name Type Description Default
query 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 
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
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:

    - cache enabled (always)
    - 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:
        if 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 
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
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 
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
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:
            if 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 
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
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:
            if 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.

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

---
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 --> CACHE_EMPTY["Create empty query meta cache"] --> CONNECT["Fill the pool with ordinary<br>connections to picodata"] --> Pool1

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

CONNECT_DETAILS_NOTE>"Connect algorithm details"]:::note
click CONNECT_DETAILS_NOTE "#picopyn.asynchronous.Pool.connect" "Connect 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)]

    CACHE1[/"Query meta cache"/]
end

Pool1 --> CONNECT_SERV["create service connection<br>for meta listener"]

NOTE_CONN_SERV>"connection options require 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 with Notice listener"]
    PICO1 -.- PIC
    PICO2 -.- PIC
    PICON -.- PIC

    CACHE[/"Query meta cache"/]
end
---
title: Query metadata -- usage lookup
---

flowchart TD

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


QUERY(["parametrized query"]):::neutral --> CALL(["Pool.execute"]) --> HASH

HASH["hash the query text"] --> SEARCH_CACHE{"cached meta by hash?"}

CACHE1[/"Pool query meta cache"/] -. read .- SEARCH_CACHE
SEARCH_CACHE -->|yes| USE_CACHE["metadata found in cache"]
SEARCH_CACHE -->|no| CACHE_MISS["no metadata yet --<br>schedule background prepare"] --> ACQUIRE_CONN

subgraph Routing["BUCKET ROUTING"]
    direction TB

    CALC["calculate `bucket_id`<br>from dk_meta"]

    CALC --> MAP_CONN["find bucket master<br>for this tier"]

    MAP_CONN -->|routed| CALL_THE_CONN["run the query on<br>the chosen connection"]
    MAP_CONN -->|cannot route| ACQUIRE_CONN

    CALL_THE_CONN --> RESULT(["give result to the user"]):::neutral
end

USE_CACHE --> CALC

subgraph Fallback["FALLBACK"]
    direction TB

    ACQUIRE_CONN["acquire pool connection<br>via the configured balance strategy"]

    ACQUIRE_CONN --> CALL_CONN["run the query on<br>the chosen connection"]

    CALL_CONN --> RESULT2(["give result to the user"]):::neutral
end
Source code in picopyn/asynchronous/pool.py 
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
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.

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

    --8<-- "_query_metadata_pool_init.md"

    --8<-- "_query_metadata_usage_lookup.md"
    """
    return await self._query_metadata_service.get_query_metadata(query)

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 
911
912
913
914
915
916
917
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 
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
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()

    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)
    try:
        self._query_metadata_service.terminate()
    except Exception as e:
        logger.warning("Pool terminate: could not terminate query-metadata service: %s", e)

    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)