asyncpg

Warnings

• breaking Placeholders are $1/$2/$3 (PostgreSQL native) NOT %s (psycopg2) or ? (sqlite3). LLMs trained on psycopg2 code consistently generate %s placeholders which fail with asyncpg.
  Fix: await conn.fetch('SELECT * FROM t WHERE id = $1 AND active = $2', id, True)
• breaking asyncpg is NOT DB-API 2.0 compliant. Code written for psycopg2/sqlite3 will not work without changes. No cursor objects, different method names (fetch/fetchrow/fetchval not execute/fetchone/fetchall).
  Fix: Use fetch() for multiple rows, fetchrow() for one row, fetchval() for a single value, execute() for DML without results.
• breaking Prepared statements break with pgbouncer in transaction/statement pool mode. Error: 'prepared statement asyncpg_stmt_X does not exist'. Affects Supabase transaction pooler (port 6543), Neon, and any pgbouncer setup.
  Fix: Set statement_cache_size=0 when connecting: await asyncpg.connect('...', statement_cache_size=0). For SQLAlchemy: connect_args={'statement_cache_size': 0}
• gotcha fetch() returns list of asyncpg.Record objects, not dicts. Record supports dict-style access (row['name']) but isinstance(row, dict) is False. Code that expects dicts breaks silently.
  Fix: Convert with dict(row) or [dict(r) for r in rows] if you need plain dicts.
• gotcha Still pre-1.0 (0.31.x). API stability not guaranteed across minor versions.
  Fix: Pin version in production: pip install asyncpg==0.31.0
• gotcha Prepared statements and cursors from Connection.prepare() become invalid once a connection is released back to the pool. Must re-prepare on next acquisition.
  Fix: Don't cache prepared statement objects across pool.acquire() calls.

Install

• pip install asyncpg Python

Imports

• connect

  import asyncpg
  import asyncio
  
  async def main():
      conn = await asyncpg.connect(
          'postgresql://user:pass@localhost/mydb'
      )
      # $1, $2 placeholders — NOT %s
      row = await conn.fetchrow(
          'SELECT id, name FROM users WHERE id = $1',
          42
      )
      print(row['name'])  # Record supports dict-style access
      await conn.close()
  
  asyncio.run(main())

  asyncpg uses PostgreSQL-native $1/$2/$3 positional placeholders. NOT %s (psycopg2 style) or ? (sqlite3 style). Using %s raises a syntax error or produces unexpected behavior.
• create_pool

  import asyncpg
  import asyncio
  
  async def main():
      pool = await asyncpg.create_pool(
          'postgresql://user:pass@localhost/mydb',
          min_size=2,
          max_size=10
      )
      async with pool.acquire() as conn:
          rows = await conn.fetch('SELECT * FROM users')
          for row in rows:
              print(dict(row))  # convert Record to dict
      await pool.close()
  
  asyncio.run(main())

  Always use create_pool() for web apps. Connecting per request adds 50-200ms latency. Pool manages connection lifecycle automatically.

Quickstart

asyncpg connection pool with correct $1/$2 placeholders.

# pip install asyncpg
import asyncpg
import asyncio

async def main():
    # Connection pool for production
    pool = await asyncpg.create_pool(
        'postgresql://user:pass@localhost/mydb',
        min_size=2,
        max_size=10
    )

    async with pool.acquire() as conn:
        # $1, $2 — not %s
        await conn.execute(
            'INSERT INTO users(name, email) VALUES($1, $2)',
            'Alice', 'alice@example.com'
        )

        # fetchrow returns asyncpg.Record — dict-like
        row = await conn.fetchrow(
            'SELECT * FROM users WHERE name = $1', 'Alice'
        )
        print(row['name'])   # 'Alice'
        print(dict(row))     # convert to plain dict

        # fetch returns list of Records
        rows = await conn.fetch('SELECT id, name FROM users')

    await pool.close()

asyncio.run(main())