asyncpg is an asyncio-friendly Python client for running non-blocking queries against PostgreSQL. It uses coroutines and non-blocking sockets, allowing database operations to suspend rather than block the event loop. section 5.0 section 5.1
Core operations include:
- Connect with asyncpg.connect, execute SQL with execute, and retrieve records with fetch or fetchrow. section 5.2 section 5.4
- Represent returned rows as asyncpg Record objects, which support dictionary-like column-name access. section 5.4
- Use create_pool with min_size and max_size to share connections across concurrent coroutines; acquire waits when the pool is busy. section 5.5.2
- Wrap related statements in connection.transaction(); successful blocks commit, while exceptions automatically roll back. section 5.6
- Use nested transaction contexts as PostgreSQL savepoints when an inner failure should not undo earlier outer work. section 5.6.1
- Stream large result sets through cursors and asynchronous generators instead of loading every row into memory. section 5.7
Concurrent product queries
A pooled connection can serve a product lookup coroutine, while asyncio.gather runs many lookups together. In the demonstrated comparison, 10,000 synchronous queries took 21.8274 seconds, compared with 4.8464 seconds concurrently. section 5.5.2
Atomic updates
Place two brand inserts inside async with connection.transaction(). If the second insert violates a primary-key constraint, asyncpg raises an error and leaves neither insert committed. section 5.6
Listing 5.7 Establishing a connection pool and running queries concurrently
import asyncio
import asyncpg
product_query = \
"""
SELECT
p.product_id,
p.product_name,
p.brand_id,
s.sku_id,
pc.product_color_name,
ps.product_size_name
FROM product as p
JOIN sku as s on s.product_id = p.product_id
JOIN product_color as pc on pc.product_color_id = s.product_color_id
JOIN product_size as ps on ps.product_size_id = s.product_size_id
WHERE p.product_id = 100"""
async def query_product(pool):
async with pool.acquire() as connection:
return await connection.fetchrow(product_query)
async def main():
async with asyncpg.create_pool(host='127.0.0.1',
port=5432,
user='postgres',
password='password',
database='products',
min_size=6,
max_size=6) as pool: #1
await asyncio.gather(query_product(pool),
query_product(pool)) #2
asyncio.run(main())
Shows the pool-based query function and the contrast between sequential awaits and asyncio.gather. section 5.5.2
