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
**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
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.
        **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,
        **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,
            **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 
166
167
168
169
170
171
172
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 
77
78
79
80
81
82
83
84
85
86
87
88
89
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 
 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
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 
156
157
158
159
160
161
162
163
164
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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
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
**_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.

Source code in picopyn/asynchronous/connection.py 
 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
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.
        **_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.
    """

    def __init__(self, dsn: str, **_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

    def is_closed(self) -> bool:
        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 new connection to Picodata
        """
        if self.conn and not self.is_closed():
            await self.close()

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

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

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

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

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

close(*args, **kwargs) async

Close the connection gracefully.

Source code in picopyn/asynchronous/connection.py 
139
140
141
142
143
144
145
146
147
148
149
async def close(self, *args: Any, **kwargs: Any) -> None:
    """
    Close the connection gracefully.
    """
    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 new connection to Picodata

Source code in picopyn/asynchronous/connection.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
async def connect(self) -> None:
    """
    Create new connection to Picodata
    """
    if self.conn and not self.is_closed():
        await self.close()

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

execute(*args, **kwargs) async

Execute an SQL command

Source code in picopyn/asynchronous/connection.py 
64
65
66
67
68
69
70
71
72
73
74
75
async def execute(self, *args: Any, **kwargs: Any) -> str:
    """
    Execute an SQL command
    """

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

Source code in picopyn/asynchronous/connection.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def fetch(self, *args: Any, **kwargs: Any) -> list[asyncpg.Record]:
    """
    Run a query and return the results as a list.
    """

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

Source code in picopyn/asynchronous/connection.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def fetchrow(self, *args: Any, **kwargs: Any) -> asyncpg.Record | None:
    """
    Run a query and return the first row.
    """

    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:
        raise RuntimeError(
            f"Failed to execute SQL query and fetch row: {e}. Query: {args}"
        ) from e

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.

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

CYCLE_NOTE>"repeated for each 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. On each cycle the pool queries picodata for the current set of online instances. Set to None to disable periodic updates. Default is 60 seconds.

 60.0
**connect_kwargs Any

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

 {}

Examples:

Pool with custom balance strategy:

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

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

Pool with multi-host DSN string. Nodes are iterated in round-robin until a connection succeeds, enabling cluster discovery:

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

SSL (kwargs-based):

import ssl

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

SSL (DSN query params):

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

    ```mermaid
    ---
    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

    CYCLE_NOTE>"repeated for each 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
    ```

    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.
            On each cycle the pool queries picodata for the current set of online instances.
            Set to None to disable periodic updates.
            Default is 60 seconds.
        **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,
        **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
        self._pool: deque[Connection] = deque()
        self._used: set[Connection] = set()
        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
        # 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
        # currently topology tracking is purely observational (see picopyn.topology)
        # it never changes pool membership, and is independent of enable_discovery so it
        # also works in bootstrap mode
        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})"

    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.

        ```mermaid
        ---
        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 Picodata cluster and retrieve online nodes with allowed tiers as a candidate node list]
        DISCOVERY -->|no| B_FILTER[Filter DSN nodes: use only online nodes with allowed tiers]

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

            D_DETAILS_NOTE>"Discovery mode details"]:::note
            click D_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_discovery" "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 to DSN nodes up to pool size]

            B_DETAILS_NOTE>"Bootstrap mode details"]:::note
            click B_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_bootstrap_dsns" "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
        ```
        """
        async with self._lock:
            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)
            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,
                )
                await self._topology_tracker.start()
            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`.

        ```mermaid
        ---
        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 DSN node list] --> DiscoverNodes
        PARSE <-. write .-> D_DSN

        D_DiscoverNodes_NOTE>"iterate over DSN nodes<br>to establish a temporary connection<br>and retrieve the list of online nodes with 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 until either:<br>- the list is empty<br>- pool_size connections 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
        ```

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

        ```mermaid
        ---
        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 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
        ```
        """
        available = list(self._connect_dsns)
        idx = 0
        while len(self._pool) < self._max_size and available:
            candidate = available[idx % len(available)]
            logger.debug("Bootstrap by DSN: connecting to %s", _dsn_hostinfo(candidate))
            try:
                conn = Connection(candidate, **self._connect_kwargs)
                await conn.connect()
                # TODO we need to check tier only once for one node
                if self._forbidden_tiers:
                    tier = await self._fetch_tier(conn)
                    if tier and tier in self._forbidden_tiers:
                        logger.debug(
                            "Bootstrap by DSN: skipping %s: tier %r is forbidden",
                            _dsn_hostinfo(candidate),
                            tier,
                        )
                        await conn.close()
                        available.remove(candidate)
                        continue
                self._pool.append(conn)
                idx += 1
            except Exception as e:
                logger.warning(
                    "Bootstrap by DSN: could not connect to %s: %s",
                    _dsn_hostinfo(candidate),
                    e,
                )
                available.remove(candidate)

    async def _fetch_tier(self, conn: Connection) -> str | None:
        row = await conn.fetchrow(CURRENT_INSTANCE_TIER_QUERY)
        return row["tier"] if row else None

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

            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) -> Connection | None:
        # try each bootstrap DSN until one succeeds
        last_error: Exception | None = None
        for dsn in self._connect_dsns:
            try:
                candidate = Connection(dsn, **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:
                # сheck 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._used.add(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 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.

        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)
            ```
        """
        return _PoolAcquireContext(self, timeout)

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

        Args:
            conn: The connection to release.
        """
        async with self._lock:
            if conn not in self._used:
                return
            self._used.remove(conn)
            self._pool.append(conn)

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

        Note:
            This should be called during application shutdown to clean up resources.
        """
        if self._topology_tracker is not None:
            await self._topology_tracker.stop()

        async with self._lock:
            total = len(self._pool) + len(self._used)
            while self._pool:
                conn = self._pool.popleft()
                try:
                    await conn.close()
                except Exception as e:
                    logger.warning("Pool close: could not close connection: %s", e)
            for conn in self._used:
                try:
                    await conn.close()
                except Exception as e:
                    logger.warning("Pool close: could not close connection: %s", e)
            self._used.clear()
            logger.info("Pool closed (%d connection(s))", total)

    @property
    def topology(self) -> Topology | None:
        """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.

        See [`Topology`][picopyn.topology] for details.
        """
        return self._topology_tracker.topology if self._topology_tracker is not None else None

    def _on_query_error(self, exc: BaseException) -> None:
        """If `exc` 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 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.
        """
        async with self.acquire() as conn:
            try:
                return await conn.execute(query, *args)
            except Exception as e:
                self._on_query_error(e)
                raise

    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.
        """
        async with self.acquire() as conn:
            try:
                return await conn.fetch(query, *args)
            except Exception as e:
                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.
        """
        async with self.acquire() as conn:
            try:
                return await conn.fetchrow(query, *args)
            except Exception as e:
                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)

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.

See Topology for details.

_fill_pool_from_bootstrap_dsns() async

Fill the pool with nodes from DSNs if they pass filters.

Current filter is forbidden_tiers.

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

    ```mermaid
    ---
    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 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
    ```
    """
    available = list(self._connect_dsns)
    idx = 0
    while len(self._pool) < self._max_size and available:
        candidate = available[idx % len(available)]
        logger.debug("Bootstrap by DSN: connecting to %s", _dsn_hostinfo(candidate))
        try:
            conn = Connection(candidate, **self._connect_kwargs)
            await conn.connect()
            # TODO we need to check tier only once for one node
            if self._forbidden_tiers:
                tier = await self._fetch_tier(conn)
                if tier and tier in self._forbidden_tiers:
                    logger.debug(
                        "Bootstrap by DSN: skipping %s: tier %r is forbidden",
                        _dsn_hostinfo(candidate),
                        tier,
                    )
                    await conn.close()
                    available.remove(candidate)
                    continue
            self._pool.append(conn)
            idx += 1
        except Exception as e:
            logger.warning(
                "Bootstrap by DSN: could not connect to %s: %s",
                _dsn_hostinfo(candidate),
                e,
            )
            available.remove(candidate)

_fill_pool_from_discovery() async

Fill the pool with online nodes from Picodata cluster if they pass filters. Current filter is forbidden_tiers.

---
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 DSN node list] --> DiscoverNodes
PARSE <-. write .-> D_DSN

D_DiscoverNodes_NOTE>"iterate over DSN nodes<br>to establish a temporary connection<br>and retrieve the list of online nodes with 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 until either:<br>- the list is empty<br>- pool_size connections 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
Source code in picopyn/asynchronous/pool.py 
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
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`.

    ```mermaid
    ---
    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 DSN node list] --> DiscoverNodes
    PARSE <-. write .-> D_DSN

    D_DiscoverNodes_NOTE>"iterate over DSN nodes<br>to establish a temporary connection<br>and retrieve the list of online nodes with 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 until either:<br>- the list is empty<br>- pool_size connections 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
    ```

    """
    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()
            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

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.

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)
Source code in picopyn/asynchronous/pool.py 
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
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.

    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)
        ```
    """
    return _PoolAcquireContext(self, timeout)

close() async

Closes all connections in the pool.

Note

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

Source code in picopyn/asynchronous/pool.py 
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
async def close(self) -> None:
    """
    Closes all connections in the pool.

    Note:
        This should be called during application shutdown to clean up resources.
    """
    if self._topology_tracker is not None:
        await self._topology_tracker.stop()

    async with self._lock:
        total = len(self._pool) + len(self._used)
        while self._pool:
            conn = self._pool.popleft()
            try:
                await conn.close()
            except Exception as e:
                logger.warning("Pool close: could not close connection: %s", e)
        for conn in self._used:
            try:
                await conn.close()
            except Exception as e:
                logger.warning("Pool close: could not close connection: %s", e)
        self._used.clear()
        logger.info("Pool closed (%d connection(s))", total)

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

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

    D_DETAILS_NOTE>"Discovery mode details"]:::note
    click D_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_discovery" "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 to DSN nodes up to pool size]

    B_DETAILS_NOTE>"Bootstrap mode details"]:::note
    click B_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_bootstrap_dsns" "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
Source code in picopyn/asynchronous/pool.py 
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
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.

    ```mermaid
    ---
    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 Picodata cluster and retrieve online nodes with allowed tiers as a candidate node list]
    DISCOVERY -->|no| B_FILTER[Filter DSN nodes: use only online nodes with allowed tiers]

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

        D_DETAILS_NOTE>"Discovery mode details"]:::note
        click D_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_discovery" "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 to DSN nodes up to pool size]

        B_DETAILS_NOTE>"Bootstrap mode details"]:::note
        click B_DETAILS_NOTE "#picopyn.asynchronous.Pool._fill_pool_from_bootstrap_dsns" "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
    ```
    """
    async with self._lock:
        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)
        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,
            )
            await self._topology_tracker.start()
        return

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.
Source code in picopyn/asynchronous/pool.py 
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
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.
    """
    async with self.acquire() as conn:
        try:
            return await conn.execute(query, *args)
        except Exception as e:
            self._on_query_error(e)
            raise

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 
808
809
810
811
812
813
814
815
816
817
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 
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
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.
    """
    async with self.acquire() as conn:
        try:
            return await conn.fetch(query, *args)
        except Exception as e:
            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 
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
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.
    """
    async with self.acquire() as conn:
        try:
            return await conn.fetchrow(query, *args)
        except Exception as e:
            self._on_query_error(e)
            raise

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 
694
695
696
697
698
699
700
701
702
703
704
async def release(self, conn: Connection) -> None:
    """Release a previously acquired connection back to the pool.

    Args:
        conn: The connection to release.
    """
    async with self._lock:
        if conn not in self._used:
            return
        self._used.remove(conn)
        self._pool.append(conn)