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*