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", "[email protected]")
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 loopedINSERTfor bulk data
async with conn.transaction()nests safely — inner blocks become savepoints
fetchrow()returnsNoneif no row matches — always check before accessing fields
Enableserver_settings={"statement_timeout": "5000"}to protect against runaway queries
Follow me for more Python tips! 🐍
💡 Related: **Content Creator Ultimate Bundle (Save 33%)* — $29.99*
SOCIAL SHARE CARD GENERATOR