A 4-second API response time is not a mystery — it's a measurement problem. Every slow API has a specific cause: a query that fetches 200 rows to return 5, a cache miss on data that's identical for every user, a synchronous external API call that adds 2 seconds, or a connection pool that's exhausted before the request starts. The bottleneck is findable with the right instrumentation.
This post covers the diagnostic process and the fixes that actually move the number.
Step 1: Get Actual Timing Data
Before optimizing anything, get precise timing data at the right granularity. A 4-second total response time can be: - 3.8 seconds of database time - 3.5 seconds of a third-party API call - 2 seconds of serialization of a massive JSON payload - Or all of the above
Without per-component timing, you're guessing.
Add OpenTelemetry tracing if you don't have it:
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
# Auto-instrument FastAPI, SQLAlchemy, and httpx
FastAPIInstrumentor.instrument_app(app)
SQLAlchemyInstrumentor().instrument()
HTTPXClientInstrumentor().instrument()
Auto-instrumentation for web frameworks, ORMs, and HTTP clients captures per-query and per-external-call timing without manual span creation. Send traces to Jaeger, Tempo, or DataDog — whichever you already have.
If you can't add full instrumentation immediately, add timing logs to the specific endpoint:
import time
import logging
logger = logging.getLogger(__name__)
async def slow_endpoint(request_id: str):
t0 = time.perf_counter()
data = await fetch_from_db(request_id)
t1 = time.perf_counter()
enriched = await enrich_with_external_api(data)
t2 = time.perf_counter()
response = build_response(enriched)
t3 = time.perf_counter()
logger.info("endpoint_timing", extra={
"request_id": request_id,
"db_ms": round((t1 - t0) * 1000, 1),
"external_api_ms": round((t2 - t1) * 1000, 1),
"serialization_ms": round((t3 - t2) * 1000, 1),
"total_ms": round((t3 - t0) * 1000, 1),
})
return response
Run this against your slow endpoints and look at the timing distribution — the dominant component is your starting point.
The N+1 Query Problem
N+1 queries are the most common cause of unexpectedly slow database-heavy APIs. The pattern:
# BAD: N+1 — 1 query for users, then N queries for each user's posts
users = await db.execute("SELECT * FROM users WHERE active = true")
for user in users:
posts = await db.execute(f"SELECT * FROM posts WHERE user_id = {user.id}")
user.recent_posts = posts
This runs 1 + N queries (where N = number of users). For 50 active users, that's 51 queries. Each query has network round-trip overhead — on a database server 1ms away, that's 50ms minimum overhead in network RTTs alone.
Fix: JOIN or eager loading:
# GOOD: 1 query with JOIN
result = await db.execute("""
SELECT u.*,
json_agg(p.* ORDER BY p.created_at DESC) FILTER (WHERE p.id IS NOT NULL) as recent_posts
FROM users u
LEFT JOIN posts p ON p.user_id = u.id AND p.created_at > now() - interval '7 days'
WHERE u.active = true
GROUP BY u.id
""")
With SQLAlchemy ORM, use eager loading:
from sqlalchemy.orm import selectinload
# Instead of lazy loading (N+1), use selectinload (2 queries total)
result = await db.execute(
select(User)
.options(selectinload(User.recent_posts))
.where(User.active == True)
)
Find N+1 queries in your codebase: look for queries inside loops, or use an ORM profiler (SQLAlchemy's echo=True in development, or query logging with plan analysis).
Caching: What to Cache and How
What to cache: - User profile data that doesn't change per request - Lookup tables (categories, feature flags, country codes) - Expensive aggregation query results that many users share - External API responses that are identical for the same inputs
What not to cache: - Data that must be real-time (inventory, account balance) - User-specific personalized data (unless the scope is clear) - Data with complex invalidation requirements that you'll get wrong
Redis caching pattern:
import json
import hashlib
from functools import wraps
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost:6379")
def cache(ttl_seconds: int = 300, key_prefix: str = ""):
"""Cache decorator for async functions."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Build cache key from function name and args
cache_key = f"{key_prefix}:{func.__name__}:{hashlib.md5(str(args + tuple(sorted(kwargs.items()))).encode()).hexdigest()}"
# Try cache first
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Cache miss — execute function
result = await func(*args, **kwargs)
# Store in cache
await redis_client.setex(cache_key, ttl_seconds, json.dumps(result, default=str))
return result
return wrapper
return decorator
@cache(ttl_seconds=60, key_prefix="user_profile")
async def get_user_profile(user_id: str) -> dict:
# This only hits the database once per minute per user_id
return await db_fetch_user_profile(user_id)
Async External API Calls in Parallel
Synchronous external API calls that block the response are a common slow path. If you're calling 3 APIs sequentially, you wait for all 3 in series:
# BAD: sequential external calls (sum of all latencies)
stripe_data = await stripe_client.get_customer(customer_id) # 300ms
analytics_data = await analytics.get_user_stats(user_id) # 400ms
notifications_data = await notification_service.get_count(user_id) # 200ms
# Total: 900ms
# GOOD: parallel external calls (max of all latencies)
import asyncio
stripe_task = asyncio.create_task(stripe_client.get_customer(customer_id))
analytics_task = asyncio.create_task(analytics.get_user_stats(user_id))
notifications_task = asyncio.create_task(notification_service.get_count(user_id))
stripe_data, analytics_data, notifications_data = await asyncio.gather(
stripe_task, analytics_task, notifications_task
)
# Total: max(300, 400, 200) = 400ms
Parallel external calls convert sequential latency to concurrent latency. This is often the single largest latency improvement available for APIs that aggregate from multiple services.
Database Connection Pool Exhaustion
A fully loaded API server waiting for an available database connection adds seconds to response time. Symptoms: response times spike under load, even though queries are fast in isolation.
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
DATABASE_URL,
pool_size=20, # Base pool size (default 5 — often too small)
max_overflow=10, # Additional connections allowed under load
pool_timeout=30, # Raise error rather than wait indefinitely
pool_pre_ping=True, # Verify connections are live before use
)
Connection pool sizing: start with pool_size = (expected peak concurrent requests) × 0.5. Monitor pool_size vs. actual pool usage — a pool that's always at max is undersized; a pool at 30% is wasteful.
Rutagon Fixes Performance Problems That Are Blocking Growth
If your API performance is limiting customer experience or scaling costs, Rutagon diagnoses and implements the fixes. Contact us at rutagon.com/contact.
See also: database query optimization service and engineering capabilities at Rutagon.
FAQ
How do I identify which endpoint is slow without full APM tooling?
Access logs are your starting point. Most web frameworks log request path and response time. Sort your access logs by response time in descending order — the slowest endpoints will surface. Alternatively, query your logging service (CloudWatch, Datadog) for p99 response time by endpoint path. Even basic access log analysis identifies the 20% of endpoints causing 80% of latency.
Is it worth caching database queries in production?
For read-heavy APIs with predictable query patterns, yes — caching can reduce database load by 60-80% and cut response times dramatically. The cost: cache invalidation complexity and the risk of stale data. Start with low-risk data (lookup tables, user profiles) with short TTLs before caching data with complex invalidation requirements. Never cache data where staleness would cause incorrect business decisions (pricing, inventory, balances).
How does database indexing affect API performance?
Missing indexes on frequently queried columns are one of the top performance problems in growing applications. A query that scans 100,000 rows to find 10 results can be transformed to a direct lookup with the right index. Use EXPLAIN ANALYZE on slow queries in PostgreSQL to see whether indexes are being used and where the query time is spent. Add indexes for: columns in WHERE clauses of high-frequency queries, foreign keys used in JOINs, and sort columns in ORDER BY clauses.
When is it time to consider read replicas vs. query optimization?
Optimize queries first — a read replica doesn't help an N+1 query. When your write-heavy primary database is genuinely capacity-constrained (CPU > 70% sustained, IOPS limits), or when read traffic is dominating over write traffic, read replicas allow distributing read load. Most applications need query optimization before they need read replicas.
What's the performance impact of serializing large JSON responses?
For APIs returning large JSON payloads (thousands of objects), serialization can be a significant percentage of total response time. Options: reduce the payload (return only fields the client uses), use more efficient serialization libraries (orjson is 5-10× faster than Python's built-in json), implement pagination to limit response size, or use projection parameters to let clients specify which fields to return. Measure serialization time specifically before assuming it's a bottleneck.