DEV Community: Sir Max

4 Docker Compose Patterns That Saved Our Team Hours of Debugging

Sir Max — Tue, 11 Aug 2026 02:09:18 +0000

4 Docker Compose Patterns That Saved Our Team Hours of Debugging

I spent most of last year running a half-dozen services on a single VPS. Database, API gateway, a few workers, Redis, the usual stack. Docker Compose was the obvious choice — but my early configs were a mess of trial and error.

Here are four patterns I learned the hard way. Each one solved a real problem that cost me at least an evening of frustrated debugging.

1. Healthchecks That Actually Work

My first healthcheck looked like this:

# ❌ Too aggressive — restarts the container before it's ready
healthcheck:
  test: ["CMD", "curl", "-f", "http://localhost:3000/health"]
  interval: 5s
  timeout: 3s
  retries: 3
  start_period: 5s

The problem: my app took 15–20 seconds to warm up (DB migrations, cache loading). Compose gave it 5 seconds, then killed it. I'd get a cascade of depends_on failures and every container restarting in a loop.

The fix is simpler than you think — make start_period match reality:

# ✅ Give the app time to actually start
healthcheck:
  test: ["CMD-SHELL", "wget -q -O- http://localhost:3000/health || exit 1"]
  interval: 10s
  timeout: 5s
  retries: 5
  start_period: 30s  # This is the key line

The start_period is how long Compose waits before counting failures. Set it to your app's actual warmup time. For a Node.js app with migrations, 30 seconds is a safe default. For a Go binary, 5 seconds might be enough.

Pro tip: Use CMD-SHELL instead of CMD. The shell form handles pipes and redirects. The array form doesn't — which is why curl often fails silently when your endpoint returns a redirect.

2. Profiles: One File, Two Environments

I used to maintain two separate compose files:

docker-compose.yml for production
docker-compose.dev.yml for development

Every change had to be synced. I missed a port mapping once and spent an hour wondering why my local debugger couldn't connect.

Docker Compose profiles solve this cleanly:

services:
  api:
    build: .
    ports: ["3000:3000"]
    environment:
      - NODE_ENV=production

  # Dev-only services
  adminer:
    image: adminer
    profiles: ["dev"]
    ports: ["8080:8080"]

  mailhog:
    image: mailhog/mailhog
    profiles: ["dev"]
    ports: ["1025:1025", "8025:8025"]

  # Prod-only overrides
  api-dev:
    extends:
      service: api
    profiles: ["dev"]
    volumes: ["./src:/app/src"]
    environment:
      - NODE_ENV=development

Now I just flip a flag:

# Production
docker compose up -d

# Development (all dev tools)
docker compose --profile dev up -d

All the core services start in both environments. Dev tools (Adminer, MailHog, hot-reload) only come up with --profile dev. One file, no duplication, no drift.

3. `extends` for Dry Compose Files

Here is a pattern I wish I'd discovered sooner. Instead of repeating environment variables across services:

# ❌ Repeated env vars everywhere
services:
  worker-1:
    environment:
      - DB_HOST=postgres
      - DB_PORT=5432
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - LOG_LEVEL=info
  worker-2:
    environment:
      - DB_HOST=postgres
      - DB_PORT=5432
      - REDIS_HOST=redis
      # ... same thing again

Use extends with a base service:

# ✅ Define once, reuse everywhere
x-base-env: &base-env
  DB_HOST: postgres
  DB_PORT: 5432
  REDIS_HOST: redis
  REDIS_PORT: 6379
  LOG_LEVEL: info

services:
  worker-1:
    environment:
      <<: *base-env
      WORKER_NAME: processor-1

  worker-2:
    environment:
      <<: *base-env
      WORKER_NAME: processor-2

This is YAML anchors, not a Docker feature — but Docker Compose supports them fully. I use anchors for environment blocks, volume mounts, and even entire service definitions when I need multiple similar containers.

4. Wait-For-It Without Custom Scripts

The classic problem: your API starts before the database is ready. Connection refused. Container restarts. Log spam.

The old solution was a wait-for-it.sh script in every container. The modern solution is built into Compose:

services:
  api:
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy

  postgres:
    image: postgres:16
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

The condition: service_healthy tells Compose: "Do not start this container until that container passes its healthcheck." No custom scripts, no sleep loops, no guesswork.

What These Patterns Share

Looking back, all four patterns solve the same underlying problem: making startup order predictable. Most Docker Compose debugging sessions start with "which container failed first?" Once you can answer that reliably, the rest is just configuration.

I still keep a docker-compose.debug.yml override file for those 2 a.m. incidents — it cranks up log levels, exposes extra ports for profiling, and adds a tty: true so I can docker attach when things really go sideways. But the four patterns above eliminated about 80% of my late-night debugging sessions.

If you have your own Compose patterns that saved you, drop them in the comments. I am always looking for new tricks.

3 Things I Wish I Knew Before Building with LLM APIs

Sir Max — Mon, 10 Aug 2026 02:15:30 +0000

3 Things I Wish I Knew Before Building with LLM APIs

I've spent the last year shipping features backed by LLM APIs — GPT, Claude, Gemini, and a handful of open-source models via Ollama. Along the way I burned through a lot of credits, broke production a couple of times, and learned things the hard way.

Here are three lessons I wish someone had told me before I started.

1. Token counting is not optional — it's the whole job

Early on, I treated token limits as a "nice to have" — something you throw a rough character count at and hope for the best. That worked fine until a user pasted a 40KB log file into a chat window and my backend silently truncated their context, giving answers that looked confident but were completely wrong.

What I do now: Every request goes through a token counter before it touches the LLM.

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

For multi-model setups, I built a small abstraction that maps model names to their tokenizers. claude-* uses Anthropic's claude-tokenizer, Gemini uses google-generativeai, and for Ollama models I fall back to a conservative estimate based on character count.

The real insight: You need two limits — a hard cutoff for the full context window, and a softer "reserved output" budget so the model has room to actually respond. I set aside 25% of the context window for output tokens. If your input + reserved output > context limit, trim the input — not the other way around.

This one change eliminated about 90% of the "why is the AI giving incomplete answers?" bug reports.

2. Streaming is table stakes, but structured output is the hard part

Everyone knows you should stream LLM responses — users don't want to stare at a spinner for 15 seconds. That's the easy part.

The hard part is what to do with the stream. If you're calling an LLM from a backend that needs to return JSON to a frontend, you can't just stream raw tokens. You need structured output on the backend side, and you need it fast.

The pattern that works for me:

import asyncio
import json

async def stream_and_parse(prompt: str, schema: dict):
    """Stream LLM response, parse JSON at the end."""
    buffer = []

    async for chunk in llm_stream(prompt):
        buffer.append(chunk)
        # Send raw text to frontend via WebSocket
        await ws.send(chunk)

    full_text = "".join(buffer)
    # Parse the structured result from completed text
    return json.loads(full_text)

But here's the gotcha: not all LLMs are equally good at JSON output. Claude is great at it. Gemini 2.0 Flash is fast but occasionally drops closing braces. GPT-4o handles it well but sometimes wraps JSON in markdown fences that you need to strip.

My rule of thumb: Always validate the parsed JSON against your schema. If parsing fails, retry with a system prompt that says "Return ONLY valid JSON, no markdown fences, no explanations." That second attempt fixes 95% of failures.

For production, I also log every JSON parse failure with the raw response — nothing is more frustrating than a silent parse error that turns into a NoneType has no attribute three callers later.

3. Fallback is not a feature — it's architecture

My first LLM integration looked like this:

user → my API → OpenAI → response

When OpenAI had an outage (and it did, twice in six months), my entire feature was dead. Users got timeout errors with no explanation.

What I run now:

user → my API → primary provider (Claude)
              → fallback #1 (GPT-4o) — different provider
              → fallback #2 (Gemini Flash) — cheap, always available

But it's not enough to just chain fallbacks. You need circuit breakers:

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = 0

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()

    def is_open(self):
        if self.failures >= self.threshold:
            if time.time() - self.last_failure_time < self.reset_timeout:
                return True  # Circuit is open — skip this provider
            self.failures = 0  # Reset timeout passed, try again
        return False

If Claude returns three consecutive 5xx errors, the circuit breaker opens for 60 seconds. Requests skip directly to GPT-4o instead of waiting for timeouts. When the breaker resets, it tests the water with a single request before fully reopening.

The lesson: LLM APIs are external dependencies. Treat them like you'd treat a database connection or a third-party payment service. They will fail, and your users shouldn't notice.

The takeaway

Building with LLMs isn't hard because the models are complex — it's hard because the infrastructure around them needs to be resilient. Token management, structured parsing, and provider fallback are not optional extras after launch. They're the stuff you need on day one.

These three patterns have been running in production for months now. They're not particularly clever or novel. They're just battle-tested.

Got your own LLM API war stories? I'd love to hear them in the comments.

3 Database Query Patterns That Kill Performance (And How to Fix Them)

Sir Max — Sun, 09 Aug 2026 02:05:49 +0000

3 Database Query Patterns That Kill Performance (And How to Fix Them)

I've spent more hours than I'd like to admit staring at slow database queries in production. Every time, I think "this should be fast" — and every time, I'm wrong. Here are three patterns I've stumbled over repeatedly, plus the fixes that actually worked.

1. The N+1 Problem: Your ORM Is Lying to You

This is the classic. You fetch a list of users, then loop through and fetch each user's orders — one query at a time.

# ❌ The N+1 trap
users = db.execute("SELECT id, name FROM users LIMIT 100").fetchall()
for user in users:
    orders = db.execute(
        "SELECT * FROM orders WHERE user_id = ?", [user.id]
    ).fetchall()
    # 101 queries total. Oof.

The ORM makes this look innocent. Django's user.order_set.all() in a template loop? That's N+1. SQLAlchemy's lazy-loading relationships? Same thing.

The Fix: JOIN or Preload

# ✅ One query
rows = db.execute("""
    SELECT u.id, u.name, o.id as order_id, o.total, o.created_at
    FROM users u
    LEFT JOIN orders o ON o.user_id = u.id
    WHERE u.id IN (SELECT id FROM users LIMIT 100)
    ORDER BY u.id, o.created_at
""").fetchall()

# Group in application code
from collections import defaultdict
user_orders = defaultdict(list)
for row in rows:
    user_orders[row['id']].append(row)

If you're stuck with an ORM, use select_related() / prefetch_related() in Django or joinedload() / selectinload() in SQLAlchemy. Just know what SQL they generate — run EXPLAIN and check.

2. LIKE '%wildcard%' Searches: The Index Killer

A leading wildcard in LIKE (%something) prevents any B-tree index from being used. The database has to scan every row.

-- ❌ Full table scan — index is useless
SELECT * FROM articles WHERE title LIKE '%postgresql%';

On a table with a few thousand rows, you might not notice. At 500K rows? Your API timeout is gone.

The Fix: Full-Text Search

PostgreSQL has built-in full-text search. MySQL has it too. Use it.

-- PostgreSQL: create a tsvector column and index it
ALTER TABLE articles ADD COLUMN search_vector tsvector
    GENERATED ALWAYS AS (to_tsvector('english', title || ' ' || body)) STORED;

CREATE INDEX idx_articles_search ON articles USING GIN(search_vector);

-- Query with proper ranking
SELECT title, ts_rank(search_vector, query) AS rank
FROM articles, plainto_tsquery('english', 'postgresql performance') query
WHERE search_vector @@ query
ORDER BY rank DESC
LIMIT 20;

For simpler cases, ILIKE with a trigram index (pg_trgm) handles partial matches reasonably well:

CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX idx_articles_title_trgm ON articles USING GIN(title gin_trgm_ops);

-- Now this uses the index
SELECT * FROM articles WHERE title ILIKE '%postgresql%';

When to NOT do full-text search: If your search is always prefix-based (WHERE title LIKE 'postgresql%'), a regular B-tree index works fine. Don't over-engineer.

3. Missing Composite Indexes: One Column at a Time Won't Cut It

I've seen this everywhere: a table with separate indexes on user_id, status, and created_at, and a query like:

SELECT * FROM orders
WHERE user_id = 42 AND status = 'pending'
ORDER BY created_at DESC
LIMIT 10;

PostgreSQL might use the user_id index and then filter status manually. Or it might scan the status index and filter user_id. Either way, it's sorting in memory and throwing away rows.

The Fix: Composite Index That Matches the Query

-- ✅ One index covers the full query pattern
CREATE INDEX idx_orders_user_status_created
    ON orders(user_id, status, created_at DESC);

Column order matters. Put the equality filters first, then the range/sort column:

user_id = 42 → equality, goes first
status = 'pending' → equality, goes second
ORDER BY created_at DESC → sort column, goes last with the sort direction

With this index, the query scans exactly the right rows in order — no sorting, no filtering junk rows.

Use EXPLAIN ANALYZE to verify:

EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 42 AND status = 'pending'
ORDER BY created_at DESC LIMIT 10;

-- Look for: Index Scan (not Seq Scan), actual rows close to planned rows

What I Actually Do Now

Every time I write a query that touches more than one table or has a WHERE clause beyond a primary key lookup, I:

Run EXPLAIN before merging the PR. Not in production — in the PR.
Set statement_timeout on read replicas. A 30-second query at 3 AM is a 3-second query times 10 concurrent users at 3 PM.
Log slow queries with log_min_duration_statement. You can't fix what you don't measure.

-- In postgresql.conf or per-session
SET statement_timeout = '10s';
SET log_min_duration_statement = '1s';

Most of the "mysterious" performance issues I've debugged came down to one of these three patterns. The fix usually took 10 minutes once I found it — the hard part was knowing where to look.

What query patterns have bitten you in production? I'd love to hear about them.

How I Cut Debugging Time in Half: 4 Practices That Actually Work

Sir Max — Sat, 08 Aug 2026 02:06:16 +0000

How I Cut Debugging Time in Half: 4 Practices That Actually Work

I used to spend hours chasing bugs that, in hindsight, should have taken 15 minutes. You know the pattern: stare at code, add print statements, refresh, stare more, eventually fix the wrong thing, repeat.

After shipping production code for a few years and debugging over 100 incidents, I've found four practices that consistently cut my debugging time. Not "productivity hacks" — just actual habits that changed how I approach broken code.

1. Structured Logging Before You Need It

The worst time to realize your logs are useless is at 2 AM during an incident.

I learned this the hard way. We had a payment processing service that would occasionally fail with a generic "transaction error." The logs? Just the error message. No request ID, no payload, no upstream response.

What I do now: Every HTTP request gets a unique request_id (UUID) that propagates through all downstream calls:

import uuid
import logging
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id")

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        try:
            record.request_id = request_id_var.get()
        except LookupError:
            record.request_id = "no-request-id"
        return True

def log_incoming(func):
    async def wrapper(request, *args, **kwargs):
        rid = str(uuid.uuid4())
        request_id_var.set(rid)
        logger.info("request_start", extra={"method": request.method, "path": request.url.path})
        try:
            response = await func(request, *args, **kwargs)
            logger.info("request_end", extra={"status": response.status_code})
            return response
        except Exception:
            logger.exception("request_error")
            raise
    return wrapper

Use structured logging (JSON), not formatted strings. You can query request_id=abc123 and see the entire lifecycle of that one request — from gateway through service to database and back.

2. Reproduce First, Fix Second

This sounds obvious. But I've watched developers (including past me) jump straight into code changes based on a vague error description. "The search page is slow" → start adding indexes. Wrong move.

The rule: If you can't reproduce it, you don't understand it.

My workflow now:

Capture the exact input — request payload, user session, timing. Screenshots aren't enough.
Reproduce locally with that input — if you can't, the bug isn't what you think it is.
Write a failing test — this is your proof you understand the problem.

I once spent two hours "fixing" a race condition that didn't exist. The real problem was a misconfigured load balancer timeout. If I'd actually reproduced the issue — sent the same request payload and watched the full trace — I would have spotted it in 10 minutes.

Hard rule: no code change until I've reproduced the failure with my own eyes.

3. Binary Search Debugging

When the bug could be anywhere in a complex system, don't guess. Bisect.

The technique is the same as git bisect, but applied to the runtime pipeline:

Check the middle: Is it still broken if you skip half the steps?
Narrow down: Which half contains the problem?
Repeat: Until you're down to one function or one configuration change.

Example from a real incident: Users reported intermittent 500 errors on a dashboard page. The pipeline was: Browser → CDN → API Gateway → Auth Service → Dashboard Service → Database.

Instead of checking each layer sequentially, I started in the middle:

Hit the Dashboard Service directly (bypass gateway + auth) → still failing. Right half (Auth→DB) is ruled out.
Check the gateway logs → errors there too. Left half contains the issue.
Compare CDN vs direct-to-gateway → CDN was fine. Gateway was the culprit.
Gateway config diff → someone had changed the timeout from 30s to 5s a day earlier.

Twenty minutes total. Sequential debugging would have taken hours.

4. The Debug Journal

I keep a plain text file open during any non-trivial debugging session. Every observation goes in:

14:05 - user reports "page not loading" with screenshot
14:08 - reproduced on staging. 502 from upstream
14:12 - upstream service logs show OOMKilled
14:15 - memory graph shows spike at 13:45 — same time as batch job
14:20 - confirmed: batch job memory leak, fixed in v2.1 but not deployed
14:30 - deployed v2.1, monitoring for 5 min, stable

Why this works:

Prevents looping — if you've already tried something, the journal shows it. No repeating dead ends.
Captures context — 3 hours later, you won't remember the exact error message. The journal does.
Makes handoffs easy — paste the journal into the incident ticket and someone else can pick it up.
Builds pattern recognition — after 50 entries, you start seeing recurring failure modes.

It sounds too simple to matter. But after 100+ incidents, my journal has caught patterns I wouldn't have noticed otherwise: "third time this month the Redis connection pool exhausts under batch load" — that's not a bug, that's a design problem.

Putting It Together

None of these are revolutionary. They just require doing them consistently:

Practice	Before (time wasted)	After
Structured logging	Hours searching scattered logs	30 seconds with request ID
Reproduce first	2h fixing wrong thing	10 min finding real cause
Binary search	1h checking every layer	15 min of bisecting
Debug journal	Repeating dead ends	Never loops, builds patterns

The hardest one for me was "reproduce first." There's a strong urge to just do something — change code, restart a service, clear a cache. Resist it. Understanding must come before action.

What debugging habits made the biggest difference for you? I'm always looking for new ones.

4 API Design Rules I Learned the Hard Way (With Real Code)

Sir Max — Thu, 06 Aug 2026 02:08:53 +0000

4 API Design Rules I Learned the Hard Way (With Real Code)

I've spent the last few years building and maintaining APIs — both internal microservices and public-facing REST endpoints. Along the way I broke every "best practice" in the book. Each time it cost me — late-night pages, angry users, or embarrassing rollbacks.

Here are the four rules I now enforce on every project, with the real code that could have prevented each disaster.

1. Never Ship Without Versioning — Even Internally

I used to think versioning was for "public APIs" only. Internal services? "We'll just coordinate deployments." This worked for about three months. Then two teams deployed on the same day and suddenly 40% of requests to our billing service were returning 500s.

The mistake: I changed a field from amount (integer, cents) to amount (float, dollars) because "it was cleaner." The consuming team had no warning.

The fix — URL-based versioning from day one:

# fastapi example
from fastapi import FastAPI, APIRouter

app = FastAPI()
v1 = APIRouter(prefix="/api/v1")
v2 = APIRouter(prefix="/api/v2")

@v1.get("/invoices/{id}")
async def get_invoice_v1(id: str):
    return {"amount": 1500}  # cents

@v2.get("/invoices/{id}")
async def get_invoice_v2(id: str):
    return {"amount": 15.00}  # dollars, with deprecation header

app.include_router(v1)
app.include_router(v2)

The /api/v1 endpoint kept running for six months with a Sunset header. Zero breaking changes, zero late-night pages. The cost? One extra directory in your project structure. That's it.

Rule: If another team (or future you) depends on your endpoint, version it. Even if it's "just internal."

2. Don't Over-Engineer Error Responses

Early in my career, I built what I thought was a "comprehensive" error format:

{
  "error": {
    "code": "VALIDATION_ERR_0023",
    "message": "Field 'email' failed regex validation",
    "details": {
      "field": "email",
      "value": "not-an-email",
      "rule": "RFC5322",
      "rule_version": "3.4",
      "suggestion": "Provide a valid email address"
    },
    "request_id": "req_8a7f3c",
    "timestamp": "2024-01-15T08:23:41Z",
    "documentation_url": "/docs/errors#VALIDATION_ERR_0023"
  }
}

Look at that. Eight fields. A documentation_url that went to a 404 page. A rule_version nobody ever used. A suggestion that was literally just the error message rephrased.

The client developers hated it. They just needed the HTTP status code and a human-readable message. Everything else was noise that made their error-handling code longer and harder to debug.

What actually works:

from fastapi import HTTPException

@app.get("/users/{id}")
async def get_user(id: int):
    user = await db.fetch_user(id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail=f"User {id} not found"
        )
    return user

That's it. HTTP status code carries the semantics. detail carries the message. If the client needs structured error codes, add ONE optional field:

{ "detail": "User 42 not found", "code": "not_found" }

Two fields. Maximum. Your client developers will thank you.

Rule: Start with { status_code, detail }. Add fields only when a specific client asks for them — and then add exactly one at a time.

3. Rate Limiting Isn't Optional — It's a Feature

I launched a free tier of an API without rate limiting because "we're small, nobody will abuse it." Within a week, someone's misconfigured cron job was hammering our /search endpoint at 300 requests per second — from a single IP. Our $40/month VPS fell over, and paying customers couldn't access the service.

The fix took 15 minutes with an off-the-shelf library:

import time
from collections import defaultdict

class SimpleRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        now = time.time()
        cutoff = now - self.window
        # Clean old entries
        self.requests[key] = [t for t in self.requests[key] if t > cutoff]
        if len(self.requests[key]) >= self.max_requests:
            return False
        self.requests[key].append(now)
        return True

# Usage
limiter = SimpleRateLimiter(max_requests=100)

@app.get("/search")
async def search(q: str, request: Request):
    client_ip = request.client.host
    if not limiter.is_allowed(client_ip):
        raise HTTPException(status_code=429, detail="Too many requests")
    return await do_search(q)

For production, use something like Redis + a token bucket. But even this 20-line solution would have saved me from that outage. The key insight: rate limiting protects your paying users from free-tier abuse.

Rule: Every endpoint gets a rate limit. Start conservative (100 req/min for free tier), tighten based on real traffic.

4. Deprecation Is a Process, Not an Announcement

The worst API mistake I ever made: I renamed a field from user_name to username, deployed it on a Friday afternoon, and went home. Monday morning, 17 integration tests in a downstream service were failing. The team lead was not happy.

The fix isn't complicated — it just requires patience:

from datetime import datetime, timedelta

@app.get("/api/v2/profile")
async def get_profile():
    profile = {
        "username": "sirmax",
        # Deprecated — remove after 2025-03-01
        "user_name": "sirmax"
    }
    return profile

@app.middleware("http")
async def add_deprecation_header(request: Request, call_next):
    response = await call_next(request)
    if request.url.path.startswith("/api/v1"):
        sunset_date = (datetime.now() + timedelta(days=90)).strftime("%Y-%m-%d")
        response.headers["Sunset"] = sunset_date
        response.headers["Deprecation"] = "true"
    return response

Three steps, in order:

Add the new field, keep the old one. Let clients migrate on their schedule.
Add deprecation headers. Sunset tells them when it's going away. Deprecation: true flags it in logs.
Wait. 90 days minimum. Email your API consumers. Check your logs to see who's still hitting the old endpoints. Then — and only then — remove the old field.

Rule: Never remove a field, endpoint, or behavior without at least one deprecation window. Your "quick rename" is someone else's production incident.

What Stuck With Me

Looking back, every one of these rules boils down to the same thing: assume your API has users you don't know about. The internal tool you built for one team? Another team found it and depends on it. The field you named hastily? Someone's dashboard relies on that exact key.

A good API isn't the one with the cleverest design. It's the one that doesn't break unexpectedly at 3 AM.

What API design rules have you learned the hard way? I'd love to hear your stories in the comments.

3 Async Python Patterns I Wish I Learned Sooner (With Real Code)

Sir Max — Wed, 05 Aug 2026 02:10:22 +0000

3 Async Python Patterns I Wish I Learned Sooner (With Real Code)

I spent two years writing async Python wrong. Not "my code crashed" wrong — more like "I was leaving 70% of the performance on the floor and didn't know it" wrong.

Here are three patterns that actually changed how I write async code. Each one came from a real production problem that forced me to dig deeper than the basic async/await syntax everyone learns on day one.

Pattern 1: `asyncio.gather` — Stop Awaiting Things One at a Time

The mistake I kept making for way too long: chaining await calls sequentially even when they had no reason to wait for each other.

# What I used to write (embarrassing in hindsight):
user = await db.fetch_user(user_id)
orders = await db.fetch_orders(user_id)
notifs = await db.fetch_notifications(user_id)
# Total: fetch_user + fetch_orders + fetch_notifications = ~600ms

Each await blocks the coroutine until that specific call finishes. If each database query takes ~200ms, you're looking at 600ms total. But here's the thing — the database can handle all three queries at once. I was the bottleneck, not the database.

# What I write now:
user, orders, notifs = await asyncio.gather(
    db.fetch_user(user_id),
    db.fetch_orders(user_id),
    db.fetch_notifications(user_id),
)
# Total: max(200ms, 200ms, 200ms) = ~200ms

asyncio.gather schedules all the coroutines at once. It returns only when the slowest one finishes. Three independent I/O calls that took 600ms now take 200ms. That's a 3x speedup from one line change.

Real numbers from a project dashboard endpoint I optimized: 1.2 seconds → 340 milliseconds. Three database queries and one Redis cache lookup all running concurrently instead of one after another. The user noticed — 340ms feels instant, 1.2s feels sluggish.

One gotcha: exceptions inside gather can be tricky. By default, the first exception raised cancels all other tasks and propagates up. If you want to collect all results (including failures), use return_exceptions=True:

results = await asyncio.gather(
    fetch_a(), fetch_b(), fetch_c(),
    return_exceptions=True
)
for i, r in enumerate(results):
    if isinstance(r, Exception):
        logger.warning(f"Source {i} failed: {r}")
    else:
        process(r)

This pattern alone probably saved me more latency than every other optimization combined.

Pattern 2: `asyncio.Semaphore` — The Bouncer Your API Calls Need

I learned this one the hard way. I had a FastAPI endpoint that needed to enrich results by calling a third-party API. Simple enough — fire off 100 concurrent requests, wait for all of them, and return.

What actually happened: the third-party API had a rate limit of 10 requests per second. My endpoint sent all 100 in the first 100 milliseconds. They rate-limited me. My endpoint returned errors. Users were not happy.

Enter asyncio.Semaphore. Think of it as a bouncer at a club — only N people get in at a time:

import asyncio

async def fetch_with_limit(urls: list[str], max_concurrent: int = 5):
    sem = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url):
        async with sem:
            # Only `max_concurrent` coroutines run this block at once
            resp = await http_client.get(url)
            return resp.json()

    tasks = [fetch_one(url) for url in urls]
    return await asyncio.gather(*tasks)

# 100 URLs, never more than 5 concurrent HTTP calls:
results = await fetch_with_limit(all_urls, max_concurrent=5)

The async with sem acquires a slot. If all 5 are taken, the 6th coroutine just pauses there — no error, no polling, no busy-waiting. It resumes automatically when a slot frees up.

I use semaphores everywhere now: database connection pools, file descriptor limits, rate-limited APIs. It's 6 lines of code (the sem = line and the async with sem: block) that prevent entire categories of production incidents.

Pro tip: combine with exponential backoff for extra resilience:

async def fetch_with_retry(url, sem, max_retries=3):
    for attempt in range(max_retries):
        try:
            async with sem:
                return await http_client.get(url)
        except RateLimitError:
            wait = 2 ** attempt  # 1s → 2s → 4s
            await asyncio.sleep(wait)
    raise Exception(f"Failed after {max_retries} retries: {url}")

Pattern 3: `asyncio.Queue` — Backpressure Without the Headache

The hardest bug I've debugged in async Python: memory slowly creeping up over hours until the process got OOM-killed at 2 AM. No crash, no traceback, just a container restart and confused morning investigation.

The culprit was deceptively simple — a list acting as an unbounded buffer between a fast producer and a slow consumer:

# The problem (memory leak in disguise):
results = []
async for item in fast_source():  # Produces 10,000 items/second
    results.append(item)          # List grows without bound

for item in results:              # Consumer processes at 100 items/second
    await slow_process(item)      # 10,000 items sitting in RAM, growing every second

The producer was an event stream that never stopped. The consumer was a database writer that could only handle ~100 inserts per second. The gap was filled by RAM — until RAM ran out.

asyncio.Queue with a maxsize fixes this by applying natural backpressure:

import asyncio

async def pipeline():
    queue = asyncio.Queue(maxsize=50)  # Never more than 50 items in memory

    async def producer():
        async for item in fast_source():
            await queue.put(item)  # BLOCKS if queue is full — backpressure!
        await queue.put(None)      # Sentinel value signals "we're done"

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:       # Sentinel received → exit cleanly
                break
            await slow_process(item)
            queue.task_done()

    await asyncio.gather(producer(), consumer())

The magic is queue.put(item) blocking when the queue is full. The producer automatically slows down to the consumer's speed. Memory usage stays flat at maxsize items. No monitoring alerts, no OOM-kills, no 2 AM wake-up calls.

I now use this pattern for:

Streaming large files through a transform pipeline (read → parse → write)
Batch-inserting database records (consumer collects 100 items, flushes in one INSERT)
Web scraping with download → parse → store stages

The maxsize number matters. Too small and you underutilize the consumer. Too large and you waste memory. My rule of thumb: start with maxsize = consumer_items_per_second * 2 and tune from there.

What I'd Tell My Past Self

Every sequential await is a question. "Does this actually depend on the previous call?" If not → gather it.
Every external API call gets a semaphore. It's 6 lines of code that prevent midnight incidents.
Every pipeline with different speeds gets a queue with maxsize. Backpressure is automatic — you just have to set it up.

None of these are advanced async patterns. They're basic building blocks that the async/await tutorial everyone reads somehow skips. I missed them for two years. Don't make the same mistake.

Found a pattern that changed how you write async code? I'm always looking for the next thing I should have learned two years ago.

Webhooks Are Harder Than You Think — 4 Lessons from Building a Reliable Webhook System

Sir Max — Mon, 03 Aug 2026 02:19:10 +0000

Webhooks Are Harder Than You Think — 4 Lessons from Building a Reliable Webhook System

I used to think webhooks were simple. You get a POST request, you process it, you return 200. Done.

Then I built a system that processed 50,000 webhooks a day. And everything broke.

Here are the four things I wish someone had told me before I started.

Lesson 1: "At-Least-Once" Is a Lie You Need to Handle

Every webhook provider says "at-least-once delivery." What they don't tell you is that "at-least-once" can mean "three times in the same second."

I learned this the hard way when a payment processor retried a payment.succeeded event — three times, all within 800ms. My system processed all three, charged the customer's card for a second subscription, and triggered a very angry support email at 3 AM.

The fix isn't complicated, but you have to do it before anything else:

import hashlib
import redis

redis_client = redis.Redis()

def process_webhook(event_id: str, payload: dict):
    idempotency_key = hashlib.sha256(event_id.encode()).hexdigest()

    # Try to claim this event. If we can't, it's already being processed.
    if not redis_client.set(idempotency_key, "processing", nx=True, ex=300):
        return {"status": "duplicate", "event_id": event_id}

    try:
        # Your actual processing logic here
        handle_payment(payload)
        redis_client.set(idempotency_key, "completed", ex=86400)
        return {"status": "ok"}
    except Exception as e:
        redis_client.delete(idempotency_key)  # Allow retry
        raise

Key points:

Use SET NX (set if not exists) — it's atomic, so no race conditions
Set a TTL (300s here) so a crashed worker doesn't lock the event forever
Delete the key on failure so the event can be retried

We went from 3-5 duplicate processing incidents per week to zero. It's the single highest-ROI change I've ever made to a webhook pipeline.

Lesson 2: Return 200 Fast, Process Later

Webhook providers have short timeouts. Stripe gives you 20 seconds. GitHub gives you 10. If your processing takes longer than that — and it will, once you add database writes, cache updates, and downstream calls — the provider marks it as failed and retries.

Now you're in a death spiral: each retry creates more work, which takes longer, which causes more retries.

The pattern that saved us:

from fastapi import FastAPI, Request, BackgroundTasks
import asyncio
from datetime import datetime

app = FastAPI()
event_queue = asyncio.Queue()

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request, bg: BackgroundTasks):
    payload = await request.json()
    event_id = request.headers.get("stripe-signature", "")

    # Verify signature first (always do this!)
    if not verify_stripe_signature(payload, request.headers):
        return {"status": "invalid_signature"}

    # Acknowledge immediately — don't wait for processing
    bg.add_task(process_in_background, event_id, payload)

    return {"status": "received", "timestamp": datetime.utcnow().isoformat()}

async def process_in_background(event_id: str, payload: dict):
    # Now you can take as long as you need
    await event_queue.put({"id": event_id, "payload": payload, "received_at": datetime.utcnow()})

The provider sees a fast 200. Your system processes at its own pace. Everybody wins.

One caveat: this means you need a retry mechanism on your side. If the background worker crashes after acknowledging the webhook but before finishing, the event is lost. We solved this by writing raw events to a database table first, then processing from the queue.

Lesson 3: Webhook Signatures Are Not Optional

I've seen production systems that skip signature verification because "it's behind a firewall" or "the endpoint URL is secret." Both are wrong.

An unverified webhook endpoint is an open POST endpoint on the internet. Anyone who discovers the URL can inject fake events. And URL discovery is easier than you think — DNS logs, misconfigured CORS headers, leaked in client-side code.

Here's what proper verification looks like for Stripe:

import stripe
import os

stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

def verify_stripe_signature(payload: bytes, headers: dict) -> bool:
    try:
        signature = headers.get("stripe-signature", "")
        stripe.Webhook.construct_event(
            payload=payload,
            sig_header=signature,
            secret=os.getenv("STRIPE_WEBHOOK_SECRET")
        )
        return True
    except stripe.error.SignatureVerificationError:
        return False
    except ValueError:
        return False

For custom webhooks where the provider doesn't give you an SDK, HMAC works fine:

import hmac
import hashlib

def verify_hmac(payload: bytes, signature_header: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header)

Always use hmac.compare_digest() — it's constant-time, which prevents timing attacks.

Lesson 4: Monitor Everything (Or Regret It Later)

The scariest moment of my webhook journey wasn't a crash. It was discovering that a critical webhook had been silently failing for six hours because we didn't have monitoring.

A downstream API went down. Our worker retried, failed, retried again, and eventually gave up. We had no alert. We found out when a customer emailed us.

Here's the minimum monitoring I now set up before deploying any webhook endpoint:

from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

webhook_received = Counter(
    'webhook_received_total',
    'Total webhooks received',
    ['provider', 'event_type']
)
webhook_processed = Counter(
    'webhook_processed_total', 
    'Successfully processed webhooks',
    ['provider', 'event_type']
)
webhook_failures = Counter(
    'webhook_failures_total',
    'Failed webhook processing attempts',
    ['provider', 'event_type', 'error_type']
)
webhook_latency = Histogram(
    'webhook_processing_seconds',
    'Webhook processing duration'
)

@app.post("/webhooks/{provider}")
async def webhook_endpoint(provider: str, request: Request):
    payload = await request.json()
    event_type = payload.get("type", "unknown")

    webhook_received.labels(provider=provider, event_type=event_type).inc()

    try:
        with webhook_latency.time():
            result = await process_event(provider, event_type, payload)
        webhook_processed.labels(provider=provider, event_type=event_type).inc()
        return result
    except Exception as e:
        webhook_failures.labels(
            provider=provider,
            event_type=event_type,
            error_type=type(e).__name__
        ).inc()
        raise

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

Set up an alert on rate(webhook_failures_total[5m]) > 0. You'll sleep better.

What I'd Do Differently Next Time

If I were starting a new webhook system tomorrow, I'd build these four things before writing any business logic:

Idempotency layer (Redis SET NX with TTL)
Async acknowledge pattern (return 200, process in background)
Signature verification (HMAC or provider SDK)
Prometheus metrics + alerts

Everything else — the actual payment processing, the notification sending, the data syncing — comes after.

Webhooks aren't hard because the concept is complex. They're hard because the failure modes are subtle and the consequences are expensive. A duplicate event costs you real money. A dropped event costs you customer trust.

Build the guardrails first. The business logic can wait.

Building reliable API infrastructure. Follow for more lessons from production.

3 API Authentication Methods I Tested in Production — Here's What Actually Broke

Sir Max — Sun, 02 Aug 2026 02:05:39 +0000

3 API Authentication Methods I Tested in Production — Here's What Actually Broke

Last year I built an API gateway that had to support three different authentication methods. I thought it would be straightforward — just pick the standard, right?

Wrong. Each method broke in ways I didn't expect. Here's what I learned, with real code examples so you can avoid the same headaches.

The Setup

I needed to authenticate three types of clients:

End users logging into a web app
Third-party developers calling our API from their servers
Internal microservices talking to each other

The obvious choices: JWT for users, API Keys for third parties, OAuth2 for microservices. Simple.

Then production happened.

1. JWT: The Token That Wouldn't Expire

What Went Wrong

We issued JWTs with a 24-hour expiry. Standard stuff. But I forgot one thing: there's no built-in revocation.

A user reported their account was compromised. We changed their password, expecting their old tokens to stop working. They didn't. The JWT was still valid — it's just a signed blob of data, completely independent from the database.

The Fix

We added a token blacklist in Redis:

import redis
import jwt
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, db=0)

def is_token_revoked(jti: str) -> bool:
    """Check if a JWT ID has been revoked."""
    return r.exists(f"revoked:{jti}") > 0

def revoke_token(jti: str, exp: int):
    """Add token to blacklist until it naturally expires."""
    ttl = exp - int(datetime.utcnow().timestamp())
    if ttl > 0:
        r.setex(f"revoked:{jti}", ttl, "1")

# Middleware check on every request
def verify_token(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    if is_token_revoked(payload["jti"]):
        raise InvalidTokenError("Token has been revoked")
    return payload

The TTL trick is important — don't store revoked tokens forever. Set the Redis key to expire at the same time the JWT expires. No memory leak, no stale entries.

What I Wish Someone Told Me

JWTs are stateless, but revocation makes them stateful. Factor that into your architecture from day one.
Keep payloads small. Every extra field bloats every HTTP request. Put user roles in the JWT, but fetch permissions from the database.
Use short-lived access tokens (15-30 min) with refresh tokens. This limits the damage window if a token leaks.

2. API Keys: The Invisible Leak

What Went Wrong

This one was embarrassing. A developer accidentally committed their API key to a public GitHub repo. By the time we noticed, someone had racked up significant unauthorized API usage over a weekend.

The Fix: Three Layers

Layer 1 — Rate Limits Per Key

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self):
        self._windows = defaultdict(list)

    def is_allowed(self, api_key: str, max_requests: int = 100, window: int = 60) -> bool:
        now = time.time()
        cutoff = now - window

        self._windows[api_key] = [t for t in self._windows[api_key] if t > cutoff]

        if len(self._windows[api_key]) >= max_requests:
            return False

        self._windows[api_key].append(now)
        return True

Layer 2 — Usage Alerts

We set up a webhook that fires when a key exceeds 80% of its daily quota:

def check_usage_alert(api_key: str, usage: int, daily_limit: int = 10000):
    ratio = usage / daily_limit
    if 0.8 <= ratio < 1.0:
        send_alert(f"Key ending in ...{api_key[-4:]} at {ratio:.0%} of daily limit")
    elif usage >= daily_limit:
        deactivate_key(api_key)
        send_alert(f"Key EXCEEDED limit — DEACTIVATED")

Layer 3 — Key Rotation API

Let developers rotate their own keys without emailing you:

@app.post("/keys/{key_id}/rotate")
def rotate_api_key(key_id: str, user=Depends(get_current_user)):
    old_key = db.get_key(key_id)
    if old_key.owner_id != user.id:
        raise HTTPException(403)

    new_key = secrets.token_urlsafe(32)
    db.update_key(key_id, new_key)
    db.revoke_key(old_key.value)

    return {"new_key": new_key, "old_key_expires": "24h"}

What I Wish Someone Told Me

Never log API keys. Configure your logging framework to redact them automatically.
Prefix your keys (like sk_live_xxx) so you can grep for leaks. GitHub's secret scanning also works better with recognizable patterns.
Don't use a single master key. Issue per-developer keys with different rate limits.

3. OAuth2 for Microservices: Overkill That Killed Performance

What Went Wrong

I used OAuth2 Client Credentials flow for internal service-to-service auth. Every call required a token exchange round-trip:

Service A → Auth Server: "Give me a token"
Auth Server → Service A: "Here's your token (valid 1 hour)"
Service A → Service B: "Here's my request + token"
Service B → Auth Server: "Is this token valid?"
Auth Server → Service B: "Yes"
Service B → Service A: "Here's your data"

Four network calls for one request. Latency jumped from 12ms to 85ms.

The Fix: mTLS + Signed JWTs

For internal traffic that never leaves the VPC, we replaced OAuth2 with mutual TLS:

# Service A — make a signed request
import httpx
import jwt
import time
import secrets

def call_service_b(endpoint: str, data: dict):
    payload = {
        "iss": "service-a",
        "aud": "service-b",
        "iat": int(time.time()),
        "exp": int(time.time()) + 300,
        "jti": secrets.token_hex(16)
    }
    token = jwt.encode(payload, SERVICE_A_PRIVATE_KEY, algorithm="RS256")

    response = httpx.post(
        f"https://service-b.internal/{endpoint}",
        json=data,
        headers={"Authorization": f"Bearer {token}"},
        cert=("/etc/ssl/service-a.crt", "/etc/ssl/service-a.key"),
        verify="/etc/ssl/ca.crt"
    )
    return response.json()

# Service B — verify with public key (no auth server call needed)
@app.post("/api/data")
def handle_request(request: Request):
    token = request.headers["Authorization"].split(" ")[1]
    try:
        payload = jwt.decode(
            token, 
            SERVICE_A_PUBLIC_KEY,
            algorithms=["RS256"],
            options={"require": ["iss", "exp"]}
        )
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    # Process request...

Latency dropped back to 14ms. The mTLS certs handle transport security, and the short-lived signed JWTs handle identity — no auth server needed.

What I Wish Someone Told Me

mTLS + signed JWTs is the sweet spot for internal services. Certificates authenticate the service, JWTs authenticate the request.
OAuth2 is great for third-party integrations, overkill for your own services. Don't use it internally unless you have a specific compliance requirement.
Measure the overhead before committing. A quick benchmark on your proposed auth flow will save you weeks of "why is everything slow?"

The Decision Matrix

Here's how I now decide which auth method to use:

Scenario	Recommendation	Why
User-facing web app	JWT + Refresh Tokens	Stateless, scalable, easy to invalidate with short TTL
Third-party API access	API Keys with HMAC	Easy for developers, scoped permissions, monitorable
Internal microservices	mTLS + Signed JWTs	Zero-auth-server latency, transport security built in
OAuth2 social login	OAuth2 (Authorization Code)	Industry standard, users already have Google/GitHub accounts
Mobile app	JWT + Refresh + Device Binding	Resist token theft with device fingerprint

The Rule I Live By Now

The right auth method is the simplest one that meets your threat model.

Don't reach for OAuth2 because Google uses it. Don't use JWTs because they're trendy. Ask yourself:

Who is calling my API? (Users? Servers? Both?)
What happens if a credential leaks? (Financial loss? Data breach? Nothing?)
Do I need to revoke access instantly? (If yes, JWT alone won't cut it)
How many auth calls per second can I afford? (Every round-trip adds latency)

Answer those four questions honestly, and the right choice becomes obvious.

What auth horror stories do you have? I'd love to hear them in the comments — especially if you've dealt with SAML. I still have scars from that one.

The Right Way to Handle API Rate Limits — A Practical Guide with Python Code

Sir Max — Sat, 01 Aug 2026 02:09:28 +0000

The Right Way to Handle API Rate Limits — A Practical Guide with Python Code

A few years ago I was building an integration with a payment API. Everything worked fine in testing. Then Black Friday hit.

Our backend started making calls as fast as it could. The API responded with 429 Too Many Requests. Our code, in its infinite wisdom, saw the error and retried immediately — adding even more pressure. Ten minutes later, we were rate-limited for the rest of the day. We lost thousands in transactions.

I learned the hard way that retry logic is not just about retrying. It's about how and when you retry. Here's everything I wish someone had told me before that day.

The Naive Approach (Don't Do This)

import requests

def call_api_naive(url, headers):
    for attempt in range(5):
        resp = requests.get(url, headers=headers)
        if resp.status_code == 429:
            continue  # try again immediately — BAD
        return resp.json()
    raise Exception("API failed after 5 attempts")

This makes everything worse. Every retry hits the API at the same moment, creating a tight loop of 429s. You're not solving the problem — you're part of it.

The Three Rules of Good Retry Logic

Rule 1: Exponential Backoff

Don't retry immediately. Wait. Then wait longer. Then even longer.

The math is simple: delay = base_delay * (2 ** attempt). With a 1-second base and 5 attempts, your delays look like this:

Attempt 0: instant
Attempt 1: wait 1 second
Attempt 2: wait 2 seconds
Attempt 3: wait 4 seconds
Attempt 4: wait 8 seconds

import time

def call_with_backoff(url, headers, max_retries=5, base_delay=1):
    for attempt in range(max_retries):
        resp = requests.get(url, headers=headers)
        if resp.status_code != 429:
            return resp.json()

        delay = base_delay * (2 ** attempt)
        print(f"Rate limited. Waiting {delay}s before retry {attempt + 1}")
        time.sleep(delay)

    raise Exception(f"Still rate-limited after {max_retries} attempts")

This alone would have saved my Black Friday. But there's more.

Rule 2: Add Jitter

If you have 100 servers all hitting the same API, and they all use exponential backoff with the same base delay — they're still in sync. They all retry at exactly the same moment.

The fix is jitter: add a random component to the delay.

import random

delay = base_delay * (2 ** attempt) + random.uniform(0, 1)

Now your 100 servers spread their retries across a 1-second window instead of colliding at the exact same millisecond. The API sees a smooth flow of requests instead of spikes.

Rule 3: Respect the `Retry-After` Header

Most APIs that return 429 also send a Retry-After header telling you exactly how long to wait. Ignoring it is like ignoring a traffic cop.

def call_api_smart(url, headers, max_retries=5, base_delay=1):
    for attempt in range(max_retries):
        resp = requests.get(url, headers=headers)

        if resp.status_code == 429:
            # Try to read Retry-After header
            retry_after = resp.headers.get("Retry-After")
            if retry_after and retry_after.isdigit():
                delay = int(retry_after)
            else:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)

            print(f"429 received. Waiting {delay:.1f}s (attempt {attempt + 1}/{max_retries})")
            time.sleep(delay)
            continue

        resp.raise_for_status()
        return resp.json()

    raise Exception(f"API unavailable after {max_retries} attempts")

Putting It All Together

Here's a reusable RateLimitHandler class I've been using in production:

import time
import random
import requests
from typing import Optional, Dict, Any

class RateLimitHandler:
    def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    def call(self, url: str, headers: Dict[str, str], 
             method: str = "GET", json_data: Optional[Dict] = None) -> Dict[str, Any]:

        for attempt in range(self.max_retries):
            resp = requests.request(method, url, headers=headers, json=json_data)

            if resp.status_code == 429:
                delay = self._get_delay(resp, attempt)
                print(f"[429] Backing off {delay:.1f}s (attempt {attempt + 1})")
                time.sleep(delay)
                continue

            if resp.status_code >= 500:
                delay = self._get_delay(resp, attempt)
                print(f"[5xx] Server error. Retrying in {delay:.1f}s")
                time.sleep(delay)
                continue

            resp.raise_for_status()
            return resp.json()

        raise Exception(f"All {self.max_retries} retries exhausted")

    def _get_delay(self, resp, attempt: int) -> float:
        retry_after = resp.headers.get("Retry-After")
        if retry_after and retry_after.isdigit():
            return float(retry_after)
        return self.base_delay * (2 ** attempt) + random.uniform(0, 1)

# Usage
handler = RateLimitHandler(max_retries=5, base_delay=1.0)
data = handler.call("https://api.example.com/v1/data", 
                     headers={"Authorization": "Bearer YOUR_TOKEN"})

What I Learned the Hard Way

Never retry without a delay. Immediate retries compound the problem.
Jitter is not optional at scale. Without it, thundering herds still happen.
Retry-After is a gift. Use it. The API is telling you what it needs.
Log everything. When something breaks at 3 AM, you'll want to see the backoff pattern.
Have a circuit breaker. If the API keeps returning 429s for 30+ seconds, stop trying. Return a degraded response or cached data. Don't take down your own service.

The Black Friday disaster was avoidable. A few dozen lines of code would have kept us running through the traffic spike. If you're calling any third-party API — payment, AI, social media, anything — take 10 minutes to add proper retry logic. Your future self will thank you.

What's your approach to handling rate limits? I'd love to hear what patterns you use in the comments.

The Hidden Cost of Not Handling API Errors Properly (And How to Fix It in 5 Steps)

Sir Max — Fri, 31 Jul 2026 02:12:46 +0000

The Hidden Cost of Not Handling API Errors Properly (And How to Fix It in 5 Steps)

I once spent a Friday evening debugging a production outage that had nothing to do with bad code or infrastructure failure. The culprit? A third-party weather API that decided to return HTTP 500 for three hours straight. Our service didn't handle it. It just... crashed.

That night cost us five hours of sleep and a weekend's worth of customer complaints. The fix took 20 minutes. The error handling? We never thought we needed it — until we desperately did.

Here's what I learned, and a 5-step approach you can implement today.

Why Most Error Handling Falls Short

The default approach in most codebases looks something like this:

import requests

def get_weather(city):
    resp = requests.get(f"https://api.weather.com/v1/{city}")
    return resp.json()

This works perfectly — until it doesn't. The API goes down, the network flakes, the rate limit kicks in, and suddenly your entire request chain unravels with a cryptic traceback.

What's worse is that the way an API fails tells you what to do next. A 429 means "slow down." A 503 means "try again later." A 401 means "your token expired." But if you treat them all the same way, you're flying blind.

Step 1: Map the Error Landscape

Before writing a single line, categorise every possible failure mode:

Category	Examples	Action
Client error (4xx)	400, 401, 403, 404, 429	Fix the request or back off
Server error (5xx)	500, 502, 503, 504	Retry with exponential backoff
Network error	Timeout, DNS failure, connection refused	Retry or fail gracefully
Data error	Malformed JSON, missing fields, type mismatch	Log and degrade

This isn't theoretical. Every API your service depends on will hit every one of these eventually. The question is whether your code is ready when it happens.

Step 2: Build a Retry Layer That Actually Works

Naive retry looks like this:

for _ in range(3):
    try:
        return requests.get(url)
    except:
        time.sleep(1)

The problem? If the upstream is genuinely down, you're hammering it with retries at exactly the worst moment. A thundering herd of retries can turn a brief hiccup into a cascading outage.

Instead, use exponential backoff with jitter:

import time
import random

def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code < 500:
                resp.raise_for_status()
                return resp
            # Server error — retry
        except (requests.Timeout, requests.ConnectionError):
            pass  # Network error — retry

        if attempt < max_retries - 1:
            sleep_time = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(sleep_time)

    raise Exception(f"Failed after {max_retries} attempts: {url}")

The jitter (random component) prevents the thundering herd. Two services retrying at 1s, 2s, 4s in lockstep will collide again. Adding ±0.5s jitter spreads them out.

Step 3: Handle Status Codes Individually

A blanket raise_for_status() is better than nothing, but it treats 404 and 429 the same way. They are not the same.

class APIError(Exception):
    def __init__(self, status_code, message, retry_after=None):
        self.status_code = status_code
        self.message = message
        self.retry_after = retry_after

def handle_response(resp):
    if resp.status_code == 429:
        retry_after = resp.headers.get("Retry-After", "60")
        raise APIError(429, "Rate limited", retry_after=int(retry_after))

    if resp.status_code == 401:
        raise APIError(401, "Authentication failed — refresh your token")

    if resp.status_code == 403:
        raise APIError(403, "Forbidden — check your permissions")

    if resp.status_code >= 500:
        raise APIError(resp.status_code, f"Upstream server error")

    resp.raise_for_status()
    return resp

Now your retry layer can be smart: respect Retry-After headers for 429s, retry on 5xx only, and bail immediately on 4xx client errors (except 429).

Step 4: Degrade Gracefully, Don't Crash

Not every API call is mission-critical. If the weather widget fails, users should still see their dashboard — just without the weather.

def get_dashboard_data(user_id):
    data = {
        "user": fetch_user(user_id),
        "orders": fetch_orders(user_id),
    }

    try:
        data["weather"] = fetch_weather(user_id)
    except Exception:
        data["weather"] = {"status": "unavailable"}
        # Log it, but don't block the entire response

    return data

The key insight: degrade by feature, not by the whole page. Users won't notice a missing weather widget. They will absolutely notice a blank page.

Step 5: Monitor What Breaks

Error handling without monitoring is like a smoke detector with no battery. You need to see the failures to fix them.

At minimum, track these three things:

Error rate by endpoint and status code — so you know which upstream is flaking
Retry success rate — if retries are consistently failing, the backoff strategy needs tuning
P95 latency of downstream calls — degradation often starts as slowness, not errors

A simple logging approach:

import logging
import time

logger = logging.getLogger("api-client")

def instrumented_call(fn, name, *args, **kwargs):
    start = time.monotonic()
    try:
        result = fn(*args, **kwargs)
        elapsed = time.monotonic() - start
        logger.info("api_call", extra={
            "name": name, "status": "ok", "elapsed_ms": round(elapsed * 1000)
        })
        return result
    except APIError as e:
        elapsed = time.monotonic() - start
        logger.error("api_call", extra={
            "name": name, "status": e.status_code, "elapsed_ms": round(elapsed * 1000)
        })
        raise

Feed these logs into your existing monitoring stack (Datadog, Grafana, whatever you use). Set alerts on error rate spikes. You'll catch problems before your users do.

What I Wish I'd Known Earlier

The biggest lesson from that Friday night outage wasn't technical. It was this: error handling is not a nice-to-have. It's not "we'll add it later." It's as fundamental as the happy path itself.

Every external API — every single one — will fail at some point. Your customers don't care whose fault it is. They just see a broken page.

The five steps above aren't groundbreaking. They're battle-tested patterns that take maybe an afternoon to implement. That afternoon buys you sleep on the nights when things inevitably go wrong.

Start with Step 1 today. Map your external dependencies. For each one, ask: "What happens when this breaks?" If the answer is "I don't know," you've found your next task.

Have a different approach to API error handling? I'd love to hear it in the comments.

How I Cut My API Costs by 60% With a 50-Line Python Cache

Sir Max — Thu, 30 Jul 2026 02:19:21 +0000

How I Cut My API Costs by 60% With a 50-Line Python Cache

Last month I got a bill from an AI API provider that made me choke on my coffee. $340. For a side project that maybe 50 people use.

The culprit? I was calling the same endpoints with the same parameters over and over. Every user refresh hit the API. Every page reload. Every test run.

I checked the logs and found something embarrassing: 73% of my API calls were identical duplicates — same endpoint, same payload, same response.

That's when I built this. It took 30 minutes and I haven't touched it since. Here's the code, why it works, and the one edge case that almost broke everything.

The Problem, Quantified

Before I show you the code, here's what my API usage looked like:

Metric	Before	After
Daily API calls	8,400	2,300
Duplicate call rate	73%	12%
Monthly bill	$340	$135
Avg response time	1.2s	80ms

The 12% that still slip through are dynamic queries — things like "what's the weather right now" where caching makes no sense. Everything else gets served from disk.

The Code

Here's the full thing. 50 lines of Python, no dependencies outside the standard library:

import hashlib
import json
import os
import time
from functools import wraps

CACHE_DIR = "./api_cache"
DEFAULT_TTL = 3600  # 1 hour


def _make_key(method: str, url: str, params: dict, body: dict) -> str:
    """Create a deterministic cache key from request components."""
    payload = json.dumps({
        "m": method,
        "u": url,
        "p": sorted(params.items()) if params else [],
        "b": sorted(body.items()) if body else [],
    }, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()


def _path(key: str) -> str:
    return os.path.join(CACHE_DIR, key[:2], key)


def api_cache(ttl: int = DEFAULT_TTL):
    """Cache API responses to disk with configurable TTL."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Extract identifiable request components
            # Adjust these based on how YOUR API client works
            req = kwargs.get("request_data", {})
            key = _make_key(
                method=req.get("method", "GET"),
                url=req.get("url", func.__name__),
                params=req.get("params", {}),
                body=req.get("body", {}),
            )

            cache_path = _path(key)

            # Cache hit — check freshness
            if os.path.exists(cache_path):
                age = time.time() - os.path.getmtime(cache_path)
                if age < ttl:
                    with open(cache_path, "r") as f:
                        return json.load(f)

            # Cache miss — call the real API
            result = func(*args, **kwargs)

            # Don't cache errors. This is the part I got wrong the first time.
            if result and not isinstance(result, Exception):
                os.makedirs(os.path.dirname(cache_path), exist_ok=True)
                with open(cache_path, "w") as f:
                    json.dump(result, f)

            return result

        return wrapper
    return decorator

How to Use It

Wrap your API call function and pass in request metadata:

import requests

@api_cache(ttl=1800)  # 30 minutes
def call_openai(prompt: str, request_data: dict = None):
    resp = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['OPENAI_KEY']}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}],
        },
    )
    return resp.json()


# Call it — first time hits the API, subsequent calls in the
# next 30 minutes return from cache
result = call_openai(
    "summarize this text: ...",
    request_data={
        "method": "POST",
        "url": "openai/chat/completions",
        "params": {},
        "body": {"prompt": "summarize this text: ..."},
    },
)

For more complex APIs, include the relevant distinguishing fields in request_data.body. The hash function handles the rest.

The Edge Case I Got Wrong

On day one, I cached everything. Including errors.

My AI provider went down for 8 minutes. During that window, my cache stored 429 responses — HTTP errors — and served them to every user for the full TTL. Users saw "rate limit exceeded" for things that normally worked fine.

That's why there's this check in the code:

if result and not isinstance(result, Exception):
    os.makedirs(os.path.dirname(cache_path), exist_ok=True)
    ...

Two rules I now follow:

Never cache 4xx or 5xx responses.
Never cache empty responses. They hide real outages.

If you're wrapping requests, here's a quick filter you can add:

def _is_cacheable(response) -> bool:
    if response is None:
        return False
    if isinstance(response, dict) and "error" in response:
        return False
    # If using requests.Response object
    if hasattr(response, "status_code") and response.status_code >= 400:
        return False
    return True

When NOT to Use This

This works well for:

AI text generation with deterministic prompts (same prompt = same response)
REST API read endpoints (GET /users/42)
Configuration fetches
Anything where staleness is acceptable for a few minutes

It does not work for:

Real-time data (stock prices, live sports scores)
Write operations (POST that creates resources)
User-specific data unless you include user ID in the cache key
Endpoints with rapidly changing state

I learned that last one the hard way when I cached a "current server status" endpoint and spent an afternoon debugging why my dashboard showed "healthy" for a server that had been down for two hours.

Going Further: Memory + Disk

The version above uses disk only. That's fine for hundreds of requests per minute. If you need more speed, add an in-memory LRU layer first:

from collections import OrderedDict

_mem_cache: OrderedDict[str, tuple[float, dict]] = OrderedDict()
MAX_MEM_ITEMS = 500

def _cache_get(key: str, ttl: int) -> dict | None:
    # Check memory first
    if key in _mem_cache:
        ts, val = _mem_cache[key]
        if time.time() - ts < ttl:
            _mem_cache.move_to_end(key)
            return val
        del _mem_cache[key]

    # Fall back to disk
    cache_path = _path(key)
    if os.path.exists(cache_path):
        age = time.time() - os.path.getmtime(cache_path)
        if age < ttl:
            with open(cache_path) as f:
                val = json.load(f)
            _mem_cache[key] = (time.time(), val)
            if len(_mem_cache) > MAX_MEM_ITEMS:
                _mem_cache.popitem(last=False)
            return val

    return None

This adds about 15 lines and makes cache hits virtually instant. For my use case the disk-only version was enough, but if you're serving web traffic, the memory layer is worth it.

What I'd Do Differently

If I were building this again from scratch:

Use SQLite instead of files. Better concurrent access, built-in TTL with a timestamp column, and one file instead of thousands.
Add cache statistics. Hit rate, miss rate, total bytes saved. I ended up adding this later anyway to justify the build to myself.
Make TTL configurable per endpoint. Some endpoints are fine at 1 hour, others need 5 minutes.

But honestly? The 50-line version has been running untouched for a month and saved me $205. Sometimes the simple thing is good enough.

Have you built something similar? What's your approach to API cost control? I'd love to hear about it in the comments.

I Spent 3 Years Building APIs — Here Are 6 Things Nobody Tells You

Sir Max — Wed, 29 Jul 2026 02:08:36 +0000

I Spent 3 Years Building APIs — Here Are 6 Things Nobody Tells You

I've been building and maintaining REST APIs professionally for the past few years. Not the kind you build in a weekend hackathon — the kind that serve real users, at real scale, with real consequences when they break.

Here are six lessons I learned the hard way. Some of them cost me sleep at 3 AM. Others cost me users. All of them made me a better engineer.

1. Your API Is a Product, Not a Backend

The biggest mindset shift I had wasn't technical — it was realizing that an API is a product with its own users, its own UX, and its own support burden.

When a frontend developer integrates your API, they're your customer. They don't care about your clean architecture or your elegant ORM queries. They care about three things:

Does it work as documented?
Is it fast enough?
When it breaks, do I get a useful error message?

I once spent two weeks optimizing our database queries, only to discover that the real bottleneck was a missing index on a JSONB column — something I would have caught in five minutes if I'd looked at the slow query log first. The frontend team didn't notice my query optimization. They did notice the 500 errors when that index was missing.

What I do now: Treat every endpoint like a feature. Before shipping, ask: "Would I be happy if I had to integrate this?"

2. Documentation Is Your First Line of Defense

Early in my career, I thought documentation was something you wrote after the code was done. I was wrong. Documentation is something you write before the code — it's a design tool.

Here's why: writing the docs first forces you to think about the user's perspective. What parameters do they actually need? What should the response look like? What errors should they expect?

# Bad: "We'll figure out the contract later"
@app.post("/orders")
async def create_order(request: Request):
    data = await request.json()
    # ... 200 lines of logic ...
    return {"status": "ok"}

# Better: The contract is clear from the start
@app.post("/orders")
async def create_order(
    items: list[OrderItem],
    shipping_address: Address,
    payment_method_id: str,
) -> OrderResponse:
    """
    Create a new order. Returns 201 on success, 422 on validation error.

    Rate limit: 10 requests per minute per customer.
    """
    ...

I now write the OpenAPI spec before I write a single line of implementation code. It catches design problems early — if I can't explain the endpoint clearly in the spec, I don't understand it well enough to build it.

3. Rate Limiting Isn't Optional — It's Infrastructure

In my second year, someone wrote a script that accidentally hammered our search endpoint at 200 requests per second. The database didn't crash, but every other API consumer experienced 5-second timeouts for the next ten minutes.

Here's what I learned: rate limiting isn't about protecting against malicious actors. It's about protecting your well-intentioned users from each other.

A simple in-memory rate limiter in Python:

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        now = time.time()
        cutoff = now - self.window
        self.requests[key] = [t for t in self.requests[key] if t > cutoff]
        if len(self.requests[key]) >= self.max_requests:
            return False
        self.requests[key].append(now)
        return True

This is fine for a single process, but in production you'll want Redis with sorted sets for atomicity. The key principle is the same: always return a clear Retry-After header so the client knows when to come back.

# Good error response
{
    "error": "rate_limit_exceeded",
    "message": "Too many requests. Try again in 12 seconds.",
    "retry_after": 12
}

A rate-limited client that keeps retrying is a friendly client. A rate-limited client that has no idea when to retry is an angry one.

4. Error Messages Are User Experience

Nothing frustrates an API consumer more than this:

{"error": "Internal Server Error"}

That's not an error message. That's a shrug in JSON format.

Every error response should answer three questions:

What went wrong? — Be specific: "Field 'email' must be a valid email address", not "Validation error"
Whose fault is it? — Is it the client's input (4xx) or the server's state (5xx)?
What should they do next? — Should they fix their request, retry later, or contact support?

Here's a pattern I use now:

class APIError(Exception):
    def __init__(self, status_code: int, code: str, message: str, details: dict | None = None):
        self.status_code = status_code
        self.body = {
            "error": code,
            "message": message,
            "details": details or {}
        }

# Usage
raise APIError(
    400,
    "invalid_payment_method",
    "Payment method 'card_abc123' has expired.",
    {"expired_at": "2026-01-15", "suggested_action": "update_payment_method"}
)

The code field is machine-readable (great for SDKs). The message is human-readable. The details give actionable context. This pattern has saved me more support tickets than any other single change.

5. Versioning Is a Social Contract

I've seen teams argue for weeks about API versioning strategies — URL-based (/v1/, /v2/), header-based (Accept: application/vnd.api+json;version=2), query parameter-based. The strategy matters less than the commitment.

When you publish a /v1/users endpoint, you're making a promise: this will keep working the same way until we give you a clear migration path to v2.

Breaking that promise — even for a "minor" change — erodes trust. I once changed a field name from created_at to created because "it's cleaner." A production integration broke because someone was deserializing by field name. It took two hours to roll back. I never made that mistake again.

My rule: If a change might break any existing consumer, it goes in a new version. Period. Even if it's "just" renaming a field or changing a default value.

And document your deprecation policy:

v1 endpoints: supported until 2026-12-31
v2 endpoints: current stable, guaranteed through 2027-06-30

6. Observability Before Optimization

I've lost count of how many times I've seen engineers (including myself) optimize something without measuring it first.

Before you add a caching layer, ask: what's the p95 latency of this endpoint right now? Before you denormalize a database table, ask: which query is actually slow and by how much?

Here's the minimum observability stack I'd recommend for any production API:

import time
import logging
from functools import wraps

logger = logging.getLogger("api.metrics")

def instrument(endpoint: str):
    """Decorator that logs request duration and status."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.monotonic()
            try:
                result = await func(*args, **kwargs)
                duration_ms = (time.monotonic() - start) * 1000
                logger.info(f"endpoint={endpoint} status=ok duration_ms={duration_ms:.1f}")
                return result
            except Exception as e:
                duration_ms = (time.monotonic() - start) * 1000
                logger.error(f"endpoint={endpoint} status=error duration_ms={duration_ms:.1f} error={type(e).__name__}")
                raise
        return wrapper
    return decorator

This is dead simple, but with it you can answer questions like:

Which endpoints are slowest at p95?
Is there a spike in errors at a specific time?
Did that last deploy make things better or worse?

Optimize what you measure. Measure what matters.

The Thread That Runs Through All Six

Looking back, every one of these lessons traces back to the same root: respect the person on the other side of the API.

They're trying to build something. They're on a deadline. They're debugging at midnight just like you. Every design decision you make either helps them or hurts them.

Build APIs like you're the one who has to integrate them. Because someday, you will be.

What's the hardest API lesson you've learned? I'd love to hear about it in the comments.

DEV Community: Sir Max

4 Docker Compose Patterns That Saved Our Team Hours of Debugging

4 Docker Compose Patterns That Saved Our Team Hours of Debugging

1. Healthchecks That Actually Work

2. Profiles: One File, Two Environments

3. extends for Dry Compose Files

4. Wait-For-It Without Custom Scripts

What These Patterns Share

3 Things I Wish I Knew Before Building with LLM APIs

3 Things I Wish I Knew Before Building with LLM APIs

1. Token counting is not optional — it's the whole job

2. Streaming is table stakes, but structured output is the hard part

3. Fallback is not a feature — it's architecture

The takeaway

3 Database Query Patterns That Kill Performance (And How to Fix Them)

3 Database Query Patterns That Kill Performance (And How to Fix Them)

1. The N+1 Problem: Your ORM Is Lying to You

2. LIKE '%wildcard%' Searches: The Index Killer

3. Missing Composite Indexes: One Column at a Time Won't Cut It

What I Actually Do Now

How I Cut Debugging Time in Half: 4 Practices That Actually Work

How I Cut Debugging Time in Half: 4 Practices That Actually Work

1. Structured Logging Before You Need It

2. Reproduce First, Fix Second

3. Binary Search Debugging

4. The Debug Journal

Putting It Together

4 API Design Rules I Learned the Hard Way (With Real Code)

4 API Design Rules I Learned the Hard Way (With Real Code)

1. Never Ship Without Versioning — Even Internally

2. Don't Over-Engineer Error Responses

3. Rate Limiting Isn't Optional — It's a Feature

4. Deprecation Is a Process, Not an Announcement

What Stuck With Me

3 Async Python Patterns I Wish I Learned Sooner (With Real Code)

3 Async Python Patterns I Wish I Learned Sooner (With Real Code)

Pattern 1: asyncio.gather — Stop Awaiting Things One at a Time

Pattern 2: asyncio.Semaphore — The Bouncer Your API Calls Need

Pattern 3: asyncio.Queue — Backpressure Without the Headache

What I'd Tell My Past Self

Webhooks Are Harder Than You Think — 4 Lessons from Building a Reliable Webhook System

Webhooks Are Harder Than You Think — 4 Lessons from Building a Reliable Webhook System

Lesson 1: "At-Least-Once" Is a Lie You Need to Handle

Lesson 2: Return 200 Fast, Process Later

Lesson 3: Webhook Signatures Are Not Optional

Lesson 4: Monitor Everything (Or Regret It Later)

What I'd Do Differently Next Time

3 API Authentication Methods I Tested in Production — Here's What Actually Broke

3 API Authentication Methods I Tested in Production — Here's What Actually Broke

The Setup

1. JWT: The Token That Wouldn't Expire

What Went Wrong

The Fix

What I Wish Someone Told Me

2. API Keys: The Invisible Leak

What Went Wrong

The Fix: Three Layers

What I Wish Someone Told Me

3. OAuth2 for Microservices: Overkill That Killed Performance

What Went Wrong

The Fix: mTLS + Signed JWTs

What I Wish Someone Told Me

The Decision Matrix

The Rule I Live By Now

The Right Way to Handle API Rate Limits — A Practical Guide with Python Code

The Right Way to Handle API Rate Limits — A Practical Guide with Python Code

The Naive Approach (Don't Do This)

The Three Rules of Good Retry Logic

Rule 1: Exponential Backoff

Rule 2: Add Jitter

Rule 3: Respect the Retry-After Header

Putting It All Together

What I Learned the Hard Way

The Hidden Cost of Not Handling API Errors Properly (And How to Fix It in 5 Steps)

The Hidden Cost of Not Handling API Errors Properly (And How to Fix It in 5 Steps)

Why Most Error Handling Falls Short

Step 1: Map the Error Landscape

Step 2: Build a Retry Layer That Actually Works

Step 3: Handle Status Codes Individually

Step 4: Degrade Gracefully, Don't Crash

3. `extends` for Dry Compose Files

Pattern 1: `asyncio.gather` — Stop Awaiting Things One at a Time

Pattern 2: `asyncio.Semaphore` — The Bouncer Your API Calls Need

Pattern 3: `asyncio.Queue` — Backpressure Without the Headache

Rule 3: Respect the `Retry-After` Header