Python PostgreSQL with asyncpg: Async Database Operations

asyncpg is the fastest PostgreSQL driver for Python - pure asyncio, no thread overhead, and up to 3× faster than psycopg2 on typical workloads. It is the go-to choice for any async Python backend.

Installation

pip install asyncpg

PostgreSQL server must already be running.

Connect and Create a Pool

import asyncio
import asyncpg
from datetime import datetime

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

async def create_pool() -> asyncpg.Pool:
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=2,
        max_size=10,
        command_timeout=30,
        server_settings={"application_name": "myapp"},
    )
    print("Pool created.")
    return pool

Schema Setup

CREATE_TABLES = """
CREATE TABLE IF NOT EXISTS users (
    id BIGSERIAL PRIMARY KEY,
    username TEXT NOT NULL UNIQUE,
    email TEXT NOT NULL UNIQUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS posts (
    id BIGSERIAL PRIMARY KEY,
    user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    title TEXT NOT NULL,
    body TEXT NOT NULL DEFAULT '',
    published BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_posts_user ON posts(user_id);
CREATE INDEX IF NOT EXISTS idx_posts_created ON posts(created_at DESC);
"""

async def setup_schema(pool: asyncpg.Pool) -> None:
    async with pool.acquire() as conn:
        await conn.execute(CREATE_TABLES)
    print("Schema ready.")

INSERT - Adding Records

async def create_user(pool: asyncpg.Pool, username: str, email: str) -> int:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """
            INSERT INTO users (username, email)
            VALUES ($1, $2)
            ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email
            RETURNING id, created_at
            """,
            username,
            email,
        )
    return row["id"]

async def create_post(
    pool: asyncpg.Pool,
    user_id: int,
    title: str,
    body: str,
    published: bool = False,
) -> int:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """
            INSERT INTO posts (user_id, title, body, published)
            VALUES ($1, $2, $3, $4)
            RETURNING id
            """,
            user_id,
            title,
            body,
            published,
        )
    return row["id"]

# Bulk insert with copy_records_to_table - extremely fast
async def bulk_insert_posts(pool: asyncpg.Pool, rows: list[tuple]) -> None:
    async with pool.acquire() as conn:
        await conn.copy_records_to_table(
            "posts",
            records=rows,
            columns=["user_id", "title", "body", "published"],
        )
    print(f"Bulk inserted {len(rows)} posts.")

SELECT - Querying Records

async def get_user_posts(
    pool: asyncpg.Pool,
    user_id: int,
    limit: int = 20,
    offset: int = 0,
) -> list[asyncpg.Record]:
    async with pool.acquire() as conn:
        return await conn.fetch(
            """
            SELECT p.id, p.title, p.published, p.created_at, u.username
            FROM posts p
            JOIN users u ON u.id = p.user_id
            WHERE p.user_id = $1
            ORDER BY p.created_at DESC
            LIMIT $2 OFFSET $3
            """,
            user_id,
            limit,
            offset,
        )

async def search_posts(pool: asyncpg.Pool, keyword: str) -> list[asyncpg.Record]:
    async with pool.acquire() as conn:
        return await conn.fetch(
            """
            SELECT p.title, u.username, p.created_at
            FROM posts p
            JOIN users u ON u.id = p.user_id
            WHERE to_tsvector('english', p.title || ' ' || p.body)
                  @@ plainto_tsquery('english', $1)
            ORDER BY p.created_at DESC
            LIMIT 50
            """,
            keyword,
        )

UPDATE, DELETE, and Transactions

async def publish_post(pool: asyncpg.Pool, post_id: int) -> bool:
    async with pool.acquire() as conn:
        result = await conn.execute(
            "UPDATE posts SET published = TRUE WHERE id = $1",
            post_id,
        )
    return result == "UPDATE 1"

async def transfer_posts(
    pool: asyncpg.Pool,
    from_user: int,
    to_user: int,
) -> int:
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Both statements run atomically
            result = await conn.execute(
                "UPDATE posts SET user_id = $1 WHERE user_id = $2",
                to_user,
                from_user,
            )
            count = int(result.split()[-1])
            await conn.execute(
                "INSERT INTO audit_log (action, detail) VALUES ($1, $2)",
                "transfer_posts",
                f"from={from_user} to={to_user} count={count}",
            )
    return count

Full Working Example

async def main():
    pool = await create_pool()
    await setup_schema(pool)

    uid = await create_user(pool, "alice", "alice@example.com")
    print(f"Created user id={uid}")

    pid = await create_post(pool, uid, "asyncpg Deep Dive", "asyncpg is blazing fast...", True)
    print(f"Created post id={pid}")

    posts = await get_user_posts(pool, uid)
    for p in posts:
        print(f" [{p['id']}] {p['title']} published={p['published']}")

    await pool.close()

if __name__ == "__main__":
    asyncio.run(main())

Practical Tips

• Always use a pool - create_pool() is cheap once, connection acquisition is fast
• Use $1, $2, ... placeholders to prevent SQL injection (asyncpg uses positional params)
• conn.copy_records_to_table() is 10-50× faster than looped INSERT for bulk data
• async with conn.transaction() nests safely - inner blocks become savepoints
• fetchrow() returns None if no row matches - always check before accessing fields
• Enable server_settings={"statement_timeout": "5000"} to protect against runaway queries

Follow me for more Python tips! 🐍 💡

Related: **Content Creator Ultimate Bundle (Save 33%)* - $29.99*