Python Performance Engineer

{
  "analysis": {
    "bottleneck_type": "database_io",
    "severity": "high",
    "affected_endpoints": ["/api/v1/orders", "/api/v1/products"],
    "root_cause": "N+1 query pattern in order items loader"
  },
  "metrics": {
    "before": {"p50_ms": 450, "p95_ms": 1200, "p99_ms": 2500},
    "after": {"p50_ms": 45, "p95_ms": 120, "p99_ms": 250},
    "improvement": "10x latency reduction"
  },
  "optimizations_applied": [
    {"type": "query", "description": "Added eager loading for order_items", "impact": "Reduced queries from N+1 to 2"},
    {"type": "cache", "description": "Added Redis cache for product catalog", "impact": "90% cache hit rate"},
    {"type": "pool", "description": "Tuned connection pool: min=5, max=20", "impact": "Eliminated connection wait time"}
  ],
  "recommendations": [
    {"priority": "high", "action": "Add database index on orders.customer_id"},
    {"priority": "medium", "action": "Consider read replicas for reporting queries"}
  ],
  "load_test_results": {
    "tool": "k6",
    "scenario": "100 VUs, 5 min duration",
    "throughput_rps": 850,
    "error_rate": "0.1%"
  }
}

# Quick profiling with py-spy
# py-spy record -o profile.svg --pid <PID>

# Code-level profiling
import cProfile
import pstats
from io import StringIO

def profile_function(func, *args, **kwargs):
    profiler = cProfile.Profile()
    profiler.enable()
    result = func(*args, **kwargs)
    profiler.disable()

    stream = StringIO()
    stats = pstats.Stats(profiler, stream=stream)
    stats.sort_stats('cumulative')
    stats.print_stats(20)
    print(stream.getvalue())

    return result

# Line-level profiling
# pip install line_profiler
# kernprof -l -v script.py
@profile  # decorator for line_profiler
def expensive_function():
    pass

import tracemalloc
from memory_profiler import profile

# Track memory allocations
tracemalloc.start()
# ... code to analyze ...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
for stat in top_stats[:10]:
    print(stat)

# Find memory leaks
import objgraph
objgraph.show_growth(limit=10)
objgraph.show_most_common_types(limit=10)

# Function-level memory
@profile
def memory_intensive():
    data = [i ** 2 for i in range(1000000)]
    return sum(data)

import asyncio
from asyncio import TaskGroup

# Parallel I/O with TaskGroup (Python 3.11+)
async def fetch_all_data(ids: list[str]) -> list[dict]:
    async with TaskGroup() as tg:
        tasks = [tg.create_task(fetch_one(id)) for id in ids]
    return [t.result() for t in tasks]

# Connection pooling for asyncpg
import asyncpg

pool = await asyncpg.create_pool(
    dsn,
    min_size=5,
    max_size=20,
    max_inactive_connection_lifetime=300,
    command_timeout=60,
)

# Bounded concurrency
sem = asyncio.Semaphore(10)
async def limited_fetch(url):
    async with sem:
        return await fetch(url)

from sqlalchemy.orm import selectinload, joinedload

# BEFORE: N+1 queries
orders = await session.execute(select(Order))
for order in orders.scalars():
    print(order.items)  # Triggers query per order!

# AFTER: Eager loading
stmt = select(Order).options(
    selectinload(Order.items),  # 2 queries total
    joinedload(Order.customer), # Single join
)
orders = await session.execute(stmt)

# Index hints for PostgreSQL
from sqlalchemy import Index
Index('ix_orders_customer_date', Order.customer_id, Order.created_at)

# Query analysis
from sqlalchemy import event

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def log_query(conn, cursor, statement, parameters, context, executemany):
    logger.debug(f"Query: {statement}")
    logger.debug(f"Params: {parameters}")

import redis.asyncio as redis
from functools import wraps
import hashlib
import json

redis_client = redis.from_url("redis://localhost:6379")

def cache(ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Generate cache key
            key_data = f"{func.__name__}:{args}:{kwargs}"
            cache_key = hashlib.md5(key_data.encode()).hexdigest()

            # Try cache
            cached = await redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # Execute and cache
            result = await func(*args, **kwargs)
            await redis_client.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cache(ttl=60)
async def get_product(product_id: str) -> dict:
    return await db.fetch_product(product_id)

# Cache invalidation
async def invalidate_product(product_id: str):
    pattern = f"get_product:{product_id}:*"
    keys = await redis_client.keys(pattern)
    if keys:
        await redis_client.delete(*keys)

// k6 load test script (load-test.js)
import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  stages: [
    { duration: '1m', target: 50 },  // Ramp up
    { duration: '3m', target: 100 }, // Hold
    { duration: '1m', target: 0 },   // Ramp down
  ],
  thresholds: {
    http_req_duration: ['p(95)<200', 'p(99)<500'],
    http_req_failed: ['rate<0.01'],
  },
};

export default function () {
  const res = http.get('http://localhost:8000/api/v1/orders');
  check(res, {
    'status is 200': (r) => r.status === 200,
    'response time < 200ms': (r) => r.timings.duration < 200,
  });
  sleep(1);
}

# Run k6
k6 run load-test.js

# With HTML report
k6 run --out json=results.json load-test.js

{
  "bottleneck": "N+1 queries for order items",
  "solution": "selectinload() + Redis cache",
  "improvement": "p95: 1200ms -> 85ms"
}

# Check process memory
ps aux | grep python

# Live CPU profile
py-spy top --pid <PID>

# Database slow queries (PostgreSQL)
psql -c "SELECT query, calls, mean_time FROM pg_stat_statements ORDER BY mean_time DESC LIMIT 10;"

# Redis memory stats
redis-cli INFO memory

# Connection pool status (from app)
curl localhost:8000/health/db-pool