asyncpg Documentation

The recommended way to use transactions is via an async with statement on a Connection.transaction() object.

Nested Transactions: asyncpg supports nested transactions. When you create a nested transaction context, it automatically creates a PostgreSQL savepoint. If an exception occurs within the nested block, that specific savepoint is rolled back without affecting the outer transaction.

async with connection.transaction():
    await connection.execute("INSERT INTO mytable VALUES(1, 2, 3)")

# Nested transaction example
async with connection.transaction():
    await connection.execute('CREATE TABLE mytab (a int)')
    try:
        async with connection.transaction():
            await connection.execute('INSERT INTO mytab (a) VALUES (1), (2)')
            raise Exception
    except:
        pass
    # The nested transaction was rolled back, but the outer one can continue
    assert await connection.fetch('SELECT a FROM mytab') == []

To interact with a PostgreSQL database, use asyncpg.connect() to establish a new session. This returns a Connection instance used to execute statements and manage transactions.

Note that asyncpg uses native PostgreSQL syntax for query arguments using $n (e.g., $1, $2).

import asyncio
import asyncpg
import datetime

async def main():
    # Establish a connection to an existing database
    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 using $n positional arguments
    await conn.execute('''
        INSERT INTO users(name, dob) VALUES($1, $2)
    ''', 'Bob', datetime.date(1984, 3, 1))

    # Select a single row
    row = await conn.fetchrow(
        'SELECT * FROM users WHERE name = $1', 'Bob')
    # row is an asyncpg.Record

    # Close the connection.
    await conn.close()

asyncio.run(main())

For server-side applications handling frequent requests, use asyncpg.create_pool() to create a Pool object. This provides an advanced connection pooling implementation that eliminates the need for external poolers like PgBouncer.

To use a connection from the pool, use pool.acquire() within an async with block.

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

# ... (rest of the aiohttp setup) ...

Cursors allow you to iterate over large query results without fetching all rows into memory at once.

Important Requirements:

1. Transactions: Cursors created via Connection.cursor() or PreparedStatement.cursor() must be used within a transaction block. Attempting to use them outside a transaction will raise asyncpg.exceptions.InterfaceError.
2. Direction: Cursors provided by asyncpg are non-scrollable (forward-only). To use scrollable cursors, you must execute DECLARE ... SCROLL CURSOR directly via SQL.

Usage Patterns:

• Asynchronous Iteration: Use async for with con.cursor(query) for efficient prefetching.
• Manual Navigation: Use cur.forward(n), cur.fetchrow(), or cur.fetch(n) to move through the result set manually.

async def iterate(con: Connection):
    async with con.transaction():
        # Efficient prefetching via async for
        async for record in con.cursor('SELECT generate_series(0, 100)'):
            print(record)

async def manual_iterate(con: Connection):
    async with con.transaction():
        cur = await con.cursor('SELECT generate_series(0, 100)')
        await cur.forward(10)
        print(await cur.fetchrow())
        print(await cur.fetch(5))

Use asyncpg.create_pool() to manage a set of connections. It is recommended to use it within an async with block to ensure the pool is closed properly.

Key parameters:

• dsn: Connection string.
• min_size: Minimum number of connections to keep in the pool (default: 10).
• max_size: Maximum number of connections in the pool (default: 10).
• max_queries: Maximum number of queries a connection can execute before being closed and replaced (default: 50000).
• max_inactive_connection_lifetime: Maximum time (in seconds) a connection can be idle before being closed (default: 300.0).
• setup: An async function called for each connection after it is created.
• init: An async function called for each connection after it is created but before it is used.
• reset: An async function called for each connection when it is released back to the pool.

async with asyncpg.create_pool(user='postgres',
                               command_timeout=60) as pool:
    await pool.fetch('SELECT 1')