Python PostgreSQL with asyncpg: Async Database Operations

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: **Python Automation Scripts Pack (10 Ready-to-Use Tools)* — $14.99*