3 Faster PostgreSQL with asyncpg

#python #postgres #tutorial #async

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

🛠️ Useful resource: **Content Creator Ultimate Bundle (Save 33%)* — $29.99. Check it out on Gumroad!*

📧 Get my FREE Python Cheatsheet → Follow me on Dev.to and drop a comment below — I'll DM you the cheatsheet directly!

🐍 50+ essential Python patterns, one-liners, and best practices for everyday development. Free for all readers.

喜欢这篇文章？关注获取更多Python自动化内容！

DEV Community