1. Production Outage: Backend Collapses Under a Database Query Spike
Around 2 PM last Tuesday, our IoT system started triggering PagerDuty alerts non-stop. API latency surged from 40ms to 1.8s in just five minutes. Gateways receiving data from 15,000 sensor devices suffered mass timeouts without a circuit breaker to protect the upstream services, while our Redis queue piled up with over 200,000 messages.
The team’s first instinct was to check the database. Strangely enough, after SSHing into the PostgreSQL cluster (8 vCPUs, 32GB RAM) and checking htop alongside pg_stat_activity, DB CPU utilization was only at 22%, nearly half the RAM was free, and disk writes hovered around 15MB/s. The database was completely healthy. The real culprit lived at the Python application layer: hundreds of workers were I/O blocked because they exhausted the connection pool whenever querying Postgres.
2. Why Traditional psycopg2 Became the Bottleneck
The project originally relied on psycopg2 using synchronous (blocking I/O) mechanics. At low traffic around a few hundred req/s, everything ran smoothly. But hitting the 5,000 req/s threshold broke the architecture due to 3 fundamental weaknesses:
- Thread-blocking execution: Every time
cursor.execute()is called, the worker process must sit idle waiting for the database response. Python cannot leverage this I/O wait time to process other incoming tasks. - Connection handshake overhead: Establishing a new PostgreSQL connection takes 30-50ms for the TLS handshake and user authentication. If the app creates connections continuously, server CPU gets depleted just handling connection churn.
- Text protocol overhead: Many legacy drivers exchange data in plain text. When decoding heavy data types such as JSONB, Timestamp, or UUID, the application CPU must continuously parse strings, causing a significant drop in throughput.
3. Weighing 3 Potential Solutions
To rescue the system, our team evaluated 3 practical options:
Option 1: Scale up Gunicorn / Uvicorn workers
This approach was the easiest to implement, but unreasonably expensive. Each Python worker consumed around 110MB of RAM—profiling with tools for memory leak detection showed significant footprint overhead. Scaling from 8 to 32 workers burned through nearly 4GB of RAM and still bottlenecked whenever slow queries hit the DB. This was merely a temporary workaround that failed to address the root I/O problem.
Option 2: Use psycopg3 in async mode
psycopg3 supports async/await syntax while retaining DB-API 2.0 compliance. If you are refactoring a massive legacy codebase relying on existing database migration management for SQLAlchemy, this is a safe bet. However, maintaining backward compatibility means its raw data processing throughput still lags behind drivers built exclusively for AsyncIO.
Option 3: Fully migrate to asyncpg
asyncpg was written from scratch in Cython by MagicStack, engineered specifically for maximum PostgreSQL performance. It bypasses DB-API 2.0 to communicate directly with Postgres over the binary protocol. Real-world load testing and benchmark results: throughput increased by 3.5x, and application server CPU usage dropped by nearly 40% compared to psycopg2.
4. Implementing asyncpg: From Basic Connection to High-Throughput Processing
Installation
Install the library via pip:
pip install asyncpg
Initializing a Shared Connection Pool
Never open isolated connections inside individual endpoints. Create a single pool when the application starts up and share it across all requests:
import asyncio
import asyncpg
DATABASE_URL = "postgresql://postgres:[email protected]:5432/iot_db"
async def create_db_pool():
return await asyncpg.create_pool(
dsn=DATABASE_URL,
min_size=10, # Keep 10 idle connections ready
max_size=30, # Maximum 30 connections per instance
max_queries=50000, # Automatically recycle connection after 50k queries to prevent memory leaks
timeout=10.0 # Timeout if acquiring a connection from the pool exceeds 10 seconds
)
async def get_device_by_id(pool: asyncpg.Pool, device_id: str):
# Acquire 1 connection from the pool, automatically released when exiting the async with block
async with pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT id, name, firmware_version, created_at FROM devices WHERE id = $1",
device_id
)
return dict(row) if row else None
Key takeaway: asyncpg uses positional placeholders $1, $2, $3... instead of %s or ?.
High-Throughput Ingestion: Skip the For Loop, Use copy_records_to_table
When you need to persist 100,000 sensor logs into the database, your implementation determines whether the system survives. Iterating with a for loop and calling await conn.execute() per row takes over 45 seconds. Switching to executemany() drops this to around 4.2 seconds. But using copy_records_to_table (leveraging Postgres binary COPY) takes just 0.82 seconds:
import time
async def bulk_insert_sensor_logs(pool: asyncpg.Pool, logs_data: list[tuple]):
async with pool.acquire() as conn:
start_time = time.perf_counter()
# logs_data format: [('dev-001', 28.5, 65.2, 1718000000), ...]
await conn.copy_records_to_table(
table_name="sensor_logs",
records=logs_data,
columns=["device_id", "temperature", "humidity", "recorded_at"]
)
elapsed = time.perf_counter() - start_time
print(f"Inserted {len(logs_data):,} records in {elapsed:.2f}s")
Safe Transaction Management
Financial transactions or inventory updates require strict ACID guarantees. With asyncpg, transaction management is exceptionally clean via context managers:
async def deduct_wallet_balance(pool: asyncpg.Pool, user_id: int, amount: float):
async with pool.acquire() as conn:
async with conn.transaction():
# Lock row to prevent race conditions
balance = await conn.fetchval(
"SELECT balance FROM wallets WHERE user_id = $1 FOR UPDATE",
user_id
)
if balance is None or balance < amount:
raise ValueError("Insufficient balance")
await conn.execute(
"UPDATE wallets SET balance = balance - $1 WHERE user_id = $2",
amount, user_id
)
await conn.execute(
"INSERT INTO audit_logs (user_id, amount, action) VALUES ($1, $2, 'WITHDRAW')",
user_id, amount
)
5. Production Operational Best Practices
- Pool sizing formula: Never configure
max_size = 100arbitrarily. A reliable real-world formula is:Pool Size = ((CPU Cores * 2) + Disk Count) / Number of app instances. For a 4-worker app cluster on an 8-core NVMe Postgres server, each worker only needs a pool of 5-10 connections. - No json.dumps() needed for JSONB: asyncpg automatically decodes JSONB columns directly into Python dict/list and vice versa using the binary protocol. Strip out manual
json.loads()calls to save CPU cycles. - Graceful Shutdown: Always listen for the
SIGTERMsignal and callawait pool.close()inside the FastAPI/Sanic lifespan handler. This ensures all connection sockets are cleanly closed, preventing orphaned (idle in transaction) processes from hanging the database.
