Asyncpg

https://github.com/magicstack/asyncpg
Admin
A fast PostgreSQL Database Client Library for Python/asyncio.
Tokens:122,037
Snippets:898
Trust Score:7.9
License:Apache-2.0
Update:1 week ago
Show doc for...
Tokens:
Raw
### Start a transaction with conn.transaction()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Creates a Transaction object for use as an async context manager. The example starts a serializable, read-only transaction and executes an UPDATE.

```python
async with conn.transaction(isolation='serializable', readonly=True):
    await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
```

--------------------------------

### Build asyncpg from source (editable install)

Source: https://github.com/magicstack/asyncpg/blob/master/docs/installation.rst

Install asyncpg from a Git checkout in editable mode. Requires a cloned repo with --recurse-submodules, a C compiler, and CPython header files.

```bash
$ pip install -e .
```

--------------------------------

### Get connection settings with conn.get_settings()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Retrieves the ConnectionSettings object exposing server startup settings as attributes. The example prints the server version and encoding.

```python
settings = conn.get_settings()
print(settings.server_version, settings.server_encoding)
```

--------------------------------

### Temporary cluster usage example

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Shows the typical lifecycle of a temporary cluster: initialize, start, connect, run a query, then stop and destroy. The connect method returns an asyncpg connection using the cluster's connection spec.

```python
cluster = asyncpg.cluster.TempCluster()
try:
    cluster.init()
    cluster.start()
    conn = await cluster.connect(user='postgres')
    try:
        print(await conn.fetchval('SELECT 1'))
    finally:
        await conn.close()
finally:
    cluster.stop()
    cluster.destroy()
```

--------------------------------

### Install asyncpg with GSSAPI/SSPI authentication support

Source: https://github.com/magicstack/asyncpg/blob/master/README.rst

Install asyncpg with the 'gssauth' extra to enable GSSAPI/SSPI authentication. Use this command when such authentication is required.

```bash
$ pip install 'asyncpg[gssauth]'
```

--------------------------------

### Basic asyncpg.connect() example

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Connects to a PostgreSQL database using asyncpg.connect() with keyword arguments, fetches rows with a parameterized query, and closes the connection in a finally block. The example uses asyncio.run() to execute the async function.

```python
import asyncio, asyncpg

async def basic():
    conn = await asyncpg.connect(
        user='user', password='password',
        database='database', host='127.0.0.1', port=5432,
        command_timeout=30, ssl='verify-full',
        server_settings={'application_name': 'myapp'},
    )
    try:
        rows = await conn.fetch('SELECT * FROM mytable WHERE id = $1', 10)
        print(rows)
    finally:
        await conn.close()

asyncio.run(basic())
```

--------------------------------

### Cluster.start

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Starts the PostgreSQL server. Waits for the server to become ready by connecting shortly after startup.

```APIDOC
## Cluster.start

### Description
Starts the server. `wait` is the connection-retry budget in seconds. `server_settings` become `-c key=value` arguments. `opts` become `--key=value` arguments (e.g., `port='dynamic'` picks a free port). The server is verified by connecting shortly after startup; returns when status is 'running'. Raises ClusterError for non-zero pg_ctl start or if the server never becomes ready.

### Method
Method

### Parameters
- **wait** (int) - Optional - Connection-retry budget in seconds. Default is 60.
- **server_settings** (dict) - Optional - Server settings passed as `-c key=value` arguments.
- **opts** (keyword arguments) - Optional - Additional options passed as `--key=value` arguments.

### Returns
None

### Errors
- **ClusterError** - If pg_ctl start fails or the server does not become ready.
```

--------------------------------

### Install asyncpg with pip

Source: https://github.com/magicstack/asyncpg/blob/master/README.rst

Install asyncpg from PyPI using pip. This command is used when GSSAPI/SSPI authentication is not needed.

```bash
$ pip install asyncpg
```

--------------------------------

### Range usage example

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/types.md

Usage example showing construction, property access, subset tests, and round-trip with a PostgreSQL int4range query result.

```python
from asyncpg import Range

r = Range(1, 10, lower_inc=True, upper_inc=False)   # [1, 10)
assert r.lower == 1 and r.lower_inc and r.upper_inc is False
assert Range().isempty
assert Range(1, 5).issubset(Range(0, 10))

val = await conn.fetchval("SELECT int4range(1, 10)")
assert val == Range(1, 10)
```

--------------------------------

### Basic asyncpg connection and query example

Source: https://github.com/magicstack/asyncpg/blob/master/docs/usage.rst

Establishes a connection to a PostgreSQL database, creates a table, inserts a record, fetches a row, and closes the connection. Uses native PostgreSQL $n argument syntax.

```python
import asyncio
import asyncpg
import datetime

async def main():
    # Establish a connection to an existing database named "test"
    # as a "postgres" user.
    conn = await asyncpg.connect('postgresql://postgres@localhost/test')
    # Execute a statement to create a new table.
    await conn.execute('''
        CREATE TABLE users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
    ''')

    # Insert a record into the created table.
    await conn.execute('''
        INSERT INTO users(name, dob) VALUES($1, $2)
    ''', 'Bob', datetime.date(1984, 3, 1))

    # Select a row from the table.
    row = await conn.fetchrow(
        'SELECT * FROM users WHERE name = $1', 'Bob')
    # *row* now contains
    # asyncpg.Record(id=1, name='Bob', dob=datetime.date(1984, 3, 1))

    # Close the connection.
    await conn.close()

asyncio.run(main())
```

--------------------------------

### async def start()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/transaction.md

Starts a transaction or savepoint block. This method is called by `Connection.transaction()` and can be used explicitly.

```APIDOC
## [POST] /transaction/start

### Description
Enters the transaction or savepoint block. For a top-level block issues `BEGIN` (plus `ISOLATION LEVEL <LEVEL>`, `READ ONLY`, `DEFERRABLE` clauses as configured); for nested blocks issues `SAVEPOINT __asyncpg_savepoint_<id>__;` after verifying that a nested transaction's isolation level matches the outer one (`SHOW transaction_isolation;` is consulted when the outer level is `None`).

### Method
POST

### Endpoint
/transaction/start

### Parameters
#### Request Body
- **isolation** (string) - Optional - Isolation level: one of `read_committed`, `read_uncommitted`, `serializable`, `repeatable_read`.
- **readonly** (boolean) - Optional - Whether the transaction is read-only.
- **deferrable** (boolean) - Optional - Whether the transaction is deferrable.

### Request Example
```json
{
  "isolation": "serializable",
  "readonly": true,
  "deferrable": false
}
```

### Response
#### Success Response (200)
- **state** (string) - New state: `STARTED`.

#### Response Example
```json
{
  "state": "STARTED"
}
```

### Error Handling
- `InterfaceError('cannot start; the transaction is already started')` — called twice.
- `InterfaceError('cannot use Connection.transaction() in a manually started transaction')` — the connection is already inside a transaction not managed by asyncpg.
- `InterfaceError('nested transaction has a different isolation level: ...')` — nested isolation mismatch.
- `InterfaceError('cannot {start|commit|rollback}; the transaction is already committed/rolled back/in error state/not yet started')` — state guard violations.
- On any SQL failure during start the state becomes `FAILED` and the error is re-raised.
```

--------------------------------

### Get server version with conn.get_server_version()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Fetches the server version as a ServerVersion named tuple. The example asserts that the major and minor version are at least 14.0.

```python
v = conn.get_server_version()
assert (v.major, v.minor) >= (14, 0)
```

--------------------------------

### asyncpg.connect

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Documentation for the asyncpg.connect function, including parameters, return value, exceptions, and usage examples.

```APIDOC
## asyncpg.connect

### Description
Establishes a new connection to a PostgreSQL database. This is the primary entry point for creating a connection, supporting a wide range of parameters for authentication, SSL, timeouts, and server settings.

### Method
async function

### Signature
`async def connect(dsn=None, *, host=None, port=None, user=None, password=None, passfile=None, service=None, servicefile=None, database=None, loop=None, timeout=60, statement_cache_size=100, max_cached_statement_lifetime=300, max_cacheable_statement_size=1024*15, command_timeout=None, ssl='prefer', direct_tls=None, connection_class=Connection, record_class=protocol.Record, server_settings=None, target_session_attrs='any', krbsrvname='postgres', gsslib=None) -> Connection`

### Parameters

| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| `dsn` | str | no | `None` | libpq connection URI: `postgres://user:password@host:port/database?option=value`. Recognized options: `host`, `port`, `user`, `database`/`dbname`, `password`, `passfile`, `sslmode`, `sslcert`, `sslkey`, `sslrootcert`, `sslcrl`, `sslpassword`, `sslnegotiation`, `ssl_min_protocol_version`, `ssl_max_protocol_version`, `target_session_attrs`, `krbsrvname`, `gsslib`, `service`. Unrecognized options become server settings. Multiple comma-separated hosts supported |
| `host` | str \| list[str] | no | env/`PGHOST`/defaults | IP/name, Unix socket dir, or sequence tried in order |
| `port` | int \| list[int] | no | env/`PGPORT`/`5432` | TCP port or Unix socket file extension |
| `user` | str | no | env/`PGUSER`/OS user | Role name used for authentication |
| `password` | str \| callable | no | env/`PGPASSWORD`/pgpass | String or callable (sync or `async def`) returning the password; called per new connection |
| `passfile` | str | no | `~/.pgpass` (`%APPDATA%\postgresql\pgpass.conf` on Windows) | Password file path |
| `service` | str | no | `PGSERVICE` env | Service name in the connection service file |
| `servicefile` | str | no | `PGSERVICEFILE` env / `~/.pg_service.conf` | Connection service file location |
| `database` | str | no | env/`PGDATABASE`/`user` | Database name |
| `loop` | asyncio.AbstractEventLoop | no | current loop | Event loop instance |
| `timeout` | float | no | `60` | Connection timeout in seconds |
| `statement_cache_size` | int | no | `100` | Prepared-statement LRU cache size; `0` disables the cache |
| `max_cached_statement_lifetime` | int | no | `300` | Seconds a statement stays cached; `0` = cache indefinitely |
| `max_cacheable_statement_size` | int | no | `1024 * 15` | Max size of a cacheable statement; `0` = all sizes cacheable |
| `command_timeout` | float | no | `None` | Default per-command timeout (no timeout when `None`); must be `> 0` |
| `ssl` | bool \| str \| ssl.SSLContext | no | `'prefer'` (TCP) | `True`, an `SSLContext`, or one of `'disable'`, `'allow'`, `'prefer'`, `'require'`, `'verify-ca'`, `'verify-full'` |
| `direct_tls` | bool | no | `None` | `True` = direct TLS instead of STARTTLS; requires `ssl` mode ≥ `'require'` |
| `connection_class` | type | no | `Connection` | Must be a subclass of `asyncpg.Connection` |
| `record_class` | type | no | `protocol.Record` | Must be `Record` or a subclass that does not redefine `__new__`/`__init__` |
| `server_settings` | dict[str, str] | no | `None` | Server runtime parameters sent at startup |
| `target_session_attrs` | str \| `SessionAttribute` | no | `PGTARGETSESSIONATTRS` env / `'any'` | `'any'`, `'primary'`, `'standby'`, `'read-write'`, `'read-only'`, `'prefer-standby'` |
| `krbsrvname` | str | no | `PGKRBSRVNAME` env / `'postgres'` | Kerberos service name for GSSAPI |
| `gsslib` | str | no | `PGGSSLIB` env / `'sspi'` on Windows else `'gssapi'` | GSS library: `'gssapi'` or `'sspi'` |

### Returns
A new `Connection` instance.

### Raises
- `asyncpg.exceptions.InterfaceError` - bad `connection_class`/`record_class`
- `ClientConfigurationError` - bad DSN scheme, unknown `sslmode`/`sslnegotiation`/`target_session_attrs`/`gsslib`, direct TLS without a strong SSL mode, missing root certificate, mismatched host/port list lengths, undeterminable user/database
- `ValueError` - negative cache sizes, non-positive `command_timeout`
- `PostgresConnectionError` subclasses - connection failures
- `TargetServerAttributeNotMatched` - host-attribute screening failures

### Example
```python
import asyncio, asyncpg

async def basic():
    conn = await asyncpg.connect(
        user='user', password='password',
        database='database', host='127.0.0.1', port=5432,
        command_timeout=30, ssl='verify-full',
        server_settings={'application_name': 'myapp'},
    )
    try:
        rows = await conn.fetch('SELECT * FROM mytable WHERE id = $1', 10)
        print(rows)
    finally:
        await conn.close()

asyncio.run(basic())
```

### Example with Error Handling
```python
import asyncpg

async def error_handling():
    try:
        conn = await asyncpg.connect(
            'postgres://user:pass@[fe80::1ff:fe23:4567:890a%25eth0]/dbname',
            statement_cache_size=0,      # e.g. when using pgbouncer transaction mode
            target_session_attrs='read-write',
        )
    except asyncpg.PostgresConnectionError as e:
        print('could not connect:', e.sqlstate, e.message)
        return
    except asyncpg.exceptions.ClientConfigurationError as e:
        print('bad configuration:', e)
        return
    finally:
        # conn may be unbound here; check with locals()
        pass

asyncio.run(error_handling())
```
```

--------------------------------

### asyncpg.connect() error handling example

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Shows how to catch asyncpg.PostgresConnectionError and asyncpg.exceptions.ClientConfigurationError when connecting. The example uses a DSN with an IPv6 address, disables the statement cache for pgbouncer transaction mode, and sets target_session_attrs to 'read-write'.

```python
import asyncpg

async def error_handling():
    try:
        conn = await asyncpg.connect(
            'postgres://user:pass@[fe80::1ff:fe23:4567:890a%25eth0]/dbname',
            statement_cache_size=0,      # e.g. when using pgbouncer transaction mode
            target_session_attrs='read-write',
        )
    except asyncpg.PostgresConnectionError as e:
        print('could not connect:', e.sqlstate, e.message)
        return
    except asyncpg.exceptions.ClientConfigurationError as e:
        print('bad configuration:', e)
        return
    finally:
        # conn may be unbound here; check with locals()
        pass

asyncio.run(error_handling())
```

--------------------------------

### Run asyncpg tests

Source: https://github.com/magicstack/asyncpg/blob/master/docs/installation.rst

Execute the asyncpg test suite. Requires PostgreSQL to be installed.

```bash
$ python setup.py test
```

--------------------------------

### Get reset query with conn.get_reset_query()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Returns the SQL sent on connection release, composed from server capabilities. The example prints the cached query, showing the advisory unlock, CLOSE ALL, UNLISTEN *, and RESET ALL statements.

```python
print(conn.get_reset_query())
# SELECT pg_advisory_unlock_all();
# CLOSE ALL;
# UNLISTEN *;
# RESET ALL;
```

--------------------------------

### Basic asyncpg usage example

Source: https://github.com/magicstack/asyncpg/blob/master/README.rst

A minimal Python script demonstrating how to connect to a PostgreSQL database, execute a query with a parameter, and close the connection using asyncio. The query uses the $1 placeholder for the parameter.

```python
import asyncio
import asyncpg

async def run():
    conn = await asyncpg.connect(user='user', password='password',
                                 database='database', host='127.0.0.1')
    values = await conn.fetch(
        'SELECT * FROM mytable WHERE id = $1',
        10,
    )
    await conn.close()

asyncio.run(run())
```

--------------------------------

### Manual transaction start/commit/rollback

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/transaction.md

Shows explicit start(), commit(), and rollback() usage. On exception, rollback is called and the error is re-raised.

```python
tr = conn.transaction()
await tr.start()
try:
    await conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id=1')
    await tr.commit()
except Exception:
    await tr.rollback()
    raise
```

--------------------------------

### Fetch all rows with conn.fetch()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Runs a query and returns all rows as a list of Record objects. The example fetches users older than 18, passing the age as a parameter.

```python
rows = await conn.fetch('SELECT id, name FROM users WHERE age > $1', 18)
```

--------------------------------

### create_pool() with custom init and reset hooks

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/pool.md

Example of using custom init and reset hooks with create_pool(). The init hook sets a built-in type codec for hstore, and the reset hook executes a custom SQL command. The pool is created with dsn, min_size=1, max_size=5, and then acquire/release/close are used.

```python
import asyncpg

async def with_custom_hooks():
    async def init(con):
        await con.set_builtin_type_codec(
            'hstore', codec_name='pg_contrib.hstore')

    async def reset(con):
        await con.execute('RESET myapp.session_var')

    pool = await asyncpg.create_pool(
        dsn='postgres:///db',
        min_size=1, max_size=5,
        init=init, reset=reset,
    )
    try:
        con = await pool.acquire(timeout=10)
        try:
            await con.fetchval('SELECT 1')
        finally:
            await pool.release(con)
    finally:
        await pool.close()

asyncio.run(with_custom_hooks())
```

--------------------------------

### Using asyncpg Record access

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/types.md

Example of using a Record instance: access by name or position, unpacking via iteration, and using 'in' and len(). Demonstrates the dual mapping/sequence behavior of asyncpg Record rows.

```python
row = await conn.fetchrow('SELECT * FROM users WHERE id = $1', 1)
assert row['name'] == row[1]            # mapping and sequence access
name, email = row                       # unpacking via __iter__
assert 'name' in row and len(row) == len(row.keys())
```

--------------------------------

### HotStandbyCluster.__init__ signature

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Initializes a HotStandbyCluster with a master cluster dict and replication user. The cluster is created with pg_basebackup; init() writes standby.signal (PG ≥ 12) or recovery.conf (older), and start() sets primary_conninfo.

```python
def __init__(self, *, master, replication_user, data_dir_suffix=None,
             data_dir_prefix=None, data_dir_parent=None, pg_config_path=None):
```

--------------------------------

### Fetch first row with conn.fetchrow()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Returns the first row as a Record, or None if no rows. The example fetches a user by id and prints the name if the row exists.

```python
row = await conn.fetchrow('SELECT * FROM users WHERE id = $1', 7)
if row is not None:
    print(row['name'])
```

--------------------------------

### Accessing connection settings with ConnectionSettings

Source: https://github.com/magicstack/asyncpg/blob/master/docs/api/index.rst

Shows how to retrieve a setting value from a connection's settings collection. The example accesses the client_encoding setting.

```pycon
>>> connection.get_settings().client_encoding
'UTF8'
```

--------------------------------

### Execute SQL with conn.execute()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Executes SQL commands and returns the status string. The example shows a multi-command execution and a parameterized insert, asserting the returned statuses.

```python
status = await conn.execute('''
    CREATE TABLE mytab (a int);
    INSERT INTO mytab (a) VALUES (100), (200), (300);
''')
assert status == 'INSERT 0 3'

result = await conn.execute('INSERT INTO mytab (a) VALUES ($1)', 1)
assert result == 'INSERT 0 1'
```

--------------------------------

### asyncpg connection pool with aiohttp web service

Source: https://github.com/magicstack/asyncpg/blob/master/docs/usage.rst

Complete example of a simple web service that computes the requested power of two. Uses asyncpg.create_pool() to create a pool, acquires connections from the pool, and opens a transaction for each request. Requires aiohttp and asyncpg.

```python
import asyncio
import asyncpg
from aiohttp import web


async def handle(request):
    """Handle incoming requests."""
    pool = request.app['pool']
    power = int(request.match_info.get('power', 10))

    # Take a connection from the pool.
    async with pool.acquire() as connection:
        # Open a transaction.
        async with connection.transaction():
            # Run the query passing the request argument.
            result = await connection.fetchval('select 2 ^ $1', power)
            return web.Response(
                text="2 ^ {} is {}".format(power, result))


async def init_db(app):
    """Initialize a connection pool."""
     app['pool'] = await asyncpg.create_pool(database='postgres',
                                             user='postgres')
     yield
     await app['pool'].close()


def init_app():
    """Initialize the application server."""
    app = web.Application()
    # Create a database context
    app.cleanup_ctx.append(init_db)
    # Configure service routes
    app.router.add_route('GET', '/{power:\d+}', handle)
    app.router.add_route('GET', '/', handle)
    return app


app = init_app()
web.run_app(app)
```

--------------------------------

### transaction(*, isolation=None, readonly=False, deferrable=False)

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Creates a `Transaction` object; does not start it.

```APIDOC
## `transaction(*, isolation=None, readonly=False, deferrable=False)`

### Description
Creates a `Transaction` object; does not start it. `isolation` may be `'serializable'`, `'repeatable_read'`, `'read_uncommitted'`, `'read_committed'` (or `None`, deferring to the server). Raises `InterfaceError` when the connection is closed.

### Method
`transaction`

### Parameters
#### Keyword Arguments
- **isolation** (str or None) - Optional - Transaction isolation level.
- **readonly** (bool) - Optional - Whether the transaction is read-only.
- **deferrable** (bool) - Optional - Whether the transaction is deferrable.

### Returns
- **Transaction** - A `Transaction` object.

### Example
```python
async with conn.transaction(isolation='serializable', readonly=True):
    await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
```
```

--------------------------------

### explain() with and without analyze

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/prepared-statement.md

Gets the execution plan as deserialized JSON from EXPLAIN (FORMAT JSON, VERBOSE). With analyze=True the statement is executed inside a transaction that is rolled back to discard side effects.

```python
stmt = await conn.prepare('SELECT * FROM big WHERE x = $1')
plan = await stmt.explain(42)
print(plan[0]['Plan']['Total Cost'])
plan = await stmt.explain(42, analyze=True)
```

--------------------------------

### CursorFactory.__await__

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cursor.md

Manual paging mode of the CursorFactory: await the factory to get an open Cursor for fetch operations.

```APIDOC
## [AWAIT] CursorFactory.__await__

### Description
Binds the statement and returns an open Cursor for manual paging with fetch(n), fetchrow(), and forward(n). Raises InterfaceError if prefetch was set on the factory.

### Method
__await__

### Parameters
- **prefetch** (None) - Required - Must be None; otherwise raises InterfaceError.

### Request Example
```python
cur = await conn.cursor('SELECT * FROM big_table WHERE id > $1', 1000)
while True:
    rows = await cur.fetch(500)
    if not rows:
        break
    for r in rows:
        print(r)
```

### Response
#### Success Response
- **Cursor** - An open cursor for manual paging.

#### Response Example
```python
cur = await conn.cursor('SELECT * FROM big_table WHERE id > $1', 1000)
rows = await cur.fetch(500)
```
```

--------------------------------

### Cursor from prepared statement

Source: https://github.com/magicstack/asyncpg/blob/master/docs/api/index.rst

Create a prepared statement with `con.prepare()`, then use `stmt.cursor()` inside a transaction to iterate over results. The example passes an argument to the query.

```python
async def iterate(con: Connection):
    # Create a prepared statement that will accept one argument
    stmt = await con.prepare('SELECT generate_series(0, $1)')

    async with con.transaction():
        # Postgres requires non-scrollable cursors to be created
        # and used in a transaction.

        # Execute the prepared statement passing `10` as the
        # argument -- that will generate a series or records
        # from 0..10.  Iterate over all of them and print every
        # record.
        async for record in stmt.cursor(10):
            print(record)
```

--------------------------------

### Fetch multiple rows with conn.fetchmany()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Runs the same query once per argument sequence in args and concatenates all result rows. The example inserts multiple rows and returns the RETURNING values, with the expected output shown in a comment.

```python
rows = await conn.fetchmany(
    'INSERT INTO mytab (a, b) VALUES ($1, $2) RETURNING a;',
    [('x', 1), ('y', 2), ('z', 3)])
# [<Record row=('x',)>, <Record row=('y',)>, <Record row=('z',)>]
```

--------------------------------

### set_builtin_type_codec: Use built-in hstore codec

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection-io.md

Declares that the 'hstore' type uses the built-in 'pg_contrib.hstore' codec. The example then fetches a value and prints the result of the 'a' key, which is '1'.

```python
# Treat "hstore" as hstore everywhere
await conn.set_builtin_type_codec('hstore', schema='public',
                                  codec_name='pg_contrib.hstore')
row = await conn.fetchval("SELECT 'a=>1'::hstore")
print(row.get('a'))   # '1'
```

--------------------------------

### Reload schema state with conn.reload_schema_state()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Tells asyncpg the database schema changed, dropping type codec and prepared-statement caches. Prevents OutdatedSchemaCacheError. The example alters a type and then reloads the schema state.

```python
await conn.execute('ALTER TYPE custom DROP ATTRIBUTE y')
await conn.execute('ALTER TYPE custom ADD ATTRIBUTE y text')
await conn.reload_schema_state()
```

--------------------------------

### copy_to_table: Import from file and async generator

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection-io.md

Copies data into a table from a source using COPY <table> FROM STDIN. The first example imports from a file path; the second uses an async generator that yields encoded rows. Both return the COPY status string, e.g. 'COPY 140000' or 'COPY 100'.

```python
result = await conn.copy_to_table(
    'mytable', source='datafile.tbl')   # 'COPY 140000'
```

```python
async def gen():
    for i in range(100):
        yield f"{i}\tfoo\n".encode()
result = await conn.copy_to_table('mytable', source=gen())
```

--------------------------------

### Fetch rows in pages with cursor.fetch

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cursor.md

Retrieves the next n rows as a list of Record. Marks the cursor exhausted when fewer than n rows remain; returns [] when exhausted. The example fetches 1000 rows at a time with a 10-second timeout and processes each page.

```python
cur = await conn.cursor('SELECT * FROM logs ORDER BY ts')
page = await cur.fetch(1000, timeout=10)
while page:
    handle(page)
    page = await cur.fetch(1000)
```

--------------------------------

### Python: Using query_logger with a custom callback class

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Registers a QuerySaver instance as a query logger for the duration of the with block. The callback's __call__ method appends each query string to a list, which is printed after the block. The example demonstrates synchronous callback usage with await conn.execute('SELECT 1').

```python
class QuerySaver:
    def __init__(self):
        self.queries = []
    def __call__(self, record):
        self.queries.append(record.query)

saver = QuerySaver()
with conn.query_logger(saver):
    await conn.execute('SELECT 1')
print(saver.queries)   # ['SELECT 1']
```

--------------------------------

### copy_records_to_table: Insert records from list and async generator

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection-io.md

Copies an iterable of row tuples into a table using binary COPY. The first example inserts a list of tuples; the second uses an async generator that yields tuples. Both return the COPY status string, e.g. 'COPY 2' or 'COPY 100'.

```python
result = await conn.copy_records_to_table(
    'mytable', records=[(1, 'foo', 'bar'), (2, 'ham', 'spam')])
# 'COPY 2'
```

```python
async def record_gen(size):
    for i in range(size):
        yield (i,)
result = await conn.copy_records_to_table(
    'mytable', records=record_gen(100))   # 'COPY 100'
```

--------------------------------

### Cluster.init

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Initializes the data directory using pg_ctl init. Unknown keyword arguments are passed as --key=value options.

```APIDOC
## Cluster.init

### Description
Initializes the data directory via `pg_ctl init`. Unknown keyword arguments are passed as `--key=value` to the init process after `-o`. The encoding is defaulted to UTF-8. Raises ClusterError if already initialized or if pg_ctl init exits non-zero.

### Method
Method

### Parameters
- **settings** (keyword arguments) - Optional - Additional settings passed as `--key=value` to the init process.

### Returns
- **str** - The captured output from pg_ctl init.

### Errors
- **ClusterError** - If already initialized or if pg_ctl init fails.
```

--------------------------------

### Get server PID with conn.get_server_pid()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Retrieves the PostgreSQL server's process ID. The returned integer can be used for diagnostics or monitoring. Example shows assignment to a variable.

```python
pid = conn.get_server_pid()   # e.g. 24601
```

--------------------------------

### Prepare and execute a statement with asyncpg

Source: https://github.com/magicstack/asyncpg/blob/master/docs/api/index.rst

Demonstrates preparing a statement with Connection.prepare() and executing it multiple times with different arguments using fetchval(). The output shows the results for arguments 10 and 20.

```pycon
>>> import asyncpg, asyncio
>>> async def run():
...     conn = await asyncpg.connect()
...     stmt = await conn.prepare('''SELECT 2 ^ $1''')
...     print(await stmt.fetchval(10))
...     print(await stmt.fetchval(20))
...
>>> asyncio.run(run())
1024.0
1048576.0
```

--------------------------------

### asyncpg.connect()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/configuration.md

Configuration options for establishing a connection to PostgreSQL using asyncpg.connect().

```APIDOC
## asyncpg.connect()

### Description
Establishes a connection to a PostgreSQL database. All configuration is passed as keyword arguments, DSN URI options, or environment variables.

### Method
connect

### Endpoint
asyncpg.connect()

### Parameters
#### Keyword Arguments
- **dsn** (str) - Optional - libpq URI `postgres://user:password@host:port/database?option=value`; keyword args override URI values
- **host** (str | list[str]) - Optional - IP/hostname, absolute Unix socket directory, or sequence tried in order
- **port** (int | list[int]) - Optional - TCP port; with a host list may also be a list (same length)
- **user** (str) - Optional - role name
- **password** (str | callable) - Optional - password, or callable (sync/async) evaluated per connection
- **passfile** (str) - Optional - password file; permissions `u=rw` enforced on Unix (group/world access warns and is ignored)
- **service** (str) - Optional - service section name in the connection service file
- **servicefile** (str) - Optional - connection service file
- **database** (str) - Optional - database name
- **loop** (AbstractEventLoop) - Optional - event loop
- **timeout** (float) - Optional - connect timeout
- **statement_cache_size** (int) - Optional - LRU prepared-statement cache size; `0` disables (marks statements unprepared each use — recommended with pgbouncer `transaction`/`statement` pooling); `None`/negative/bool raise `ValueError`
- **max_cached_statement_lifetime** (int) - Optional - seconds a statement stays cached; `0` = indefinitely
- **max_cacheable_statement_size** (int) - Optional - max cached statement bytes; `0` = all
- **command_timeout** (float) - Optional - default per-command timeout; must be `> 0` float (`ValueError` otherwise)
- **ssl** (bool | str | SSLContext) - Optional - `'disable'`, `'allow'`, `'prefer'`, `'require'`, `'verify-ca'`, `'verify-full'`, `True`, or an `SSLContext` (see connect-utils.md)
- **direct_tls** (bool) - Optional - skip STARTTLS; requires `ssl` ≥ `'require'` (`ClientConfigurationError` otherwise); PG17+ needs `postgresql` ALPN on explicit contexts
- **connection_class** (type) - Optional - subclass of `asyncpg.Connection` (`InterfaceError` otherwise)
- **record_class** (type) - Optional - `Record` subclass without redefined `__new__`/`__init__` (`InterfaceError` otherwise)
- **server_settings** (dict[str, str]) - Optional - startup session parameters; must be `Dict[str, str]` (`ClientConfigurationError` otherwise)
- **target_session_attrs** (str | SessionAttribute) - Optional - host screening, see connect-utils.md
- **krbsrvname** (str) - Optional - Kerberos service name (GSSAPI)
- **gsslib** (str) - Optional - `'gssapi'` or `'sspi'` only

### Response
Returns a connection object (type determined by `connection_class`).
```

--------------------------------

### RunningCluster.__init__ signature

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Initializes a RunningCluster with connection keyword arguments. is_managed() returns False, get_status() always returns 'running', and init/start/stop/destroy are no-ops. reset_hba() and add_hba_entry() raise ClusterError.

```python
def __init__(self, **kwargs):
```

--------------------------------

### asyncpg.connect(dsn=None, *, ...)

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Documentation for the asyncpg.connect() coroutine function, which establishes a connection to a PostgreSQL server.

```APIDOC
## asyncpg.connect(dsn=None, *, ...)

### Description
Establishes a connection to a PostgreSQL server. Connection parameters may be given as a libpq-style URI in `dsn`, as keyword arguments, or both (keyword arguments override values parsed from the URI). Defaults for most arguments can be provided via PostgreSQL environment variables.

### Method
Coroutine function

### Signature
```python
async def connect(dsn=None, *,
                  host=None, port=None,
                  user=None, password=None, passfile=None,
                  service=None,
                  servicefile=None,
                  database=None,
                  loop=None,
                  timeout=60,
                  statement_cache_size=100,
                  max_cached_statement_lifetime=300,
                  max_cacheable_statement_size=1024 * 15,
                  command_timeout=None,
                  ssl=None,
                  direct_tls=None,
                  connection_class=Connection,
                  record_class=protocol.Record,
                  server_settings=None,
                  target_session_attrs=None,
                  krbsrvname=None,
                  gsslib=None):
```

### Parameters
- **dsn** (str) - Optional - libpq-style connection URI.
- **host** (str) - Optional - Database host address.
- **port** (int) - Optional - Database port.
- **user** (str) - Optional - Database user name.
- **password** (str) - Optional - Database password.
- **passfile** (str) - Optional - Path to password file.
- **service** (str) - Optional - Service name.
- **servicefile** (str) - Optional - Path to service file.
- **database** (str) - Optional - Database name.
- **loop** (object) - Optional - Event loop.
- **timeout** (float) - Optional - Connection timeout in seconds (default 60).
- **statement_cache_size** (int) - Optional - Size of statement cache (default 100).
- **max_cached_statement_lifetime** (int) - Optional - Maximum lifetime of cached statements in seconds (default 300).
- **max_cacheable_statement_size** (int) - Optional - Maximum size of cacheable statements in bytes (default 1024 * 15).
- **command_timeout** (float) - Optional - Default command timeout.
- **ssl** (bool/object) - Optional - SSL configuration.
- **direct_tls** (bool) - Optional - Use direct TLS.
- **connection_class** (type) - Optional - Connection class to use (default `Connection`).
- **record_class** (type) - Optional - Record class to use (default `protocol.Record`).
- **server_settings** (dict) - Optional - Server settings.
- **target_session_attrs** (str) - Optional - Target session attributes.
- **krbsrvname** (str) - Optional - Kerberos service name.
- **gsslib** (str) - Optional - GSS library.

### Returns
A `Connection` object (or an instance of `connection_class`).
```

--------------------------------

### TransactionState enum definition

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/transaction.md

Defines the state machine for a Transaction: NEW, STARTED, COMMITTED, ROLLEDBACK, FAILED. Used internally by state guards.

```python
class TransactionState(enum.Enum):
    NEW = 0
    STARTED = 1
    COMMITTED = 2
    ROLLEDBACK = 3
    FAILED = 4
```

--------------------------------

### Check transaction state with conn.is_in_transaction()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Checks whether the connection is inside a transaction. The example conditionally executes a BEGIN command when not already in a transaction.

```python
if not conn.is_in_transaction():
    await conn.execute('BEGIN')
```

--------------------------------

### Execute multiple statements with conn.executemany()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Executes a command once per argument sequence. The example inserts three rows into a table using parameterized values.

```python
await conn.executemany(
    'INSERT INTO mytab (a) VALUES ($1, $2, $3);',
    [(1, 2, 3), (4, 5, 6)],
)
```

--------------------------------

### Using Record objects with asyncpg

Source: https://github.com/magicstack/asyncpg/blob/master/docs/api/index.rst

Shows how to fetch a row and access fields by index or name, convert to dict/tuple, and demonstrates the repr. Note that Record objects cannot be created from Python code.

```pycon
>>> import asyncpg
>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>> conn = loop.run_until_complete(asyncpg.connect())
>>> r = loop.run_until_complete(conn.fetchrow('''
...     SELECT oid, rolname, rolsuper FROM pg_roles WHERE rolname = user'''))
>>> r
<Record oid=16388 rolname='elvis' rolsuper=True>
>>> r['oid']
16388
>>> r[0]
16388
>>> dict(r)
{'oid': 16388, 'rolname': 'elvis', 'rolsuper': True}
>>> tuple(r)
(16388, 'elvis', True)
```

--------------------------------

### PreparedStatement.explain()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/prepared-statement.md

Returns the execution plan of the statement as deserialized JSON output of EXPLAIN (FORMAT JSON, VERBOSE[, ANALYZE]).

```APIDOC
## explain()

### Description
Returns the execution plan of the statement as deserialized JSON output of EXPLAIN (FORMAT JSON, VERBOSE[, ANALYZE]). When analyze=True, the statement is executed in a rolled-back transaction.

### Method
explain

### Signature
`async def explain(self, *args, analyze=False)`

### Parameters
- **args** (any) - Required - Query arguments.
- **analyze** (bool) - Optional - If True, execute the statement to get actual plan.

### Returns
- **object** - Deserialized JSON plan.

### Example
```python
stmt = await conn.prepare('SELECT * FROM big WHERE x = $1')
plan = await stmt.explain(42)
print(plan[0]['Plan']['Total Cost'])
plan = await stmt.explain(42, analyze=True)
```
```

--------------------------------

### Fetch a single value with conn.fetchval()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Returns the value of a specific column (default 0) from the first row, or None if no rows. The example shows fetching a count and asserting a parameterized value.

```python
count = await conn.fetchval('SELECT count(*) FROM users')
assert await conn.fetchval('SELECT $1::int', 42) == 42
```

--------------------------------

### Iterate rows with cursor.fetchrow

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cursor.md

Returns the next row as a Record, or None when exhausted. The example uses the walrus operator to loop until fetchrow returns None, printing the 'ts' column of each row.

```python
cur = await conn.cursor('SELECT * FROM logs')
while row := await cur.fetchrow():
    print(row['ts'])
```

--------------------------------

### Cluster.__init__

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/cluster.md

Initializes a Cluster instance wrapping a PostgreSQL data directory. Optionally overrides the pg_config path.

```APIDOC
## Cluster.__init__

### Description
Initializes a Cluster instance that wraps a PostgreSQL data directory. The pg_config_path parameter overrides the automatic discovery of the pg_config binary.

### Method
Constructor

### Parameters
- **data_dir** (str) - Required - Path to the PostgreSQL data directory.
- **pg_config_path** (str) - Optional - Path to the pg_config binary. Overrides automatic discovery.

### Returns
None
```

--------------------------------

### Using pool.fetch and pool.execute

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/pool.md

Shows how to use the pool's delegating query methods. The pool acquires a connection for the duration of each call, and concurrent calls serialize on different pooled connections.

```python
rows = await pool.fetch('SELECT * FROM users WHERE id = $1', 42)
status = await pool.execute('CREATE TABLE t (a int)')
```

--------------------------------

### asyncpg.create_pool()

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/configuration.md

Configuration options for creating a connection pool. Each new pooled connection uses the same configuration as asyncpg.connect().

```APIDOC
## asyncpg.create_pool()

### Description
Creates a connection pool. Each new pooled connection accepts the same configuration options as `asyncpg.connect()`.

### Method
create_pool

### Endpoint
asyncpg.create_pool()

### Parameters
#### Keyword Arguments
Same as `asyncpg.connect()` options, applied to each new pooled connection.

### Response
Returns a connection pool object.
```

--------------------------------

### Programmatic SSL context for verify-full

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connect-utils.md

Equivalent configuration for the 'verify-full' SSL mode using an ssl.SSLContext. Requires a CA bundle for certificate verification, enables hostname checking, and loads a client certificate chain. Raises ClientConfigurationError if the sslmode value does not parse or the root certificate is missing.

```python
import ssl, asyncpg

ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                 cafile='path/to/ca_bundle.pem')
ctx.check_hostname = True                      # verify-full
ctx.load_cert_chain('path/to/client.cert', keyfile='path/to/client.key')
conn = await asyncpg.connect(user='postgres', ssl=ctx)
```

--------------------------------

### asyncpg.connect() function signature

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/connection.md

Defines the connect() coroutine that establishes a PostgreSQL connection. Parameters can be passed as a libpq-style URI, keyword arguments, or both, with keyword arguments overriding URI values. Defaults for most arguments can be provided via PostgreSQL environment variables.

```python
async def connect(dsn=None, *,
                  host=None, port=None,
                  user=None, password=None, passfile=None,
                  service=None,
                  servicefile=None,
                  database=None,
                  loop=None,
                  timeout=60,
                  statement_cache_size=100,
                  max_cached_statement_lifetime=300,
                  max_cacheable_statement_size=1024 * 15,
                  command_timeout=None,
                  ssl=None,
                  direct_tls=None,
                  connection_class=Connection,
                  record_class=protocol.Record,
                  server_settings=None,
                  target_session_attrs=None,
                  krbsrvname=None,
                  gsslib=None):
```

--------------------------------

### fetch() returns all rows

Source: https://github.com/magicstack/asyncpg/blob/master/_autodocs/api-reference/prepared-statement.md

Executes the prepared statement with the given arguments and returns all rows as a list of Record.

```python
records = await stmt.fetch(1, 'a')
```
Tokens:
Raw