DEV Community: Adilet Akmatov

FastAPI Security 100 LVL: Production-Grade DDoS Protection

Adilet Akmatov — Mon, 06 Apr 2026 11:57:02 +0000

Your FastAPI is going down under load — and you're reading the slowapi docs thinking that'll be enough?

Spoiler: it won't. About as effective as a bike lock against a professional thief.

Let's break it all down — layer by layer — the way senior security engineers do it in production. No fluff, real code, some humor, and a clear understanding of why each decision is made the way it is — not just "I saw someone do it this way once."

Why FastAPI Is Especially Vulnerable

Python applications aren't killed by traffic volume — they're killed by application-layer (L7) attacks — ones that look like perfectly normal HTTP requests. A single request to a heavy endpoint can eat 5 seconds of CPU. 200 of those simultaneously = your server is dead, and you're getting a 3 AM phone call.

Attack types you need to know:

Type	Layer	What it kills	Where to handle
UDP/ICMP flood	L3	Network channel	CDN / provider
SYN flood	L4	TCP stack	iptables / Nginx
HTTP flood	L7	Workers	FastAPI + Nginx
Slowloris	L7	Connections	Nginx + Uvicorn
Heavy endpoint abuse	L7	CPU / memory	Rate Limiting

Defense in Depth

Internet → CDN/DDoS Provider → Nginx → Rate Limit Middleware → FastAPI → Business Logic

Rule: each layer handles its own threat type and reduces load on the next one. Like an onion — except instead of tears, the attacker gets a 403.

⚠️ The #1 beginner mistake — protecting only at the FastAPI level without Nginx in front. Thousands of simultaneous connections will spin up Python processes and exhaust memory before any middleware gets a chance to run. It's like posting a security guard inside the building while a crowd is already breaking through the unlocked front door.

1. Rate Limiting — First Line of Defense

Install

pip install slowapi redis limits

Yes, three packages. But at least it won't hurt later.

Basic Configuration

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware

# key_func defines what we're limiting by
limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["100/minute"],
    # Redis for distributed counter — required with multiple workers
    storage_uri="redis://localhost:6379"
)

app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)


@app.get("/public")
@limiter.limit("100/minute")
async def public_endpoint(request: Request):
    return {"status": "ok"}


# Heavy endpoint — strict limit
@app.get("/heavy-search")
@limiter.limit("10/minute;3/second")  # multiple limits separated by ;
async def heavy_search(request: Request, query: str):
    return {"results": []}


# Login — brute-force protection (5 attempts, then think)
@app.post("/auth/login")
@limiter.limit("5/minute")
async def login(request: Request):
    return {"token": "..."}

Limit by User, Not by IP

If you only limit by IP — all users behind the same corporate NAT share a single quota. Picture this: 200 office employees, one external IP, everyone hitting "Refresh" at the same time. Half of them get 429. Your CTO calls.

def get_user_or_ip(request: Request) -> str:
    """
    Authenticated → limit by user_id.
    Anonymous → limit by real IP.
    """
    token = request.headers.get("Authorization", "")
    if token.startswith("Bearer "):
        try:
            payload = decode_jwt(token[7:])
            return f"user:{payload['sub']}"
        except:
            pass

    # Respect proxy headers
    return (
        request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
        or request.headers.get("X-Real-IP", "")
        or request.client.host
    )


user_limiter = Limiter(
    key_func=get_user_or_ip,
    storage_uri="redis://localhost:6379"
)

Token Bucket — Smooth Rate Limiting Without Burst Spikes

Standard rate limiting gives you an ugly burst: 100 requests in the first second, then nothing until the minute resets. The user stares at a 429 thinking the site is broken. Token Bucket fixes this — requests flow smoothly, like water, not like rush-hour traffic:

import time
import redis.asyncio as redis


class TokenBucketRateLimiter:
    def __init__(self, redis_client, capacity: int = 10, refill_rate: float = 1.0):
        self.redis = redis_client
        self.capacity = capacity        # max tokens in the bucket
        self.refill_rate = refill_rate  # tokens per second

    async def is_allowed(self, key: str) -> tuple[bool, dict]:
        now = time.time()

        # Lua script for atomic operation — no race conditions
        script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = tonumber(bucket[1]) or capacity
        local last_refill = tonumber(bucket[2]) or now

        local elapsed = now - last_refill
        tokens = math.min(capacity, tokens + elapsed * refill_rate)

        local allowed = 0
        if tokens >= 1 then
            tokens = tokens - 1
            allowed = 1
        end

        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        redis.call('EXPIRE', key, 3600)
        return {allowed, math.floor(tokens)}
        """

        result = await self.redis.eval(script, 1, key, self.capacity, self.refill_rate, now)
        allowed, remaining = result[0], result[1]

        return bool(allowed), {
            "X-RateLimit-Limit": str(self.capacity),
            "X-RateLimit-Remaining": str(remaining),
            "X-RateLimit-Reset": str(int(now + 60))
        }

2. Security Middleware

Trusted Hosts, CORS, Headers

CORS with allow_origins=["*"] is like putting a lock on your door but taping the code next to it. We don't do that.

from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.cors import CORSMiddleware

# Only requests to our domain — guards against Host header injection
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["myapp.com", "*.myapp.com"]
)

# CORS — NEVER use allow_origins=["*"] in production
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    max_age=3600
)

app.add_middleware(GZipMiddleware, minimum_size=1000)


@app.middleware("http")
async def security_headers_middleware(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    response.headers["Server"] = "Server"  # hide server type
    return response

Timeouts: The Right Approach

Wrapping asyncio.wait_for around the entire ASGI application is a dangerous anti-pattern. When a TimeoutError is raised, a CancelledError is injected into the task. If the endpoint is mid-transaction in SQLAlchemy or inside an httpx call and doesn't explicitly catch CancelledError — the connection never returns to the pool. After a few of these incidents, the pool is exhausted and the entire API goes down. That's how Friday-evening incidents are born.

Rule: set timeouts where the actual waiting happens — in clients, not around the application.

import httpx
from sqlalchemy.ext.asyncio import create_async_engine

# Timeout on external HTTP requests
async def call_external_api(url: str):
    timeout = httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=2.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
        return response.json()

# Timeout on DB queries — in the engine config
engine = create_async_engine(
    DATABASE_URL,
    pool_timeout=10,        # wait time for a free pool connection
    pool_recycle=1800,      # reopen connections every 30 min
    connect_args={"command_timeout": 15},  # SQL query timeout (PostgreSQL)
)

The global timeout at the TCP level lives in Nginx — that's where it makes sense, because Nginx doesn't hold transactions:

proxy_read_timeout    30s;   # wait no longer than 30s for FastAPI response
proxy_connect_timeout  5s;   # connection establishment timeout
proxy_send_timeout    10s;   # request send timeout

3. Slowloris Protection

☠️ Slowloris opens thousands of connections and sends HTTP headers one byte at a time every few seconds. The server waits for the request to complete, connections pile up. A single laptop can take down your server. Named after the slow-moving primate — and yes, it's just as unstoppable without the right protection.

Uvicorn Settings

# run.py
import uvicorn

uvicorn.run(
    "main:app",
    host="0.0.0.0",
    port=8000,
    workers=4,
    timeout_keep_alive=5,           # Slowloris dies in 5 seconds
    limit_max_requests=1000,
    h11_max_incomplete_event_size=16384,  # 16KB header limit
    backlog=100,
)

Gunicorn + Uvicorn Workers (Production)

# gunicorn.conf.py
import multiprocessing

workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000

timeout = 30
keepalive = 2
graceful_timeout = 30

max_requests = 1000        # restart worker every N requests — cures memory leaks
max_requests_jitter = 100  # randomize for smoothness (more on jitter below)

bind = "0.0.0.0:8000"
backlog = 100

gunicorn -c gunicorn.conf.py main:app

4. Redis Distributed Rate Limiter

With multiple FastAPI instances behind a load balancer — each process has no idea what limits others have counted. Like three security guards at the entrance, each counting visitors from zero independently. Together they let in three times too many.

Redis fixes this.

Sliding Window Counter — The Right Algorithm

The internet is full of implementations using Redis ZSET (Sorted Set), where every request is stored as a separate element. That's Sliding Window Log — accurate, but catastrophic under attack: 20,000 req/s from one IP → 20,000 entries in Redis memory for a single key. ZREMRANGEBYSCORE becomes O(log N + M) and creates CPU spikes in single-threaded Redis.

The correct approach — Sliding Window Counter: split time into discrete intervals and store only counters. O(1) memory per IP instead of O(N). The difference is like storing every step a person takes vs just a step counter.

import redis.asyncio as aioredis
import time
from fastapi import Request, HTTPException, Depends


class SlidingWindowCounter:
    """
    Sliding Window Counter via Redis HASH.
    Window is split into 1-second intervals.
    We store only counters — not timestamps.
    Memory is O(window_size) per IP, not O(requests_count).
    """

    def __init__(self, redis: aioredis.Redis):
        self.redis = redis

    async def is_allowed(self, key: str, limit: int, window: int) -> tuple[bool, int]:
        now = int(time.time())
        pipe = self.redis.pipeline()

        # Sum across all second-buckets in the window
        buckets = [f"{key}:{now - i}" for i in range(window)]
        for bucket in buckets:
            pipe.hget("rate_counters", bucket)

        results = await pipe.execute()
        total = sum(int(r or 0) for r in results)

        if total >= limit:
            return False, 0

        # Atomically increment the current bucket
        current_bucket = f"{key}:{now}"
        await self.redis.hincrby("rate_counters", current_bucket, 1)
        # TTL on the whole hash — auto-cleanup
        await self.redis.expire("rate_counters", jittered_ttl(window + 5))  # <- with jitter!

        return True, limit - total - 1


async def rate_limit(request: Request, limit: int = 100, window: int = 60):
    key = f"{request.client.host}:{request.url.path}"
    allowed, remaining = await sliding_counter.is_allowed(key, limit, window)

    if not allowed:
        raise HTTPException(
            status_code=429,
            detail={"error": "Rate limit exceeded", "retry_after": window},
            headers={"Retry-After": str(window), "X-RateLimit-Remaining": "0"}
        )
    return remaining


@app.get("/api/expensive")
async def expensive_endpoint(
    request: Request,
    _: int = Depends(lambda r: rate_limit(r, limit=10, window=60))
):
    return {"data": "..."}

IP Blacklist via Redis

from ipaddress import ip_address, ip_network

WHITELIST_NETWORKS = [
    ip_network("10.0.0.0/8"),    # internal network
    ip_network("172.16.0.0/12"), # Docker networks
    ip_network("127.0.0.0/8"),   # localhost
]


class IPFilterMiddleware:
    def __init__(self, app, redis_client):
        self.app = app
        self.redis = redis_client

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_ip = scope["client"][0]
        ip = ip_address(client_ip)

        # Whitelist — pass through without checks
        for network in WHITELIST_NETWORKS:
            if ip in network:
                await self.app(scope, receive, send)
                return

        # Blacklist check
        is_blocked = await self.redis.get(f"blacklist:{client_ip}")
        if is_blocked:
            response = JSONResponse(status_code=403, content={"detail": "Access denied"})
            await response(scope, receive, send)
            return

        await self.app(scope, receive, send)


@app.post("/admin/blacklist/{ip}")
async def block_ip(ip: str, duration: int = 3600):
    await redis_client.setex(f"blacklist:{ip}", duration, 1)
    return {"blocked": ip, "duration": duration}

5. Thundering Herd — TTL Jitter Against the Stampede Effect

We have 9 Gunicorn workers, Redis, a Circuit Breaker with recovery_timeout = 30. Ask yourself: what happens when that timeout expires on all workers simultaneously?

A real-life scenario: the database goes down at 14:32:00. Circuit Breaker enters OPEN state. Exactly at 14:32:30 all 9 workers synchronously transition to HALF_OPEN and fire a probe request to the DB. If the DB is still unstable — it gets an instant connection spike and goes down again. Circuit Breaker goes back to OPEN. At 14:33:00, the cycle repeats. This is the Thundering Herd — the stampede effect where your protection mechanism is what breaks the thing it's protecting.

Solution: Jitter (random TTL spread). Never use static expiration times for states that expire across multiple workers simultaneously. Desynchronize them — let each worker wake up at its own moment.

import random


def jittered_ttl(base_seconds: int, variance: float = 0.15) -> int:
    """
    Adds ±15% random spread to TTL.
    base=30s → actual TTL between 25.5s and 34.5s.
    Workers wake up at different times — no spike.
    """
    delta = int(base_seconds * variance)
    return base_seconds + random.randint(-delta, delta)

Apply it in RedisCircuitBreaker:

class RedisCircuitBreaker:
    # ... (previous code)

    async def _on_failure(self, now: float):
        # Before: open_ttl = self.recovery_timeout * 2  — all workers in sync
        # After: each worker gets a slightly different TTL — beautiful
        ttl = jittered_ttl(self.recovery_timeout * 2)
        await self.redis.eval(
            self._RECORD_FAILURE_SCRIPT, 1,
            self.key, self.failure_threshold, ttl, now
        )

And in SlidingWindowCounter — window expirations across workers shouldn't be synchronized:

# Before: await self.redis.expire("rate_counters", window + 5)
await self.redis.expire("rate_counters", jittered_ttl(window + 5))

General rule: any EXPIRE in Redis that could fire across multiple consumers simultaneously must have jitter. This is especially critical for Circuit Breaker recovery_timeout and TTL on hot cache keys. A small spread — a huge relief for your database.

6. Circuit Breaker — Distributed, via Redis

If the database is down — Circuit Breaker quickly rejects requests instead of waiting for timeouts. But there's a critical catch: state stored in process memory doesn't work with multiple workers.

With gunicorn running cpu_count() * 2 + 1 workers, you have ~9 isolated Python processes. Worker A opens the circuit, workers B–I keep hammering the dead DB, thinking it's still closed. It's like one security guard knowing the building is on fire while the other eight keep letting people in. State must live in Redis.

from enum import Enum
import redis.asyncio as aioredis
import time


class CircuitState(Enum):
    CLOSED    = "closed"     # normal operation
    OPEN      = "open"       # all requests blocked
    HALF_OPEN = "half_open"  # probe request to check recovery


class RedisCircuitBreaker:
    """
    Circuit Breaker with state stored in Redis.
    All workers see the same state.
    Lua script guarantees atomic state transitions.
    """

    _RECORD_FAILURE_SCRIPT = """
    local key = KEYS[1]
    local threshold = tonumber(ARGV[1])
    local open_ttl  = tonumber(ARGV[2])
    local now       = tonumber(ARGV[3])

    local failures = redis.call('HINCRBY', key, 'failures', 1)
    redis.call('HSET', key, 'last_failure', now)

    if failures >= threshold then
        redis.call('HSET', key, 'state', 'open', 'opened_at', now)
        redis.call('EXPIRE', key, open_ttl * 2)
    end
    return failures
    """

    def __init__(
        self,
        redis: aioredis.Redis,
        name: str,
        failure_threshold: int = 5,
        recovery_timeout: int = 30,
        success_threshold: int = 2,
    ):
        self.redis = redis
        self.key = f"circuit:{name}"
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

    async def _get_state(self) -> dict:
        data = await self.redis.hgetall(self.key)
        return {k.decode(): v.decode() for k, v in data.items()} if data else {}

    async def call(self, func, *args, **kwargs):
        state = await self._get_state()
        circuit_state = state.get("state", "closed")
        now = time.time()

        if circuit_state == "open":
            opened_at = float(state.get("opened_at", 0))
            if now - opened_at >= self.recovery_timeout:
                await self.redis.hset(self.key, mapping={"state": "half_open", "successes": 0})
            else:
                raise HTTPException(503, "Service temporarily unavailable")

        try:
            result = await func(*args, **kwargs)
            await self._on_success()
            return result
        except Exception:
            await self._on_failure(now)
            raise

    async def _on_success(self):
        state = await self._get_state()
        if state.get("state") == "half_open":
            successes = await self.redis.hincrby(self.key, "successes", 1)
            if successes >= self.success_threshold:
                await self.redis.delete(self.key)
        else:
            await self.redis.hset(self.key, "failures", 0)

    async def _on_failure(self, now: float):
        ttl = jittered_ttl(self.recovery_timeout * 2)  # <- jitter here
        await self.redis.eval(
            self._RECORD_FAILURE_SCRIPT, 1,
            self.key, self.failure_threshold, ttl, now
        )


db_circuit = RedisCircuitBreaker(redis_client, name="postgres", failure_threshold=5)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return await db_circuit.call(fetch_user_from_db, user_id)

7. Nginx — Correct Configuration

http {
    limit_req_zone  $binary_remote_addr  zone=api:10m   rate=10r/s;
    limit_req_zone  $binary_remote_addr  zone=heavy:10m rate=1r/s;
    limit_conn_zone $binary_remote_addr  zone=conn:10m;
    limit_conn_status 429;
    limit_req_status  429;

    server {
        listen 443 ssl http2;
        server_name myapp.com;

        # Slowloris timeouts — wait no more than 10 seconds for headers
        client_header_timeout 10s;
        client_body_timeout   10s;
        send_timeout          10s;
        keepalive_timeout     5s;

        client_max_body_size        10m;
        client_header_buffer_size   1k;
        large_client_header_buffers 4 8k;

        # Max 10 connections per IP — prevents Slowloris from sticking
        limit_conn conn 10;

        location /api/ {
            limit_req zone=api burst=20 nodelay;

            proxy_pass            http://127.0.0.1:8000;
            proxy_read_timeout    30s;
            proxy_connect_timeout 5s;
            proxy_set_header      X-Real-IP        $remote_addr;
            proxy_set_header      X-Forwarded-For  $proxy_add_x_forwarded_for;
        }

        location ~* ^/api/(search|export|report) {
            limit_req zone=heavy burst=5;
            proxy_pass         http://127.0.0.1:8000;
            proxy_read_timeout 120s;
        }

        # Block popular scanners — looking at you, nikto
        if ($http_user_agent ~* (nikto|sqlmap|masscan|nmap)) {
            return 403;
        }

        # Empty User-Agent — close connection silently
        if ($http_user_agent = "") {
            return 444;
        }
    }
}

💡 Status code 444 in Nginx closes the TCP connection with no HTTP response. Cheaper than sending a 403 — we don't waste bandwidth on aggressive bots. Zero bytes in response = zero resources spent talking to someone who doesn't deserve a reply.

Geo-blocking for Targeted Attacks

# Requires ngx_http_geoip2_module
geoip2 /etc/nginx/GeoLite2-Country.mmdb {
    $geoip2_data_country_code country iso_code;
}

map $geoip2_data_country_code $allowed_country {
    default 0;
    RU 1;
    KZ 1;
    US 1;
}

server {
    if ($allowed_country = 0) {
        return 403;
    }
}

8. Cloudflare — CDN Layer

Feature	Free	Pro ($20/mo)	Business
L3/L4 DDoS Protection	✅	✅	✅
WAF	❌	✅	✅
Advanced Rate Limiting	❌	✅	✅
Bot Fight Mode	✅	✅	✅
Custom Rules	5	20	100+

Cloudflare Worker — Rate Limiting at the Edge

Requests are blocked before they even reach your server. It's like a bouncer at the club — they turn away troublemakers on the street, not at the bar.

const RATE_LIMIT = 100;
const WINDOW = 60; // seconds

export default {
    async fetch(request, env) {
        const ip = request.headers.get('CF-Connecting-IP');
        const key = `rate:${ip}:${Math.floor(Date.now() / 1000 / WINDOW)}`;

        const current = parseInt(await env.RATE_KV.get(key) || '0');

        if (current >= RATE_LIMIT) {
            return new Response('Too Many Requests', {
                status: 429,
                headers: {
                    'Retry-After': String(WINDOW),
                    'X-RateLimit-Remaining': '0'
                }
            });
        }

        await env.RATE_KV.put(key, String(current + 1), { expirationTtl: WINDOW });
        return fetch(request);
    }
};

Blocking Direct Server Access — This Is Mandatory

If ports 80/443 are open directly — an attacker can bypass Cloudflare by hitting your server's IP. All your CDN protection becomes meaningless. It's like having an armored door next to a hole in the wall.

# Download current Cloudflare IP ranges
wget -O /tmp/cf-ips.txt https://www.cloudflare.com/ips-v4

iptables -F INPUT
iptables -A INPUT -p tcp --dport 22 -j ACCEPT  # Keep SSH — otherwise you're locked out

while IFS= read -r ip; do
    iptables -A INPUT -p tcp --dport 80  -s "$ip" -j ACCEPT
    iptables -A INPUT -p tcp --dport 443 -s "$ip" -j ACCEPT
done < /tmp/cf-ips.txt

iptables -A INPUT -j DROP  # everything else — no

9. OS Kernel: eBPF/XDP vs iptables

In the previous section we locked down direct access using iptables. That's the standard — but it has an architectural ceiling that becomes visible under truly serious volumetric attacks.

The iptables problem under volumetric attack. Even if a rule drops a packet, the Linux kernel still allocates an sk_buff buffer for every incoming packet before iptables processes it. At 10–20 Mpps (millions of packets per second) this exhausts kernel memory before the rules even fire. You're dropping packets, but you already paid the memory cost — a paradox of protection.

Solution: XDP (eXpress Data Path) + eBPF. XDP lets you attach an eBPF program directly to the NIC driver. The packet is dropped before the kernel even sees it — before sk_buff allocation, before the IP stack, before iptables. Like a doorman who turns away unwanted guests in the parking garage — not at reception.

Without XDP: NIC → sk_buff alloc → TCP stack → iptables → DROP  (memory wasted)
With XDP:    NIC → eBPF prog → XDP_DROP                          (memory untouched)

Minimal XDP program in C (compiled via clang, loaded via ip link):

// xdp_drop_syn.c
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/tcp.h>

SEC("xdp")
int block_syn_flood(struct xdp_md *ctx) {
    void *data     = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;

    struct ethhdr *eth = data;
    if ((void *)(eth + 1) > data_end) return XDP_PASS;
    if (eth->h_proto != __constant_htons(ETH_P_IP)) return XDP_PASS;

    struct iphdr *ip = (void *)(eth + 1);
    if ((void *)(ip + 1) > data_end) return XDP_PASS;
    if (ip->protocol != IPPROTO_TCP) return XDP_PASS;

    struct tcphdr *tcp = (void *)(ip + 1);
    if ((void *)(tcp + 1) > data_end) return XDP_PASS;

    // Drop bare SYN without ACK — classic SYN flood
    if (tcp->syn && !tcp->ack) {
        // In production: check against eBPF map of blocked IPs
        return XDP_DROP;
    }

    return XDP_PASS;
}

# Load program onto the interface (native mode — maximum speed)
ip link set dev eth0 xdp obj xdp_drop_syn.o sec xdp

Layer allocation rule: iptables — for stateful rules and port-based filtering. eBPF/XDP — for high-speed dropping of volumetric L3/L4 attacks at the NIC level. For bare-metal servers under serious attack the difference in effectiveness is an order of magnitude. This isn't a replacement — it's an additional layer in front of everything else.

10. Monitoring and Automatic Auto-ban

The previous version of the anomaly detector stored request history in defaultdict(list) in process memory. That's two bugs in one:

Memory leak: 100,000 unique IPs make one request each and disappear — their keys stay in memory until Uvicorn restarts. A slow death by OOM.
Event loop blocking: list comprehension over 5,000 timestamps on every request — a synchronous operation in async context, delaying all other requests. One bad IP slows down all good users.

All state goes to Redis. TTL and LTRIM handle the cleanup automatically:

import redis.asyncio as aioredis
import logging
import time

logger = logging.getLogger(__name__)


class AnomalyDetector:
    """
    All state in Redis — no memory leaks, no event loop blocking.
    LPUSH/LTRIM keep only the last N timestamps.
    TTL automatically cleans up inactive IPs.
    """

    def __init__(self, redis: aioredis.Redis):
        self.redis = redis

    async def analyze_request(self, request: Request) -> dict:
        ip = request.client.host
        signals = {}

        # 1. High frequency — counter with TTL instead of in-memory list
        freq_key = f"freq:{ip}"
        count = await self.redis.incr(freq_key)
        await self.redis.expire(freq_key, 10)
        signals["high_frequency"] = count > 50

        # 2. Suspicious User-Agent (basic check — extended by JA3 below)
        ua = request.headers.get("user-agent", "")
        signals["suspicious_ua"] = any(
            bot in ua.lower()
            for bot in ["python-requests", "curl", "wget", "scrapy", "go-http"]
        )

        # 3. JA3 TLS fingerprint — real client identification (see section 11)
        KNOWN_BAD_JA3 = {
            "6734f37431670b3ab4292b8f60f29984",  # python-requests
            "b32309a26951912be7dba376398abc3b",  # curl
            "4d7a28d6f2263ed61de88ca66eb011e3",  # Scrapy
        }
        ja3 = request.headers.get("X-JA3-Fingerprint", "")
        signals["bad_tls_fingerprint"] = ja3 in KNOWN_BAD_JA3

        # 4. High 4xx error rate
        error_count = await self.redis.get(f"errors:{ip}")
        signals["high_errors"] = int(error_count or 0) > 20

        # 5. Scanning known vulnerable paths
        scan_paths = ["/admin", "/.env", "/wp-admin", "/phpMyAdmin", "/.git"]
        signals["scanning"] = any(request.url.path.startswith(p) for p in scan_paths)

        threat_score = sum(1 for v in signals.values() if v)
        if threat_score >= 2:
            await self._auto_block(ip, threat_score, signals)

        return signals

    async def record_error(self, ip: str):
        """Call this from middleware on 4xx responses."""
        key = f"errors:{ip}"
        await self.redis.incr(key)
        await self.redis.expire(key, 300)

    async def _auto_block(self, ip: str, score: int, signals: dict):
        duration = 86400 if score >= 3 else 3600
        # SET NX — don't overwrite if already blocked
        blocked = await self.redis.set(f"blacklist:{ip}", score, ex=duration, nx=True)
        if blocked:
            logger.warning(f"AUTO-BLOCKED {ip} for {duration}s | score={score} | {signals}")
            await self._send_telegram_alert(ip, score, signals)

    async def _send_telegram_alert(self, ip, score, signals):
        import httpx
        message = f"🚨 DDoS Alert!\nIP: {ip}\nScore: {score}/5\nSignals: {signals}"
        async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client:
            await client.post(
                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                json={"chat_id": CHAT_ID, "text": message}
            )


@app.middleware("http")
async def track_errors_middleware(request: Request, call_next):
    response = await call_next(request)
    if 400 <= response.status_code < 500:
        await anomaly_detector.record_error(request.client.host)
    return response

Prometheus Metrics

from prometheus_client import Counter, Histogram
from prometheus_fastapi_instrumentator import Instrumentator

RATE_LIMIT_HITS = Counter(
    "rate_limit_hits_total",
    "Total rate limit violations",
    ["endpoint", "ip"]
)

REQUEST_DURATION = Histogram(
    "request_duration_seconds",
    "Request processing time",
    ["method", "endpoint", "status"],
    buckets=[.005, .01, .025, .05, .1, .5, 1, 5]
)

Instrumentator().instrument(app).expose(app, endpoint="/metrics")

11. JA3/JA4 TLS Fingerprinting — Detect Bots by Their Handshake

In AnomalyDetector we check the User-Agent. That's necessary, but naive: any script changes that header in one line. It's like trying to identify a professional burglar by asking their name.

The problem: HTTP headers are controlled by the attacker. The TLS handshake is not.

When establishing an HTTPS connection, the client sends a ClientHello — a list of supported cipher suites, extensions, elliptic curve groups. Chrome sends a specific set, Python requests sends a different one, Go net/http — yet another. This set is hashed into a JA3 fingerprint (MD5, 32 characters). The fingerprint changes when the library is updated, but within the same version — it's stable. Faking it without patching the SSL library is not possible.

Chrome 120:       ja3 = "cd08e31494f9531f560d64c695473da9"  ← legitimate user
Python requests:  ja3 = "6734f37431670b3ab4292b8f60f29984"  ← instant red flag
curl 8.x:         ja3 = "b32309a26951912be7dba376398abc3b"  ← a tool, not a browser

Getting JA3 in Nginx

# Requires: openresty or nginx + lua-nginx-module + lua-resty-ja3
http {
    lua_shared_dict ja3_block 10m;

    server {
        access_by_lua_block {
            local ja3 = require("resty.ja3")
            local fingerprint = ja3.fingerprint()

            -- List of known bad fingerprints (botnets, scanners)
            local blocklist = {
                ["6734f37431670b3ab4292b8f60f29984"] = true,  -- python-requests
                ["b32309a26951912be7dba376398abc3b"] = true,  -- curl
            }

            if blocklist[fingerprint] then
                ngx.header["X-JA3-Blocked"] = fingerprint
                ngx.exit(ngx.HTTP_FORBIDDEN)
            end

            -- Forward fingerprint to FastAPI for logging and analysis
            ngx.req.set_header("X-JA3-Fingerprint", fingerprint)
        }
    }
}

FastAPI receives the fingerprint via the X-JA3-Fingerprint header and AnomalyDetector handles it (see the bad_tls_fingerprint signal above).

The Cloudflare path (no Nginx config needed): on Pro plan and above, JA3/JA4 are available as cf.bot_management.ja3_hash in Custom Rules — without a single line of server-side code. One click in the UI — and all Python bots are gone.

Important caveat: JA3 blocking is not a silver bullet. Sophisticated bots use browser engines (Playwright, Puppeteer) with a real Chrome TLS stack and a real Chrome fingerprint. But it effectively cuts out 80–90% of automated traffic from scripts and custom clients — the ones that didn't bother. And that's the majority.

12. Chaos Engineering — Proving Your Protection Actually Works

We wrote a Circuit Breaker, AnomalyDetector, Sliding Window. Beautiful code. Passed code review. Everyone said "looks good."

But how do you know it actually works in production and not just in unit tests with a mocked Redis? The honest answer: you can't — unless you deliberately broke things in staging first.

The enterprise answer: Chaos Engineering — intentionally injecting failures into staging to verify graceful degradation. Netflix calls this the "Design for failure" principle. We call it "let's break this before it breaks itself at 3 AM."

Toxiproxy — Injecting Failures at the TCP Layer

# Run Toxiproxy next to your staging DB
docker run -d -p 8474:8474 -p 5433:5433 ghcr.io/shopify/toxiproxy

# chaos_tests/test_circuit_breaker.py
import pytest
import httpx
import asyncio
from toxiproxy import Toxiproxy


@pytest.fixture
def proxy():
    client = Toxiproxy()
    # Proxy PostgreSQL: localhost:5433 → postgres:5432
    proxy = client.create(name="postgres", listen="0.0.0.0:5433", upstream="postgres:5432")
    yield proxy
    proxy.destroy()


async def test_circuit_opens_under_latency(proxy):
    """
    Inject 3-second latency into PostgreSQL.
    Circuit Breaker should open after 5 timeouts
    and start returning 503 — fast, not hanging.
    """
    proxy.add_toxic("latency", type="latency", attributes={"latency": 3000})

    failed = 0
    circuit_opened = False

    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        for i in range(20):
            r = await client.get("/users/1")
            if r.status_code == 503:
                circuit_opened = True
                break
            if r.status_code == 504:
                failed += 1

    assert circuit_opened, f"Circuit didn't open after {failed} timeouts — something's wrong"


async def test_circuit_recovers_after_heal(proxy):
    """
    After the circuit opens, remove the toxic.
    After recovery_timeout the system should self-heal.
    No restart. No 3 AM calls.
    """
    proxy.add_toxic("down", type="limit_data", attributes={"bytes": 0})
    await asyncio.sleep(2)
    proxy.remove_toxic("down")

    await asyncio.sleep(35)  # wait recovery_timeout + buffer

    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        r = await client.get("/users/1")
        assert r.status_code == 200, "System didn't recover — Circuit stuck in OPEN"


async def test_rate_limiter_under_flood(proxy):
    """
    200 requests from one IP at once.
    Should get 429, not 500.
    500 = your code crashed. 429 = your protection is working.
    """
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        tasks = [client.get("/api/expensive") for _ in range(200)]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

    status_codes = [r.status_code for r in responses if isinstance(r, httpx.Response)]
    assert 429 in status_codes, "Rate limiter didn't fire under flood"
    assert 500 not in status_codes, "Server returned 500 — this is a crash, not controlled degradation"

# Run in CI/CD pipeline (STAGING ONLY — never do this in production)
pytest chaos_tests/ -v --timeout=120

What to Test in Chaos Tests

Scenario	Tool	Expected behavior
3s latency on PostgreSQL	Toxiproxy `latency`	Circuit opens, 503 instead of 504
Full DB outage	Toxiproxy `limit_data`	Circuit open, recovery in N seconds
Redis unavailable	`docker pause redis`	Graceful degradation, no crash
1000 req/s flood	`wrk` / `locust`	429 at limit, zero 500s
Worker OOM	`kill -9 <pid>`	Gunicorn spawns new one, no traffic loss

The core Chaos Engineering insight: we don't care whether the system falls under load (it will — that's normal). We care whether it degrades gracefully. Correct degradation: 429 instead of 500, 503 instead of hanging, predictable recovery. Incorrect: OOM crash, connection leak, infinite retry storm. You only find out which one you have by breaking things intentionally — or accidentally in production.

13. Production Docker Compose

version: '3.9'

services:
  nginx:
    image: nginx:alpine
    ports: ["443:443", "80:80"]
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./certs:/etc/nginx/certs
    depends_on: [fastapi]

  fastapi:
    build: .
    command: gunicorn -c gunicorn.conf.py main:app
    environment:
      REDIS_URL: redis://redis:6379
    depends_on: [redis]
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 512M
          cpus: '1.0'

  redis:
    image: redis:7-alpine
    # allkeys-lru: on low memory evict old keys, don't crash
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes: [redis_data:/data]

  prometheus:
    image: prom/prometheus
    volumes: [./prometheus.yml:/etc/prometheus/prometheus.yml]

  grafana:
    image: grafana/grafana
    environment:
      GF_SECURITY_ADMIN_PASSWORD: strongpassword  # please change this. seriously.
    ports: ["3000:3000"]

volumes:
  redis_data:

Pre-Deploy Checklist

Print it out. Pin it above your monitor. Save it in Notion. Tattoo it on your wrist — your choice.

[ ] Nginx sits in front of FastAPI with limit_req_zone and limit_conn
[ ] client_max_body_size 10m in Nginx — not Python middleware (bypassed via Transfer-Encoding: chunked)
[ ] Redis connected for distributed rate limiting
[ ] Rate limiting uses Sliding Window Counter (HINCRBY), not ZSET Log
[ ] TTL Jitter applied to Circuit Breaker recovery_timeout and all caches
[ ] Timeouts set in httpx.AsyncClient and DB engine — not via asyncio.wait_for around the app
[ ] proxy_read_timeout in Nginx as global backstop
[ ] CORS configured strictly — only allowed origins, not *
[ ] Security headers set: HSTS, CSP, X-Frame-Options
[ ] Direct server access blocked via iptables — only Cloudflare IPs allowed
[ ] Cloudflare with Bot Fight Mode enabled
[ ] JA3 fingerprint checked at Nginx or Cloudflare Custom Rules level
[ ] Prometheus + Grafana monitoring configured
[ ] Telegram/PagerDuty alerts on anomalies
[ ] Circuit Breaker with Redis state — not in process memory
[ ] AnomalyDetector fully on Redis — no defaultdict in worker memory
[ ] Login and critical endpoints have their own strict limits
[ ] Chaos tests (Toxiproxy) run in staging and passed
[ ] XDP/eBPF considered for bare-metal under volumetric L4 attacks

Summary: 8 Layers of Defense

eBPF/XDP (NIC level — before the kernel)
  └─ CDN (L3/L4 — Cloudflare + JA3 fingerprinting)
       └─ Firewall (IP filtering, iptables)
            └─ Nginx (connection + rate limits, Slowloris timeouts)
                 └─ FastAPI Middleware (headers, CORS, security)
                      └─ Rate Limiter (Redis Sliding Window + Jitter)
                           └─ Business Logic (Circuit Breaker → Redis)
                                └─ Monitoring (Anomaly Detection + Auto-ban + Chaos Tests)

Each layer is cheaper than the one above it in CPU cost. An attack has to pass through all eight — almost nothing reaches the business logic. And if something does — monitoring already sent an alert and auto-ban is running.

If this was useful — leave a ❤️ and bookmark it. The next part will cover JWT security and credential stuffing protection. There'll be production bugs there too — there always are.

💼 Python Backend & Automation Engineer | Specialized in Scalable Data Systems & API Architecture. Relocating to Saskatoon, SK via SINP Tech Pathway — Open to joining high-growth Canadian engineering teams.

The Invisible War: Why Your Automation Fails (and How I Build Systems That Don't)

Adilet Akmatov — Tue, 24 Mar 2026 09:09:43 +0000

The Invisible War: A Complete Engineering Reference for Production Web Automation

⚖️ Legal note before anything else: Web scraping exists in a legally complex space. Before building anything like this, review the target site's Terms of Service and robots.txt, understand applicable law in your jurisdiction (CFAA in the US, Computer Misuse Act in the UK, GDPR in the EU), and — ideally — have explicit permission or a legal opinion. The techniques below are documented for educational purposes. "I found it on the internet" is not a legal defence.

I spent 8 years in retail operations before moving into development. That background shapes how I think about automation: the business problem comes first, the code is just the means. This post is about what I learned building a pricing-intelligence system that actually ran in production — including the bugs I shipped, the incidents that broke it, and the architectural gaps I only recognised after the second outage.

It's not a "look at my dashboard" post. It's an engineering breakdown, including the parts that are embarrassing to admit.

Why "Standard" Scraping Fails Immediately

If you're reaching for requests + BeautifulSoup for anything modern, the failure isn't that your selectors are wrong. The failure happens before you send a single HTTP header.

Modern anti-bot systems analyse behavioural signals and hardware fingerprints at multiple simultaneous layers. Miss any one of them and you're blocked — often silently, sometimes with poisoned data returned instead of a 403.

Here's what those layers actually look like.

Layer 0: TLS Fingerprinting — You're Dead Before HTTP

import requests

# This dies before reaching the application layer.
# The TLS handshake itself identifies you as Python requests.
response = requests.get("https://target-site.com/products")

The problem isn't your User-Agent header. It's not even your IP. Standard Python HTTP clients produce distinctive TLS fingerprints — specifically JA3, JA4, and JA4+ hashes — that Cloudflare, Akamai, and DataDome identify at the socket level, before they parse a single HTTP header.

JA3 vs JA4: Why the Generation Gap Matters

JA3 hashes the ClientHello message by concatenating: TLS version + cipher suites + extensions + elliptic curves + elliptic curve formats. It's a 32-character MD5 hash. Blocking by JA3 is table-stakes — every major WAF does it.

JA4 is a different beast. Its format is human-readable by design:

{transport}{tls_version}{sni_present}{cipher_count}{extension_count}{alpn}
_{cipher_suite_hash}_{extension_hash}

A concrete example from Chrome 122 on Windows:

t13d1516h2_acb858a92679_14cb8fd47a25
│││ │  │ ││
│││ │  │ │└── ALPN: h2 (HTTP/2 negotiated)
│││ │  │ └─── 16 TLS extensions
│││ │  └───── 15 cipher suites
│││ └──────── SNI present (d = domain)
││└────────── TLS 1.3
│└─────────── TCP transport
└──────────── t = TLS (not QUIC)

Now compare what Python's urllib3 (used by requests) produces:

t13d1912h2_acb858a92679_...
        ││
        │└── 12 extensions (vs 16 in Chrome)
        └─── 19 cipher suites (vs 15 in Chrome)

The cipher count and extension count are different before you even look at the hashes. This is why changing your User-Agent to Chrome while using requests doesn't help — the JA4 prefix t13d1912 is a hard signal that you're not Chrome, regardless of what headers you send.

What JA4 adds beyond JA3:

Extension ordering — Chrome loads extensions in a specific order that reflects its internal TLS stack. Python's ssl module uses a different order.
ALPN protocol preference — Chrome negotiates HTTP/2 by default. Some HTTP clients don't include ALPN at all, or order h2 and http/1.1 differently.
Signature algorithm list — Chrome includes specific signature algorithms (ed25519, ecdsa-secp256r1-sha256) that urllib3's default SSL context omits.
JA4+ additionally fingerprints the server's response: which cipher the server selected, certificate details, extension responses. This catches clients that successfully mimic the ClientHello but can't mimic how they respond to the ServerHello.

If your evasion strategy is JA3-era — rotating User-Agents, tweaking headers — you're a generation behind. The handshake has already identified you.

The Fix: Start Clean, Don't Patch Dirty

# ❌ Vanilla Playwright — navigator.webdriver = true, Mesa OffScreen GPU exposed,
#    and the Chromium binary produces a recognisable TLS fingerprint.
browser = p.chromium.launch(headless=True)

# ✅ Option A: Camoufox — a hardened Firefox fork built for stealth automation.
#    Ships with patched fingerprints, a realistic GPU identity, and correct
#    navigator properties out of the box. Best integrity for Firefox-compatible targets.
#    pip install camoufox
from camoufox.async_api import AsyncCamoufox
async with AsyncCamoufox(headless=True) as browser:
    page = await browser.new_page()

# ✅ Option B: rebrowser-patches — patches applied at Chrome source level,
#    not via JS injection at runtime. A source-level patch is harder to detect
#    than monkey-patching because it can't be caught by checking whether native
#    function .toString() returns "[native code]".
#    https://github.com/rebrowser/rebrowser-patches

# ✅ Option C: curl_cffi — if you need an HTTP client rather than a full browser.
#    Impersonates real browser TLS profiles at the socket level via libcurl.
#    pip install curl_cffi
from curl_cffi.requests import AsyncSession
async with AsyncSession() as session:
    # impersonate= sets the TLS fingerprint + header order to match the target browser
    response = await session.get(url, impersonate="chrome122")

Choose based on your threat model:

Pure HTTP endpoints without JS rendering → curl_cffi
Firefox-compatible targets where maximum fingerprint integrity matters → Camoufox
Chromium targets where site JS compatibility matters → rebrowser-patches or playwright-stealth (Layer 1)

The key principle: don't add stealth patches on top of a flagged baseline. Start with the cleanest baseline available, then add targeted patches for surfaces that baseline doesn't cover.

Layer 1: Stealth Patching — Necessary but Not Sufficient

playwright-stealth handles the obvious signals. Understanding which signals it patches — and which it doesn't — determines whether you need to go further.

import asyncio
from playwright.async_api import async_playwright
from playwright_stealth import stealth_async


async def create_stealth_page(browser, proxy_location: dict):
    context = await browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent=(
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/122.0.0.0 Safari/537.36"
        ),
        locale=proxy_location.get("locale", "en-US"),
        timezone_id=proxy_location.get("timezone", "America/New_York"),
    )

    page = await context.new_page()
    await stealth_async(page)

    # Never use `assert` for runtime validation in production code.
    # Python's -O optimisation flag silently strips all assert statements,
    # turning this into a no-op in deployed environments.
    webdriver_check = await page.evaluate("navigator.webdriver")
    if webdriver_check is not None:
        raise RuntimeError(
            f"Stealth patch failed: navigator.webdriver = {webdriver_check!r}. "
            "Check playwright-stealth version compatibility with your Chromium build."
        )

    return page

What playwright-stealth patches:

navigator.webdriver — the most obvious automation flag
chrome.runtime — mimics a real extension environment
permissions.query — normalises responses that headless browsers answer differently
window.screen — fixes dimension inconsistencies in headless mode
navigator.plugins — populates the plugins array (empty in headless by default)

What it doesn't patch (you need Layer 3 for these):

WebGL renderer string
Canvas fingerprint
AudioContext output
Font enumeration results

Layer 2: Behavioural Simulation — Sessions, Not Clicks

Real users don't move the mouse in straight lines or click at pixel-perfect coordinates. Behavioural analysis engines like PerimeterX score entire sessions — mouse velocity profiles, click precision, scroll patterns, time-on-page distributions. A single unnatural movement can flag a session that was otherwise clean.

The Bug in the Original Implementation

The original HumanPage had two problems:

It wasn't a Bézier curve. The docstring said "Bezier-like" but the implementation was linear interpolation with a sinusoidal noise term. A real cubic Bézier requires four control points and a specific interpolation formula.
The physics were backwards relative to Fitts's Law. The noise was multiplied by sin(progress * π) — maximum scatter in the middle of the movement, zero at start and end. Real human motor control (Fitts's Law) works the opposite way: a fast ballistic phase with minimal correction makes up the first ~70% of movement, followed by a slow corrective phase with fine Gaussian adjustments near the target. Scorers that model velocity profiles will catch the sinusoidal pattern.
Step count was distance-independent. random.randint(25, 45) regardless of whether you're moving 10px or 800px. Real mouse movement time scales logarithmically with distance (Fitts's Law: MT = a + b * log₂(2D/W)). A fixed step count produces unnatural velocity at extreme distances.

import asyncio
import random
import math
from dataclasses import dataclass


@dataclass
class Point:
    x: float
    y: float

    def __add__(self, other: "Point") -> "Point":
        return Point(self.x + other.x, self.y + other.y)

    def __mul__(self, scalar: float) -> "Point":
        return Point(self.x * scalar, self.y * scalar)

    def distance_to(self, other: "Point") -> float:
        return math.hypot(other.x - self.x, other.y - self.y)


def cubic_bezier(t: float, p0: Point, p1: Point, p2: Point, p3: Point) -> Point:
    """
    Cubic Bézier: B(t) = (1-t)³P0 + 3(1-t)²tP1 + 3(1-t)t²P2 + t³P3

    Four control points:
        P0: start position
        P1: first control point (influences departure angle and speed)
        P2: second control point (influences arrival angle and speed)
        P3: end position (target)
    """
    mt = 1.0 - t
    return (
        p0 * (mt ** 3) +
        p1 * (3 * mt ** 2 * t) +
        p2 * (3 * mt * t ** 2) +
        p3 * (t ** 3)
    )


def fitts_steps(distance: float, target_width: float = 10.0) -> int:
    """
    Estimate step count from Fitts's Law movement time model.

    MT = a + b * log2(2D / W)

    Constants (a=0.1, b=0.15) are approximate empirical values for mouse input.
    The result is clamped: very short movements need at least 15 steps for
    smoothness; very long ones are capped at 80 to avoid excessive runtime.
    """
    if distance < 1:
        return 15
    index_of_difficulty = math.log2(2 * distance / target_width)
    mt = 0.1 + 0.15 * index_of_difficulty  # seconds (approximate)
    steps = int(mt * 60)  # 60 "steps" per second of movement time
    return max(15, min(80, steps))


class HumanPage:
    """
    Playwright page wrapper with physically-motivated mouse movement.

    Movement model:
      - Cubic Bézier curve for the overall path shape
      - Ballistic phase (t=0.0–0.7): fast movement, minimal correction
      - Corrective phase (t=0.7–1.0): slow approach, Gaussian scatter near target
      - Step count scales with distance via Fitts's Law

    Why per-instance state, not a module-level dict:
      In an asyncio worker pool with multiple pages running concurrently,
      a shared global is a race condition: worker A updates position mid-move
      while worker B reads it for its own calculation. Each HumanPage owns
      position state for its page only.
    """

    def __init__(self, page):
        self._page = page
        self._pos = Point(0.0, 0.0)

    def _build_control_points(self, start: Point, end: Point) -> tuple[Point, Point]:
        """
        Generate P1 and P2 control points that produce a natural arc.

        P1 is offset perpendicular to the movement direction, biased toward
        the first third of the path — this creates the curved departure.

        P2 is placed near the target with a slight perpendicular offset in
        the opposite direction — this creates the converging approach.

        The perpendicular magnitude scales with distance so that short
        movements don't produce implausible arcs.
        """
        dx = end.x - start.x
        dy = end.y - start.y
        distance = math.hypot(dx, dy)

        # Unit perpendicular vector
        if distance > 0:
            perp_x = -dy / distance
            perp_y = dx / distance
        else:
            perp_x, perp_y = 0.0, 1.0

        arc_magnitude = distance * random.uniform(0.1, 0.25)
        arc_direction = random.choice([-1, 1])

        # P1: near start, pushed perpendicular
        p1 = Point(
            start.x + dx * 0.3 + perp_x * arc_magnitude * arc_direction,
            start.y + dy * 0.3 + perp_y * arc_magnitude * arc_direction,
        )
        # P2: near end, slightly pushed back toward centre
        p2 = Point(
            start.x + dx * 0.8 + perp_x * arc_magnitude * arc_direction * 0.3,
            start.y + dy * 0.8 + perp_y * arc_magnitude * arc_direction * 0.3,
        )
        return p1, p2

    async def move(self, target_x: float, target_y: float) -> None:
        """
        Move mouse from current position to (target_x, target_y).

        Phase split at t=0.7:
          Ballistic (t < 0.7): velocity is high, Gaussian jitter is minimal.
            Real ballistic movement is essentially open-loop: the motor command
            is issued and not corrected until the corrective phase begins.
          Corrective (t >= 0.7): velocity decreases, Gaussian scatter increases
            as the hand homes in on the target. Sigma scales with distance to
            target — wide early in the corrective phase, tight at the end.
        """
        start = self._pos
        end = Point(target_x, target_y)
        distance = start.distance_to(end)

        if distance < 2:
            self._pos = end
            return

        p1, p2 = self._build_control_points(start, end)
        steps = fitts_steps(distance)

        BALLISTIC_SPLIT = 0.7

        for i in range(steps + 1):
            t = i / steps
            pt = cubic_bezier(t, start, p1, p2, end)

            if t < BALLISTIC_SPLIT:
                # Ballistic: small fixed jitter
                sigma = 1.5
            else:
                # Corrective: jitter shrinks linearly as we approach target
                progress_in_corrective = (t - BALLISTIC_SPLIT) / (1 - BALLISTIC_SPLIT)
                remaining_dist = pt.distance_to(end)
                sigma = max(0.5, remaining_dist * 0.05)

            jitter_x = random.gauss(0, sigma)
            jitter_y = random.gauss(0, sigma)

            await self._page.mouse.move(pt.x + jitter_x, pt.y + jitter_y)

            # Velocity profile: fast in ballistic, slow in corrective
            if t < BALLISTIC_SPLIT:
                await asyncio.sleep(random.uniform(0.003, 0.010))
            else:
                await asyncio.sleep(random.uniform(0.015, 0.040))

        self._pos = end

    async def scroll(self, delta_y: int) -> None:
        """
        Scroll with variable speed — humans don't scroll at constant velocity.
        """
        ticks = random.randint(3, 7)
        per_tick = delta_y // ticks
        for _ in range(ticks):
            await self._page.mouse.wheel(0, per_tick + random.randint(-5, 5))
            await asyncio.sleep(random.uniform(0.08, 0.25))

    async def click(self, selector: str) -> None:
        element = await self._page.wait_for_selector(selector)
        box = await element.bounding_box()

        if box is None:
            raise ValueError(
                f"Element '{selector}' has no bounding box — "
                "it may be display:contents, detached, or inside a collapsed container."
            )

        target_x = box["x"] + box["width"] * random.uniform(0.2, 0.8)
        target_y = box["y"] + box["height"] * random.uniform(0.2, 0.8)

        await self.move(target_x, target_y)
        await asyncio.sleep(random.uniform(0.08, 0.35))
        await self._page.mouse.click(target_x, target_y)
        await asyncio.sleep(random.uniform(0.15, 0.45))

A Note on GIL and Event Loop Blocking

A reasonable question when looking at Bézier calculations running across dozens of parallel Playwright pages: does the CPU work cause event loop lag via the GIL?

The short answer is no, and the reasoning matters because the confusion is common.

The GIL prevents parallel execution across OS threads. asyncio is single-threaded cooperative concurrency — the GIL has essentially no relevance here unless you're explicitly mixing asyncio with threading. The actual concern in an asyncio context is event loop blocking: synchronous CPU code that runs long enough without yielding to starve other coroutines.

The numbers don't support that concern here. One complete Bézier step — four Point multiplications and three additions — takes roughly 2–5 microseconds on a modern CPU. Between every step there is an await asyncio.sleep(), which yields control back to the event loop. The CPU portion is three orders of magnitude faster than a single await page.mouse.move() call, which involves IPC to the browser process at ~1–5ms round-trip latency.

If the implementation had 200+ steps with no yield points between them, loop.run_in_executor() would be appropriate. At the current structure it would be premature optimisation with added complexity for no measurable gain. The real bottleneck is always the browser IPC, not the arithmetic.

Layer 3: Fingerprint Consistency — Coherence Matters More Than Spoofing

This is where most stealth implementations break down. A real browser has a coherent hardware identity across Canvas, WebGL, AudioContext, and font enumeration simultaneously. Patching one surface while leaving others inconsistent creates detectable contradictions that are arguably more suspicious than no patching.

Bug: The AudioContext Patch Mutated Shared Memory

The original implementation had a critical flaw:

// ❌ ORIGINAL — has a memory mutation bug
const origGetChannelData = AudioBuffer.prototype.getChannelData;
AudioBuffer.prototype.getChannelData = function(channel) {
    const data = origGetChannelData.call(this, channel);
    for (let i = 0; i < data.length; i++) {
        data[i] += (Math.random() - 0.5) * 1e-7;  // BUG
    }
    return data;
};

AudioBuffer.prototype.getChannelData returns a Float32Array that is a direct view into the AudioBuffer's internal ArrayBuffer — not a copy. Mutating data[i] modifies the AudioBuffer's underlying data in-place. Every subsequent call to getChannelData on the same buffer returns the already-mutated data, compounding the noise. More critically, any code that relies on the AudioBuffer's data after fingerprinting reads (Web Audio API nodes, audio worklets, Web Audio visualisers) will receive corrupted samples.

The same post correctly recommended working on offscreen copies for Canvas to avoid mutating page state. AudioContext needs the same treatment.

import json


async def patch_fingerprints(page, device_profile: dict):
    """
    Patch all hardware surfaces to a single consistent device profile.

    Consistency rule: every fingerprint surface must agree. A Windows/Chrome
    User-Agent with a Mesa OffScreen WebGL renderer is an instant flag. A
    Canvas patch that covers getImageData but not toDataURL creates a
    detectable contradiction. Patch all surfaces or patch none.

    Injection safety: never interpolate device_profile values directly into
    JavaScript via f-string. json.dumps() serialises Python values to valid,
    safely-escaped JS literals. float() enforces numeric type for noise,
    preventing injection there too. If device_profile comes from an external
    source (config file, API response), values containing quotes or JS syntax
    would become a code injection vulnerability without this.
    """
    vendor_js   = json.dumps(str(device_profile["webgl_vendor"]))
    renderer_js = json.dumps(str(device_profile["webgl_renderer"]))
    noise_js    = float(device_profile["canvas_noise"])

    await page.add_init_script(f"""
    (() => {{
        // ── WebGL ─────────────────────────────────────────────────────────
        // Vendor and renderer must match the claimed User-Agent OS/hardware.
        // Parameter 37445 = UNMASKED_VENDOR_WEBGL
        // Parameter 37446 = UNMASKED_RENDERER_WEBGL
        const getParameter = WebGLRenderingContext.prototype.getParameter;
        WebGLRenderingContext.prototype.getParameter = function(parameter) {{
            if (parameter === 37445) return {vendor_js};
            if (parameter === 37446) return {renderer_js};
            return getParameter.call(this, parameter);
        }};
        // Apply the same patch to WebGL2 — fingerprinters check both contexts.
        const getParameter2 = WebGL2RenderingContext.prototype.getParameter;
        WebGL2RenderingContext.prototype.getParameter = function(parameter) {{
            if (parameter === 37445) return {vendor_js};
            if (parameter === 37446) return {renderer_js};
            return getParameter2.call(this, parameter);
        }};

        // ── Canvas ────────────────────────────────────────────────────────
        // Patch all three read surfaces consistently. Patching only one is
        // worse than patching none — it creates a detectable contradiction.
        //   getImageData: pixel-level read
        //   toDataURL: base64 PNG/JPEG export
        //   toBlob: binary export (same pixel data, different delivery)
        const noise = {noise_js};

        function addNoise(imageData) {{
            for (let i = 0; i < imageData.data.length; i += 4) {{
                imageData.data[i]   += Math.floor(noise * Math.random() * 5);
                imageData.data[i+1] += Math.floor(noise * Math.random() * 5);
                imageData.data[i+2] += Math.floor(noise * Math.random() * 5);
                // Alpha channel untouched — modifying it breaks transparency.
            }}
            return imageData;
        }}

        const origGetImageData = CanvasRenderingContext2D.prototype.getImageData;
        CanvasRenderingContext2D.prototype.getImageData = function(x, y, w, h) {{
            return addNoise(origGetImageData.call(this, x, y, w, h));
        }};

        // For toDataURL and toBlob: work on an offscreen copy — never mutate
        // the original canvas. The original approach (putImageData back onto
        // the live canvas) would corrupt page state for anything that reads
        // its own canvas after calling toDataURL: game renderers, image
        // editors, captcha widgets.
        function withNoisedCopy(originalCanvas, fn) {{
            const offscreen = document.createElement('canvas');
            offscreen.width  = originalCanvas.width;
            offscreen.height = originalCanvas.height;
            const offCtx = offscreen.getContext('2d');
            offCtx.drawImage(originalCanvas, 0, 0);
            const d = offCtx.getImageData(0, 0, offscreen.width, offscreen.height);
            addNoise(d);
            offCtx.putImageData(d, 0, 0);
            return fn(offscreen);
        }}

        const origToDataURL = HTMLCanvasElement.prototype.toDataURL;
        HTMLCanvasElement.prototype.toDataURL = function(...args) {{
            if (this.getContext('2d')) {{
                return withNoisedCopy(this, c => origToDataURL.apply(c, args));
            }}
            return origToDataURL.apply(this, args);
        }};

        const origToBlob = HTMLCanvasElement.prototype.toBlob;
        HTMLCanvasElement.prototype.toBlob = function(callback, ...args) {{
            if (this.getContext('2d')) {{
                withNoisedCopy(this, c => origToBlob.call(c, callback, ...args));
                return;
            }}
            origToBlob.call(this, callback, ...args);
        }};

        // ── AudioContext ──────────────────────────────────────────────────
        // AudioContext fingerprinting renders a short buffer through the
        // browser's DSP pipeline and hashes the output. Different hardware
        // and OS audio stacks produce subtly different float results.
        //
        // FIX vs original: getChannelData returns a Float32Array that is a
        // *direct view* into the AudioBuffer's internal ArrayBuffer — not a
        // copy. Mutating data[i] in-place modifies the buffer permanently.
        // Every subsequent read gets double-noised, compounding the error.
        // Any downstream Web Audio nodes reading the same buffer get
        // corrupted samples.
        //
        // The fix: copy into a new Float32Array before modifying.
        // The copy is what gets returned; the original buffer is untouched.
        const origGetChannelData = AudioBuffer.prototype.getChannelData;
        AudioBuffer.prototype.getChannelData = function(channel) {{
            const original = origGetChannelData.call(this, channel);
            const copy = new Float32Array(original); // ← copy, not a view
            for (let i = 0; i < copy.length; i++) {{
                copy[i] += (Math.random() - 0.5) * 1e-7;
            }}
            return copy; // ← return the copy, original buffer is untouched
        }};
    }})();
    """)

Device Profile Consistency

The patches above are only effective if the profile is internally consistent. A "Windows 10 / Chrome 122" User-Agent must pair with:

DEVICE_PROFILES = {
    "windows_chrome_122": {
        "webgl_vendor":   "Google Inc. (NVIDIA)",
        "webgl_renderer": "ANGLE (NVIDIA, NVIDIA GeForce RTX 3060 Direct3D11 vs_5_0 ps_5_0, D3D11)",
        "canvas_noise":   0.4,
        "user_agent":     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "locale":         "en-US",
        "timezone":       "America/New_York",
    },
    "macos_chrome_122": {
        "webgl_vendor":   "Google Inc. (Apple)",
        "webgl_renderer": "ANGLE (Apple, Apple M2 Pro, OpenGL 4.1)",
        "canvas_noise":   0.4,
        "user_agent":     "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "locale":         "en-US",
        "timezone":       "America/Los_Angeles",
    },
}

Mismatches between any of these fields and the browser's actual reported values are detectable contradictions.

Layer 4: Honeypot Detection — The Silent Ban

Anti-bot systems plant invisible traps: zero-opacity links, off-screen form fields, elements styled display: none. Automated scripts interact with them; real users don't. Triggering a honeypot typically results in an immediate, silent ban — no 403, just poisoned data or session termination that's hard to distinguish from normal operation.

async def is_honeypot(page, selector: str) -> bool:
    """
    Returns True if the element looks like a honeypot.

    Uses a single evaluate() call instead of chained Playwright method calls
    (is_visible(), bounding_box(), evaluate()) — three separate browser
    round-trips that add up at scale. All checks run in one JS execution context.
    """
    element = await page.query_selector(selector)
    if not element:
        return True

    result = await element.evaluate("""el => {
        const box = el.getBoundingClientRect();
        const st  = window.getComputedStyle(el);
        return {
            display:    st.display,
            visibility: st.visibility,
            opacity:    parseFloat(st.opacity),
            width:      box.width,
            height:     box.height,
            x:          box.x,
            y:          box.y,
            inert:      el.inert,
            ariaHidden: el.getAttribute('aria-hidden'),
        };
    }""")

    if result["display"] == "none":      return True
    if result["visibility"] == "hidden": return True
    if result["opacity"] < 0.01:         return True
    if result["width"] < 1:              return True
    if result["height"] < 1:             return True
    if result["x"] < -100:              return True
    if result["y"] < -100:              return True
    if result["inert"]:                  return True

    return False


async def safe_click(hp: HumanPage, selector: str) -> None:
    if await is_honeypot(hp._page, selector):
        raise ValueError(f"Refusing to interact with suspected honeypot: {selector}")
    await hp.click(selector)

Layer 5: Proxy Management — Infrastructure, Not an Afterthought

A flawless browser fingerprint still fails on a flagged datacenter IP. Proxy management has three decisions that matter:

import asyncio
import random
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class ProxyConfig:
    host: str
    port: int
    username: str
    password: str
    country: str
    fail_count: int = 0
    total_requests: int = 0
    total_failures: int = 0

    @property
    def playwright_config(self) -> dict:
        return {
            "server":   f"http://{self.host}:{self.port}",
            "username": self.username,
            "password": self.password,
        }

    @property
    def subnet(self) -> str:
        """First three octets — useful for detecting provider-level issues."""
        return ".".join(self.host.split(".")[:3])


class ProxyRegistry:
    """
    Explicit proxy registry passed via dependency injection.

    Why not a global Singleton?

    1. Testability: a singleton bleeds state between test cases. With DI,
       each test instantiates its own registry with a known fixture set.

    2. asyncio event loop scope: asyncio.Lock() must be created inside the
       running event loop. A class-level lock created at import time binds to
       the wrong loop in pytest-asyncio and multiprocessing deployments,
       producing obscure "attached to a different loop" errors at runtime.

    3. Visibility: a hidden global is an invisible dependency. DI makes
       the dependency explicit in every function signature that uses it.
    """

    MAX_FAILS = 3

    def __init__(self, proxies: list[ProxyConfig]):
        self._proxies = proxies
        self._lock = asyncio.Lock()

    async def get(self, country: Optional[str] = None) -> ProxyConfig:
        async with self._lock:
            available = [
                p for p in self._proxies
                if p.fail_count < self.MAX_FAILS
                and (country is None or p.country == country)
            ]
            if not available:
                raise RuntimeError("Proxy pool exhausted.")

            # Weighted random selection — weight = 1 / (fail_count + 1).
            # Pure min(fail_count) routes 100% of traffic to one proxy until
            # it reaches fail_count=1, then the next, then the next. That's
            # sequential destruction, not load balancing.
            # Weighted random spreads load while still preferring healthier proxies.
            weights    = [1.0 / (p.fail_count + 1) for p in available]
            total      = sum(weights)
            r          = random.uniform(0, total)
            cumulative = 0.0
            chosen     = available[-1]
            for proxy, w in zip(available, weights):
                cumulative += w
                if r <= cumulative:
                    chosen = proxy
                    break

            # Increment at selection time, not at failure report time.
            # Without this, concurrent workers can all receive the same proxy
            # (fail_count=0 at get() time), then all report failure
            # simultaneously, jumping fail_count from 0 to N and quarantining
            # a healthy proxy.
            chosen.fail_count += 1
            chosen.total_requests += 1
            return chosen

    async def report(self, proxy: ProxyConfig, success: bool) -> None:
        async with self._lock:
            if success:
                proxy.fail_count = max(0, proxy.fail_count - 1)
            else:
                proxy.total_failures += 1
                if proxy.fail_count >= self.MAX_FAILS:
                    pass  # already quarantined; fail_count was set at get()

    def subnet_failure_rates(self) -> dict[str, float]:
        """
        Returns failure rate per /24 subnet.

        Used to detect when a proxy provider silently routes a batch of IPs
        through a flagged datacenter range — a failure mode that looks like
        general degradation until you correlate by subnet.
        """
        subnets: dict[str, list[ProxyConfig]] = {}
        for p in self._proxies:
            subnets.setdefault(p.subnet, []).append(p)

        return {
            subnet: sum(p.total_failures for p in proxies) / max(1, sum(p.total_requests for p in proxies))
            for subnet, proxies in subnets.items()
        }

Three decisions that matter in production:

Sticky sessions — same exit IP for the full logical session, not per-request rotation. Per-request rotation trips geo-consistency checks.
Geo-matching — proxy country must match locale and timezone_id in the browser context. Mismatches are trivial signals.
Subnet-level monitoring — per-proxy logging lets you detect when a provider silently routes a subset of IPs through a flagged datacenter. Without it, the failure looks like general degradation for days.

Layer 6: Retry Logic — Transient Failures Are Not Failures

No production system runs without transient failures: proxy timeouts, soft rate-limits, transient blocks. The response is structured retry with backoff, not except: pass.

import asyncio
import random
from typing import Callable, TypeVar

T = TypeVar("T")


async def with_retry(
    fn: Callable,
    *args,
    max_attempts: int = 3,
    base_delay: float = 2.0,
    max_delay: float = 60.0,
    retryable_exceptions: tuple = (OSError, asyncio.TimeoutError),
    **kwargs,
) -> T:
    """
    Exponential backoff with full jitter.

    Pure exponential backoff causes thundering herd: all workers that failed
    at the same moment retry at the same moment. Full jitter
    (random.uniform(0, cap)) spreads retries evenly across the window.

    Why not retryable_exceptions=(Exception,) as the default?
    Exception catches KeyboardInterrupt, SystemExit, and MemoryError —
    none of which should be silently retried. (OSError, asyncio.TimeoutError)
    covers transient network failures. Opt into broader coverage explicitly.

    Note: this function handles *per-request* transients — a proxy timeout,
    a momentary rate-limit response. It does not handle "the entire target
    domain is currently blocking us." That's Layer 7: circuit breaker.
    """
    for attempt in range(max_attempts):
        try:
            return await fn(*args, **kwargs)
        except retryable_exceptions as e:
            if attempt == max_attempts - 1:
                # Log before re-raising — without this the caller receives an
                # exception with no indication that N retries preceded it.
                print(
                    f"[Retry] All {max_attempts} attempts failed. "
                    f"Final error: {type(e).__name__}: {e}"
                )
                raise
            cap   = min(max_delay, base_delay * (2 ** attempt))
            delay = random.uniform(0, cap)
            print(f"[Retry] Attempt {attempt + 1}/{max_attempts} failed: {e}. "
                  f"Waiting {delay:.1f}s.")
            await asyncio.sleep(delay)

Layer 7: Circuit Breaker — The Layer the Original Skipped

The original post acknowledged the gap: "with_retry handles per-request transients; it doesn't handle 'this entire target site is currently blocking us.'" This section fills it — including an architectural constraint that matters the moment you scale beyond a single process.

A retry loop and a circuit breaker are different abstractions solving different failure modes:

	Retry	Circuit Breaker
Scope	Single request	Entire domain
Failure mode	Transient (proxy timeout)	Sustained (site-wide block)
Response	Wait and retry	Stop draining the proxy pool
Recovery	Automatic	Automatic, via HALF_OPEN probe

The In-Process Version (Single Container)

import asyncio
import time
from collections import deque
from enum import Enum
from typing import Callable, TypeVar

T = TypeVar("T")


class CircuitState(Enum):
    CLOSED    = "closed"     # Normal operation — all requests flow through
    OPEN      = "open"       # Sustained failure — all requests rejected
    HALF_OPEN = "half_open"  # Recovery probe — limited requests allowed


class CircuitBreaker:
    """
    Domain-level circuit breaker with sliding window block-rate tracking.

    State machine:
        CLOSED → OPEN:      block_rate > threshold over window_seconds
        OPEN → HALF_OPEN:   recovery_timeout seconds have elapsed
        HALF_OPEN → CLOSED: probe_limit consecutive successes
        HALF_OPEN → OPEN:   any failure during probe

    ⚠️  SCOPE LIMITATION: this implementation lives in process memory.
    It works correctly for a single container or a single long-running script.
    If you run multiple workers — Docker Compose, K8s pods, parallel EC2
    instances — each pod maintains its own independent breaker state.
    Pod A opens its breaker; pods B, C, D know nothing about it and continue
    burning through the proxy pool against a domain that's blocking all of
    them. See DistributedCircuitBreaker below for the multi-pod solution.
    """

    def __init__(
        self,
        domain: str,
        threshold: float = 0.20,     # open if block rate exceeds 20%
        window_seconds: int = 600,   # 10-minute sliding window
        recovery_timeout: int = 300, # wait 5 minutes before probing
        probe_limit: int = 3,        # consecutive successes to close
    ):
        self.domain = domain
        self.threshold = threshold
        self.window_seconds = window_seconds
        self.recovery_timeout = recovery_timeout
        self.probe_limit = probe_limit

        self._state = CircuitState.CLOSED
        self._lock = asyncio.Lock()
        self._events: deque[tuple[float, bool]] = deque()
        self._opened_at: float = 0.0
        self._probe_successes: int = 0

    def _prune_window(self, now: float) -> None:
        cutoff = now - self.window_seconds
        while self._events and self._events[0][0] < cutoff:
            self._events.popleft()

    def _block_rate(self, now: float) -> float:
        self._prune_window(now)
        if not self._events:
            return 0.0
        failures = sum(1 for _, ok in self._events if not ok)
        return failures / len(self._events)

    async def allow(self) -> bool:
        async with self._lock:
            now = time.monotonic()
            if self._state == CircuitState.CLOSED:
                return True
            if self._state == CircuitState.OPEN:
                if now - self._opened_at >= self.recovery_timeout:
                    self._state = CircuitState.HALF_OPEN
                    self._probe_successes = 0
                    return True
                return False
            return True  # HALF_OPEN: let probe through

    async def record(self, success: bool) -> None:
        async with self._lock:
            now = time.monotonic()
            self._events.append((now, success))

            if self._state == CircuitState.HALF_OPEN:
                if success:
                    self._probe_successes += 1
                    if self._probe_successes >= self.probe_limit:
                        self._state = CircuitState.CLOSED
                        log.info("circuit.closed", domain=self.domain)
                else:
                    self._state = CircuitState.OPEN
                    self._opened_at = now
                    log.warning("circuit.reopened", domain=self.domain)
                return

            if self._state == CircuitState.CLOSED and not success:
                rate = self._block_rate(now)
                if rate > self.threshold:
                    self._state = CircuitState.OPEN
                    self._opened_at = now
                    log.warning(
                        "circuit.opened",
                        domain=self.domain,
                        block_rate=round(rate, 3),
                        threshold=self.threshold,
                    )

    def status(self) -> dict:
        now = time.monotonic()
        return {
            "domain":      self.domain,
            "state":       self._state.value,
            "block_rate":  round(self._block_rate(now), 3),
            "window_size": len(self._events),
            "opened_at":   self._opened_at if self._state != CircuitState.CLOSED else None,
        }


class CircuitOpenError(Exception):
    pass


class CircuitBreakerRegistry:
    """One breaker per domain, created on first access."""

    def __init__(self, **breaker_defaults):
        self._breakers: dict[str, CircuitBreaker] = {}
        self._defaults = breaker_defaults

    def get(self, domain: str) -> CircuitBreaker:
        if domain not in self._breakers:
            self._breakers[domain] = CircuitBreaker(domain=domain, **self._defaults)
        return self._breakers[domain]

    def all_statuses(self) -> list[dict]:
        return [b.status() for b in self._breakers.values()]

The Distributed Version (Multi-Pod / K8s)

The in-process breaker has a hard architectural constraint: state lives in one process's memory. In a horizontal deployment — K8s pods, Docker Compose workers, parallel EC2 instances — each pod observes only its own traffic. Pod A trips its breaker; pods B through D keep hammering the same blocked domain, burning proxy IPs until their individual breakers catch up. By that point you may have lost the entire proxy subnet.

The fix is to centralise state. Redis is the standard tool for this: it's fast enough for per-request writes (sub-millisecond), supports atomic operations, and has a TTL mechanism that handles cleanup automatically.

# pip install redis[hiredis]
import redis.asyncio as aioredis
import time


class DistributedCircuitBreaker:
    """
    Circuit breaker backed by Redis sorted sets. Shares state across all pods.

    Key schema (all keys scoped to domain):
        cb:events:{domain}    — sorted set, score=timestamp, member="ok:{ts}" or "fail:{ts}"
        cb:state:{domain}     — string: "closed" | "open" | "half_open"
        cb:opened_at:{domain} — float timestamp of last OPEN transition
        cb:probe_ok:{domain}  — integer: consecutive probe successes in HALF_OPEN

    Why sorted set instead of a simple counter:
        A counter gives you total failures — you lose the sliding window.
        ZREMRANGEBYSCORE on a sorted set prunes stale events atomically,
        giving you an accurate block rate over the last N seconds without
        a separate cleanup process.

    Redis as SPOF:
        If Redis becomes unavailable, allow() returns True (fail-open).
        The alternative — fail-closed — stops all scraping across all pods
        on a Redis hiccup. In this workload, that's worse than a brief period
        of unguarded requests. If you need fail-closed, invert the except clause.

    HA:
        Use Redis Sentinel (3-node) or Redis Cluster for production.
        A standalone Redis instance is itself a single point of failure.
    """

    def __init__(
        self,
        redis_client: aioredis.Redis,
        domain: str,
        threshold: float = 0.20,
        window_seconds: int = 600,
        recovery_timeout: int = 300,
        probe_limit: int = 3,
    ):
        self.r = redis_client
        self.domain = domain
        self.threshold = threshold
        self.window_seconds = window_seconds
        self.recovery_timeout = recovery_timeout
        self.probe_limit = probe_limit

        self._key_events   = f"cb:events:{domain}"
        self._key_state    = f"cb:state:{domain}"
        self._key_opened   = f"cb:opened_at:{domain}"
        self._key_probe_ok = f"cb:probe_ok:{domain}"

    async def allow(self) -> bool:
        try:
            state = await self.r.get(self._key_state)
            state = (state or b"closed").decode()

            if state == "closed":
                return True

            if state == "open":
                opened_at = float(await self.r.get(self._key_opened) or 0)
                if time.time() - opened_at >= self.recovery_timeout:
                    # Atomic transition to half_open.
                    # GETSET is deprecated in Redis 6.2+; use SET ... GET instead.
                    await self.r.set(self._key_state, "half_open")
                    await self.r.set(self._key_probe_ok, 0)
                    log.info("circuit.half_open", domain=self.domain, source="redis")
                    return True
                return False

            # half_open: let probe requests through
            return True

        except aioredis.RedisError as e:
            # Redis unavailable — fail open rather than stop all pods.
            log.error("circuit.redis_error", domain=self.domain, error=str(e))
            return True

    async def record(self, success: bool) -> None:
        try:
            now = time.time()
            pipe = self.r.pipeline()

            # Maintain sliding window: prune old events, add new one.
            # Pipeline batches both commands into a single round-trip.
            member = f"{'ok' if success else 'fail'}:{now}"
            pipe.zremrangebyscore(self._key_events, 0, now - self.window_seconds)
            pipe.zadd(self._key_events, {member: now})
            pipe.expire(self._key_events, self.window_seconds * 2)
            await pipe.execute()

            state = (await self.r.get(self._key_state) or b"closed").decode()

            if state == "half_open":
                if success:
                    count = await self.r.incr(self._key_probe_ok)
                    if count >= self.probe_limit:
                        await self.r.set(self._key_state, "closed")
                        log.info("circuit.closed", domain=self.domain, source="redis")
                else:
                    await self.r.set(self._key_state, "open")
                    await self.r.set(self._key_opened, now)
                    log.warning("circuit.reopened", domain=self.domain, source="redis")
                return

            if state == "closed" and not success:
                # Compute block rate from the sliding window we just updated.
                all_members = await self.r.zrange(self._key_events, 0, -1)
                if all_members:
                    failures = sum(
                        1 for m in all_members if m.decode().startswith("fail")
                    )
                    rate = failures / len(all_members)
                    if rate > self.threshold:
                        await self.r.set(self._key_state, "open")
                        await self.r.set(self._key_opened, now)
                        log.warning(
                            "circuit.opened",
                            domain=self.domain,
                            block_rate=round(rate, 3),
                            source="redis",
                        )

        except aioredis.RedisError as e:
            log.error("circuit.redis_record_error", domain=self.domain, error=str(e))

    async def status(self) -> dict:
        try:
            state = (await self.r.get(self._key_state) or b"closed").decode()
            all_members = await self.r.zrange(self._key_events, 0, -1)
            failures = sum(1 for m in all_members if m.decode().startswith("fail"))
            block_rate = failures / len(all_members) if all_members else 0.0
            return {
                "domain":      self.domain,
                "state":       state,
                "block_rate":  round(block_rate, 3),
                "window_size": len(all_members),
                "backend":     "redis",
            }
        except aioredis.RedisError:
            return {"domain": self.domain, "state": "unknown", "backend": "redis"}

Choosing between the two implementations:

Deployment	Use
Single process / single container	`CircuitBreaker` (in-memory)
Docker Compose with multiple workers	`DistributedCircuitBreaker`
K8s pods, any horizontal scaling	`DistributedCircuitBreaker`

Both share the same allow() / record() interface, so the calling code at the request site is identical. The only difference is construction: one takes no external dependency, the other takes a redis.asyncio.Redis client.

Usage at the request site (works with both implementations):

async def fetch_price(
    url: str,
    hp: HumanPage,
    proxy: ProxyConfig,
    proxy_registry: ProxyRegistry,
    breaker: CircuitBreaker | DistributedCircuitBreaker,
) -> dict:
    if not await breaker.allow():
        raise CircuitOpenError(
            f"Circuit OPEN for {breaker.domain} — "
            "skipping to avoid draining proxy pool."
        )

    success = False
    try:
        await hp._page.goto(url, wait_until="domcontentloaded", timeout=30_000)
        data = await hp._page.evaluate("() => window.__PRICE_DATA__")
        success = True
        await proxy_registry.report(proxy, success=True)
        return data
    except Exception:
        await proxy_registry.report(proxy, success=False)
        raise
    finally:
        await breaker.record(success)

Layer 8: Observability — `print()` Is Not a Monitoring Strategy

The original post noted that "print() to stdout is not observability" — and then didn't show what observability actually looks like. Here it is.

The difference matters operationally. print() tells you something happened. Structured logging tells you what happened, to which resource, in which context, with which outcome, in a format that's queryable. The incident in Month 11 (proxy subnet flagged) took three days to diagnose partly because correlating proxy host to block rate required manual log parsing. With structured logging it's a one-line query.

Structured Logging with structlog

pip install structlog

import structlog
import logging
import sys


def configure_logging(level: str = "INFO") -> None:
    """
    Configure structlog for structured JSON output in production,
    human-readable output in development.

    structlog processors run in order. Each processor receives the event
    dict and returns a modified version. The final processor (JSONRenderer
    or ConsoleRenderer) serialises the dict to a string.
    """
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,          # thread-local context
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.ExceptionRenderer(),
            structlog.processors.JSONRenderer(),              # → JSON in prod
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, level.upper(), logging.INFO)
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(sys.stdout),
    )


# Module-level logger — bind context progressively
log = structlog.get_logger()

Binding session context:

async def run_session(
    url: str,
    proxy: ProxyConfig,
    circuit_registry: CircuitBreakerRegistry,
    session_id: str,
) -> dict:
    from urllib.parse import urlparse
    domain = urlparse(url).netloc

    # Bind fields that apply to every log call in this session
    session_log = log.bind(
        session_id=session_id,
        domain=domain,
        proxy_host=proxy.host,
        proxy_subnet=proxy.subnet,
        proxy_country=proxy.country,
    )

    session_log.info("session.start", url=url)

    start = time.monotonic()
    try:
        result = await fetch_price(url, ...)
        duration_ms = int((time.monotonic() - start) * 1000)
        session_log.info(
            "session.success",
            duration_ms=duration_ms,
            price=result.get("price"),
        )
        return result
    except CircuitOpenError as e:
        session_log.warning("session.circuit_open", reason=str(e))
        raise
    except Exception as e:
        duration_ms = int((time.monotonic() - start) * 1000)
        session_log.error(
            "session.failure",
            duration_ms=duration_ms,
            error_type=type(e).__name__,
            error=str(e),
        )
        raise

What the output looks like:

{"event": "session.start", "session_id": "s_1a2b3c", "domain": "example.com", "proxy_host": "192.168.1.10", "proxy_subnet": "192.168.1", "proxy_country": "US", "url": "https://example.com/product/123", "timestamp": "2024-03-15T14:23:01.123Z", "log_level": "info"}
{"event": "session.success", "session_id": "s_1a2b3c", "domain": "example.com", "proxy_host": "192.168.1.10", "proxy_subnet": "192.168.1", "duration_ms": 3241, "price": 49.99, "timestamp": "2024-03-15T14:23:04.364Z", "log_level": "info"}

Querying for the Month 11 incident in seconds:

# Block rate by subnet (what took 3 days to find manually)
cat session.log | jq -s '
  group_by(.proxy_subnet) |
  map({
    subnet: .[0].proxy_subnet,
    total: length,
    failures: map(select(.event == "session.failure")) | length,
    block_rate: (map(select(.event == "session.failure")) | length) / length
  })
' | jq 'sort_by(-.block_rate)'

Circuit Breaker Status Logging

async def log_circuit_status(circuit_registry: CircuitBreakerRegistry) -> None:
    """Log circuit breaker states periodically."""
    for status in circuit_registry.all_statuses():
        if status["state"] != "closed" or status["block_rate"] > 0.05:
            log.warning(
                "circuit.status",
                **status,
            )

Layer 9: Storage — SQLite, Postgres, and When the Answer Changes

The original post mentioned SQLite and WAL mode briefly. This section makes the decision criteria explicit.

SQLite in WAL Mode: What It Actually Handles

import aiosqlite
import asyncio
from contextlib import asynccontextmanager


async def init_db(path: str) -> aiosqlite.Connection:
    db = await aiosqlite.connect(path)
    # WAL mode: readers don't block writers, writers don't block readers.
    # Critical for a pattern where one writer task feeds many reader tasks.
    await db.execute("PRAGMA journal_mode=WAL")
    # synchronous=NORMAL: flush on checkpoint, not every write.
    # Faster than FULL; safe for crash recovery with WAL.
    await db.execute("PRAGMA synchronous=NORMAL")
    # Larger page cache reduces I/O for read-heavy workloads.
    await db.execute("PRAGMA cache_size=-65536")  # 64MB
    await db.execute("PRAGMA foreign_keys=ON")
    await db.execute("""
        CREATE TABLE IF NOT EXISTS price_snapshots (
            id           INTEGER PRIMARY KEY,
            sku          TEXT    NOT NULL,
            domain       TEXT    NOT NULL,
            price        REAL,
            currency     TEXT,
            session_id   TEXT    NOT NULL,
            proxy_host   TEXT    NOT NULL,
            status_code  INTEGER NOT NULL,
            duration_ms  INTEGER,
            captured_at  TEXT    NOT NULL DEFAULT (datetime('now'))
        )
    """)
    await db.execute("""
        CREATE INDEX IF NOT EXISTS idx_sku_captured
        ON price_snapshots(sku, captured_at DESC)
    """)
    await db.commit()
    return db


class WriteQueue:
    """
    Funnel all writes through a single asyncio task.

    SQLite's WAL mode handles concurrent reads cleanly. Concurrent writes
    still serialise at the SQLite level — but if multiple coroutines call
    db.execute() concurrently, they also contend on the aiosqlite connection
    object itself. A single writer task eliminates that contention entirely
    and makes write ordering predictable.

    maxsize is not optional:
        asyncio.Queue() with no maxsize is an unbounded queue. If disk I/O
        stalls — a slow fsync, a cloud VM with noisy-neighbour I/O — the
        worker falls behind while producers keep enqueuing. The queue grows
        without limit until the container is OOM-killed. There is no warning;
        the process just disappears.

        maxsize=500 creates backpressure: write() will await when the queue
        is full, slowing producers down rather than accumulating memory.
        The right maxsize depends on your write latency and worker count —
        measure under realistic load, not in dev.

    write_timeout controls how long write() will wait for a queue slot before
    giving up. A timeout is preferable to blocking forever: if the writer
    task dies (unhandled exception in _worker), an unbounded wait would
    silently hang all producer coroutines.
    """

    def __init__(
        self,
        db: aiosqlite.Connection,
        maxsize: int = 500,
        write_timeout: float = 10.0,
    ):
        self._db            = db
        self._write_timeout = write_timeout
        self._dropped       = 0
        # maxsize enforces backpressure — producers block instead of OOM-ing.
        self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
        self._task = asyncio.create_task(self._worker())

    async def _worker(self):
        while True:
            item = await self._queue.get()
            if item is None:
                break
            sql, params, fut = item
            try:
                await self._db.execute(sql, params)
                await self._db.commit()
                fut.set_result(None)
            except Exception as e:
                fut.set_exception(e)
            finally:
                self._queue.task_done()

    async def write(self, sql: str, params: tuple) -> None:
        loop = asyncio.get_running_loop()
        fut  = loop.create_future()
        try:
            # wait_for gives the worker time to drain a slot before we give up.
            # put_nowait would raise QueueFull immediately with no grace period.
            await asyncio.wait_for(
                self._queue.put((sql, params, fut)),
                timeout=self._write_timeout,
            )
        except asyncio.TimeoutError:
            self._dropped += 1
            log.error(
                "write_queue.backpressure_timeout",
                queue_size=self._queue.qsize(),
                dropped_total=self._dropped,
                write_timeout=self._write_timeout,
            )
            raise  # propagate — the caller decides whether to retry or skip

        await fut

    @property
    def queue_size(self) -> int:
        return self._queue.qsize()

    async def close(self):
        await self._queue.put(None)
        await self._task

When SQLite Stops Being the Right Answer

SQLite with WAL handles the pricing-intelligence workload comfortably: ~4,000 SKUs, one write per SKU per run cycle, many concurrent readers. At what point does the answer change?

Benchmark: synthetic write throughput (single writer task, WAL mode)

Batch size	Writes/sec	p99 latency
1 (unbatched)	~800	~1.8ms
10	~4,200	~3.1ms
100	~12,000	~11ms

These numbers are from a local NVMe SSD. On network-attached storage or a cloud VM with shared I/O, divide by 3–5.

Switch to Postgres when any of these are true:

Write throughput sustained above ~5,000/sec. SQLite's single-writer model becomes a bottleneck. Postgres's MVCC handles concurrent writes without serialising them.
Multiple machines need to write simultaneously. SQLite is a single-file database — not usable across a network without a proxy (Litestream, rqlite). Postgres is native client-server.
You need row-level locking for concurrent updates. SQLite locks at the database level for writes. If you're updating price records in place rather than inserting new snapshots, concurrent updates will serialise hard.
Your team's incident tooling expects SQL-compatible infrastructure. EXPLAIN ANALYZE, pg_stat_statements, and integration with tools like Datadog APM are Postgres-native. Adding this observability to SQLite requires custom instrumentation.
Audit trail requirements. If regulators or clients need immutable write history, Postgres + logical replication is the standard answer.

At the 4,000 SKU scale in this post: SQLite is correct. Don't optimise prematurely. Measure write throughput under real load before deciding.

What Actually Happened in Production

I'll be direct about the metrics, because "0 incidents, 99.x% uptime" framing is useless.

18 months, ~4,000 SKUs. The real picture:

Incident 1 — Month 4: Partial Canvas Patch Detected

Cloudflare pushed an update to their Managed Challenge. Our canvas noise patch covered getImageData but not toDataURL or toBlob. Detection rate jumped from near-zero to ~40% of sessions within 48 hours.

That figure comes from HTTP response code logging in the SQLite session table — we tracked status_code per run and could query block rate as COUNT(*) WHERE status=403 / COUNT(*) per hour. Diagnosis took a day. The fix (patching all three canvas surfaces) took two hours.

Lesson: partial fingerprint patching is worse than none — it creates a detectable contradiction. Patch all surfaces or patch none.

Incident 2 — Month 11: Silent Proxy Subnet Compromise

One residential proxy provider silently began routing a subset of IPs through a datacenter range. Cloudflare's IP reputation scoring flagged those exit nodes. We saw a gradual uptick in 403s over three days before correlating it to the proxy subnet.

We found it because we were logging proxy_host per request and could GROUP BY subnet in the session table. Without per-proxy logging, this would have looked like general degradation indefinitely.

Lesson: instrument at the proxy level, not just the request level. The subnet_failure_rates() method in the ProxyRegistry and the proxy_subnet field in structured logs address this directly.

What the incidents exposed architecturally:

Both incidents were recoverable because the failures were observable. What I didn't have — and should have — was the circuit breaker (now Layer 7). Both incidents involved the system continuing to burn through proxy pool capacity against a domain that was blocking us at scale. with_retry handles per-request transients; it doesn't stop draining the pool when the entire target is blocking you. Those are different failure modes and need different responses.

On the "0 manual interventions" claim in the original draft: true if "intervention" means a human sitting down to debug a mid-run block. The retry and rotation logic handled transient failures automatically. Not true if you count the two incidents above, which both required code changes and redeployments. Write your metrics definitions before you write your metrics.

The Full Stack

┌─────────────────────────────────────────────────────────────┐
│                    Orchestrator (asyncio)                     │
├──────────────┬──────────────────┬───────────────────────────┤
│  TLS Clean   │  Stealth         │  Fingerprint              │
│  Baseline    │  Patching        │  Consistency              │
│  (Camoufox / │  (playwright-    │  (Canvas/WebGL/Audio,     │
│  rebrowser)  │  stealth)        │  all surfaces, no leaks)  │
├──────────────┴──────────────────┴───────────────────────────┤
│  Behavioural Simulation (HumanPage)                          │
│  Cubic Bézier · Fitts's Law step count · Phase-split jitter  │
├──────────────────────────────────────────────────────────────┤
│  Honeypot Detection · single evaluate() · 7 visibility checks│
├──────────────────────────────────────────────────────────────┤
│  Retry (per-request, full jitter) │ Circuit Breaker (domain) │
│                                   │ In-process OR Redis-backed│
├──────────────────────────────────────────────────────────────┤
│  ProxyRegistry (DI) · Weighted · Geo-matched · Subnet audit  │
├──────────────────────────────────────────────────────────────┤
│  Observability: structlog JSON · session_id · proxy_subnet   │
│  Circuit status · block_rate queryable in < 1 minute         │
├──────────────────────────────────────────────────────────────┤
│  Storage: SQLite WAL + WriteQueue (bounded, backpressure)    │
│  Switch-to-Postgres criteria defined                         │
└──────────────────────────────────────────────────────────────┘

Key Takeaways

TLS fingerprinting kills you before HTTP. JA4 captures extension structure, ALPN ordering, and signature algorithms — not just cipher suites. Rotating User-Agents while using requests or vanilla Playwright doesn't help. Start with a clean baseline: Camoufox, rebrowser-patches, or curl_cffi.
Partial fingerprint patching is actively harmful. All three canvas surfaces (getImageData, toDataURL, toBlob) plus AudioContext must be patched consistently. WebGL2 needs the same patch as WebGL. Partial patches create detectable contradictions.
The AudioContext patch must copy the buffer. getChannelData returns a view into the AudioBuffer's internal ArrayBuffer. Mutating it in-place corrupts the buffer permanently. new Float32Array(original) creates a copy; return the copy.
Mouse movement physics need to match the model you're claiming. Real cubic Bézier requires four control points. Fitts's Law means step count should scale with distance. The ballistic phase (0–70%) has minimal correction; the corrective phase (70–100%) has increasing Gaussian scatter as the cursor homes in.
Behavioural simulation is session-level. Per-page state, not a global dict. Global state in a concurrent worker pool is a race condition.
Bézier arithmetic doesn't block the event loop. The GIL is irrelevant in single-threaded asyncio. One Bézier step takes ~2–5µs; a single browser IPC call takes ~1–5ms. The real bottleneck is always the browser. run_in_executor() here is premature optimisation.
Honeypots cause silent bans. Consolidate visibility checks into a single evaluate() call. Add inert to the check list.
Proxy selection needs weighted random, not always-min. Increment fail_count at selection time, not failure time. Log at the subnet level — subnet-level failure rates diagnose provider issues that look like general degradation.
Retry and circuit breakers are different abstractions. with_retry handles per-request transients. A circuit breaker handles sustained domain-level blocks. Build both.
In-process circuit breakers don't work in horizontal deployments. In K8s or Docker Compose with multiple workers, each pod's breaker is invisible to the others. Use DistributedCircuitBreaker backed by Redis whenever you scale beyond a single container. Use Redis Sentinel or Cluster — standalone Redis is itself a SPOF.
Unbounded queues are silent OOM bombs. asyncio.Queue(maxsize=500) with asyncio.wait_for() creates backpressure: producers slow down instead of accumulating memory until the container is killed.
assert is not runtime validation, Exception is not a retry target. Use explicit raise, narrow your retryable exception set, log before the final re-raise.
print() is not observability. Structured logging (structlog) with bound context makes incidents diagnosable in minutes. The Month 11 subnet incident took three days with ad-hoc logging; with subnet_failure_rates() and structured logs it's a one-command query.
SQLite is right at 4,000 SKUs. Define the switch criteria before you need them. WAL + single writer task handles the workload. Postgres becomes correct at sustained ~5,000 writes/sec, multiple concurrent writers across machines, or when your team's incident tooling expects it.

The detection systems are good and getting better. The gap you're maintaining is operational: how fast can you diagnose a new detection vector, measure its scope, and ship a fix. Everything in this post exists to make that loop shorter.

Working through a specific anti-bot system or detection layer? Drop it in the comments.

Tags: #python #automation #playwright #webdev #scraping #architecture #devops #observability

Python Internals I Wish I Knew

Adilet Akmatov — Sat, 21 Mar 2026 19:57:11 +0000

I came to Python from business — 6 years running operations, then 2 years ago switched to building automation tools. Bots, scrapers, Telegram integrations, desktop tools packaged with PyInstaller that clients use on Windows machines in retail stores across Central Asia.

For most of that time, I thought I understood Python pretty well.

Then I hit a bug I couldn't explain.

The Bug That Started All of This

I was building a high-performance web scraper — processing around 40,000 HTML pages per run, concurrently, using Playwright and BeautifulSoup. Production numbers looked fine at first. Then I noticed the process was taking 1.2 seconds per document. Over 40k documents, that becomes 13+ hours. Not acceptable.

I profiled it. The bottleneck was BeautifulSoup parsing — pure Python HTML traversal on every page. So I did what seemed logical: rewrote the hot path as a C extension using libxml2.

Result: 0.09 seconds per document. 13x speedup.

But why exactly did that work? I knew that it worked — Python is slow, C is fast. That's obvious. What I didn't understand was the mechanism. What was Python doing that made it 13x slower? What did the C extension eliminate?

That gap bothered me enough that I cloned the CPython repo and started reading.

Three months later, I understood. And it changed how I think about every Python program I write.

This post is what I found.

CPython Is Just a C Program
The Life of a Script: From Source to Execution
PyObject — The Real Cost of x = []
Memory Management: Three Tiers
Reference Counting — The Deterministic Heart
Garbage Collector — When Counters Aren't Enough
The GIL — The Most Misunderstood Thing in Python
The Eval Loop — The Heart of the Machine
C Extensions — What Actually Made My Scraper 13x Faster
PyPy, Python 3.13 Free-Threading, and What's Coming
What Changed in My Actual Code

Quick Mental Model

Before we dig in — here's the whole thing on one screen:

Layer	What It Does	Lives In
Compiler	Turns `.py` into bytecode	`Python/compile.c`
Virtual Machine	Executes bytecode instruction by instruction	`Python/ceval.c`
Object Model	Everything is a `PyObject` with a type and a refcount	`Objects/`
Memory	Reference Counting + cyclic GC	`Objects/obmalloc.c`
The Constraint	GIL — one thread executes bytecode at a time	`Python/ceval_gil.c`

.py → [Lexer] → [Parser] → AST → [Compiler] → Bytecode → [Eval Loop] → result
                                                                  ↕
                                                      PyObject / pymalloc / GC

Five layers. That's Python. Now let's open each one.

1. CPython Is Just a C Program

When I first opened the CPython repo, I expected something complicated and exotic. What I found was surprisingly clean C code with good comments and consistent naming.

When you run python script.py, you're launching CPython — an interpreter written in plain C (C11 standard per PEP 7). No C++, no JVM, no Rust. Just C. That decision was deliberate — maximum portability, minimum complexity, maximum contributor accessibility.

The directory structure maps directly to what the runtime does:

CPython repository (github.com/python/cpython):
├── Python/       ← Compiler, Eval Loop, builtins
├── Objects/      ← int, str, list, dict as C structs
├── Modules/      ← C-side of stdlib (json, re, os...)
├── Include/      ← Public C API headers for extensions
├── Lib/          ← Python-side of stdlib
└── Misc/         ← Docs, release notes

My recommendation: open Objects/listobject.c and read it. The actual implementation of Python lists. It's about 2,000 lines of very readable C. Once you've done that, Python lists stop being magic and start being a thing you understand. Everything else in the stdlib becomes easier to reason about after that.

2. The Life of a Script

You type python script.py and hit enter. Your source code goes through six stages before anything executes:

Source (.py)
    │
    ▼
[Lexer / Tokenizer]        ← tokenize.py / tokenizer.c
    │  Text → tokens: NAME, NUMBER, OP...
    ▼
[Parser]                   ← Parser/parser.c (PEG since Python 3.9)
    │  Tokens → AST (Abstract Syntax Tree)
    ▼
[AST Optimizer]            ← Python/ast_opt.c
    │  Constant folding: TIMEOUT = 60 * 60 * 24 → 86400 at compile time
    ▼
[Compiler]                 ← Python/compile.c
    │  AST → bytecode instructions
    ▼
[Bytecode cache (.pyc)]    ← __pycache__/script.cpython-312.pyc
    │  Cached to disk; skips steps 1–4 on next run
    ▼
[Eval Loop (ceval.c)]      ← Python/ceval.c
    │  Executes bytecode instruction by instruction
    ▼
[Result]

You can inspect what CPython produces from your code:

import dis

def add(a, b):
    return a + b

dis.dis(add)

Output:

  2           0 RESUME                   0

  3           2 LOAD_FAST                0 (a)
              4 LOAD_FAST                1 (b)
              6 BINARY_OP               0 (+)
             10 RETURN_VALUE

Four instructions for a + b. Each is a 1-byte opcode plus an argument. This is the thing Python is actually running — not your source code. When I started using dis regularly on my scrapers and bot code, I started seeing why some things were slow in ways profiling alone didn't reveal.

The .pyc file in __pycache__? Just a serialized code object packed with Python's marshal module. Python checks timestamp and version magic on startup — unchanged file means steps 1–4 are completely skipped.

3. PyObject — The Real Cost of x = []

This is the thing that changed how I write code more than anything else.

In Python, everything is an object. Under the hood, "object" means one specific thing: a C struct called PyObject.

// Include/object.h
typedef struct _object {
    Py_ssize_t    ob_refcnt;   // reference counter
    PyTypeObject *ob_type;     // pointer to the type (int, str, list...)
} PyObject;

Two fields. 16 bytes on a 64-bit system. Every value you've ever created in Python — every integer, every string, every function, every None — starts with these 16 bytes.

Each concrete type extends this with its own data:

// List (Objects/listobject.c)
typedef struct {
    PyObject_VAR_HEAD       // ob_refcnt + ob_type + ob_size
    PyObject **ob_item;     // array of POINTERS to elements
    Py_ssize_t allocated;   // capacity (not length)
} PyListObject;

Note: ob_item is an array of pointers, not objects. A Python list doesn't contain its elements — it contains pointers to them. This is how [1, "hello", None, []] works without type-specific list types. The list just holds PyObject* pointers and doesn't care what they point to.

Why `int` Costs 28 Bytes

import sys
# Python 3.12+
sys.getsizeof(0)      # → 24  (was 28 in Python ≤ 3.11)
sys.getsizeof(1)      # → 28
sys.getsizeof(2**30)  # → 32
sys.getsizeof(2**60)  # → 36

Python integers are bignums — no overflow, ever. Internally they store digits in base 2^30 as an array of uint32_t. That array grows as numbers grow. In C, an int is 4 bytes. In Python, the smallest integer costs 28.

In my scraper, I was creating hundreds of thousands of small integer objects per run as counters and indices. Each one: 28 bytes minimum, an allocation call, a reference count increment. At scale, that adds up.

Small Integer Interning

CPython pre-creates integers from -5 to 256 at startup and reuses them forever:

a = 100
b = 100
a is b  # True — same object in memory, zero allocation

a = 1000
b = 1000
a is b  # False — two separate allocations

Loop indices, boolean flags, small return codes — all of these hit the interned pool. Without this optimization, tight loops would hammer the allocator on numbers like 0, 1, True, False.

⚠️ Never use is for value comparison. a is b checks memory address. Past 256, "equal" integers are different objects. a == b is always correct.

String Interning

CPython silently interns strings that look like identifiers. Strings with spaces or symbols: not interned.

a = "status"
b = "status"
a is b  # True — looks like an identifier, interned

a = "status: ok"
b = "status: ok"
a is b  # False — space breaks the rule

import sys
a = sys.intern("status: ok")
b = sys.intern("status: ok")
a is b  # True — explicit

I use sys.intern() in systems that do millions of repeated dictionary lookups with the same string keys — like an ORM layer processing repeated field names or a JSON cache with known keys. String comparison becomes O(1) pointer check instead of O(n) character scan.

4. Memory Management: Three Tiers

When I write x = [], Python doesn't just call malloc. It runs through a three-tier allocation system:

┌──────────────────────────────────────────────────────┐
│  Tier 3: pymalloc  (Objects/obmalloc.c)               │
│  Objects ≤ 512 bytes — the vast majority              │
│  Custom arena/pool system, up to 10× faster           │
├──────────────────────────────────────────────────────┤
│  Tier 2: PyMem API  (PyMem_Malloc / PyMem_Free)       │
│  Thin wrapper, used for debug hooks                   │
├──────────────────────────────────────────────────────┤
│  Tier 1: OS Allocator  (malloc / free from libc)      │
│  Objects > 512 bytes and internal structures          │
└──────────────────────────────────────────────────────┘

Nearly everything lands in Tier 3. Here's how it works.

pymalloc: Arenas, Pools, Blocks

Arena (256 KB) ← one mmap() call to the OS
├── Pool [size_class=8]    ← all blocks 8 bytes
│   ├── [block][block][block]...
├── Pool [size_class=16]
...
└── Pool [size_class=512]

Term	Size	What It Is
Block	8–512 bytes (×8)	Holds one object
Pool	4 KB (OS page)	Blocks of the same size
Arena	256 KB	Up to 64 pools

Some old posts say Arena = 1 MB. Wrong. Check it: grep ARENA_SIZE cpython/Objects/obmalloc.c → #define ARENA_SIZE (256 << 10)

When you need a 24-byte block, pymalloc checks if there's a pool with that size class and a free slot. If yes — take it, done, zero system calls. That's why creating thousands of small objects in a loop is fast. The OS isn't involved on most allocations.

Why Python "Holds Onto" Memory

This one burned me on a bot that ran 24/7. After processing large batches, the process would stay at peak memory indefinitely — even after I deleted everything.

The rule:

Freed block → back to its pool
Empty pool → back to its arena
Empty arena only → released to OS

If you allocate 400 MB of small objects, delete them all, but one object in one pool in one arena is still alive — that 256 KB arena doesn't go back to the OS. From top's perspective: process still uses 400 MB.

This is not a bug. It's the arena pool holding memory for reuse. Knowing this means you stop chasing phantom memory leaks that aren't actually leaks.

5. Reference Counting — The Deterministic Heart

Every PyObject has ob_refcnt. When it drops to zero, the object is destroyed. Immediately. Not "next GC cycle" — immediately.

// Include/object.h (simplified)
#define Py_INCREF(op)  ((PyObject*)(op))->ob_refcnt++

#define Py_DECREF(op)                                \
    if (--((PyObject*)(op))->ob_refcnt == 0)         \
        _Py_Dealloc((PyObject*)(op))

These two macros are everywhere in CPython. Assign a variable → INCREF. Pass to a function → INCREF. Exit a scope → DECREF. It's constant background churn — the price of deterministic memory management.

Watching It Live

import sys

a = []
print(sys.getrefcount(a))  # 2 — variable 'a' + getrefcount's arg

b = a
print(sys.getrefcount(a))  # 3 — + 'b'

c = [a, a, a]
print(sys.getrefcount(a))  # 6 — + three list slots

del b
print(sys.getrefcount(a))  # 5

del c
print(sys.getrefcount(a))  # 2 — all three references gone at once

del a
# → 0 → _Py_Dealloc() → block back to pymalloc pool

getrefcount always reads one too high — the function call itself holds a temporary reference. Tripped me up for longer than I'd like to admit.

The Cascade Effect

When ob_refcnt hits zero, the destructor calls DECREF on everything the object holds. If a list has 10,000 items and you delete the list, all 10,000 items get DECREF-ed in one chain reaction. If any of those items also hit zero, they cascade too.

For my bots and scrapers, this actually matters. A del response_cache at the end of a batch run triggers an immediate cascade deallocation — predictable, synchronous, no surprises. This is genuinely useful when you're managing memory across long-running processes.

Context Managers Are Deterministic in CPython

In Java or Go, a file might stay open until the next GC cycle. In CPython:

with open("data.txt") as f:
    data = f.read()

# Here: f's refcount drops to 0 → f.__exit__() → file closed → memory freed
# Right here. Not "eventually".

This is exactly why with open(...) as f guarantees the file closes the moment you exit the block — not "eventually."

The `del` Trap

registry = []

class Zombie:
    def __del__(self):
        registry.append(self)  # refcount > 0 again — resurrection!

z = Zombie()
del z         # __del__ fires — but object survives into registry
registry.clear()  # dies for real — __del__ does NOT fire again

CPython calls __del__ exactly once. If the object survives, fine. When it finally dies, the destructor won't run again.

Don't use __del__ for cleanup. It's not reliable for resources. Use context managers.

The Fatal Flaw: Cycles

a = {}
a["self"] = a   # a → a → a...
del a           # refcount drops to 1, not 0. Nobody can reach it. Won't die.

Reference counting is completely blind to cycles. This is why the GC exists.

6. Garbage Collector — When Counters Aren't Enough

Python's GC (Modules/gcmodule.c) has one job: find cycles that reference counting missed and destroy them.

It doesn't track everything. Only container objects that can hold references — list, dict, set, custom class instances. A str is never GC-tracked because strings can't point to other objects.

Three Generations

Generation 0  (threshold=700)   ← new objects
Generation 1  (threshold=10)    ← survived 1 collection
Generation 2  (threshold=10)    ← long-lived

Gen 0 runs frequently. Gen 2 rarely. The logic: most objects die young. The ones that survive long enough are probably going to live forever.

Python 3.14: GC became incremental — work is split into small chunks interleaved with execution. No more single stop-the-world pause. Big deal for latency-sensitive services.

How It Finds Cycles

1. Copy ob_refcnt → gc_refs (scratch field)
2. For each tracked object: walk its references,
   subtract 1 from gc_refs of each pointed-to object
3. After the walk:
   gc_refs > 0  → reachable from outside → ALIVE
   gc_refs == 0 → only referenced within cycle → DEAD → destroy

Elegant. No global reachability scan — just local reference subtraction.

In Production

import gc

# ⚠️ Only disable if you're certain there are no cycles.
# In my bots: loggers, closures, Playwright browser contexts — all have cycles.
# Disabling GC with cycles = silent memory growth until OOM.
gc.disable()

gc.collect(0)  # Gen 0 only
gc.collect()   # Everything

print(gc.get_count())      # (480, 3, 1)
print(gc.get_threshold())  # (700, 10, 10)
gc.set_debug(gc.DEBUG_LEAK)

In my high-throughput scrapers, I've used scheduled GC to avoid surprise pauses mid-request:

import gc, threading, time, logging
logger = logging.getLogger(__name__)

def gc_worker():
    while True:
        time.sleep(30)
        n = gc.collect()
        logger.debug(f"GC collected {n} objects")

threading.Thread(target=gc_worker, daemon=True).start()

A predictable pause you control beats a random pause you don't. But: measure first. Most codebases don't need this.

`weakref` — Better Than Fighting Cycles

The right move is often to not create cycles in the first place. In any tree structure where nodes need a back-reference to their parent, use weakref:

import weakref

class NodeSafe:
    def __init__(self, value):
        self.value = value
        self._parent = None

    @property
    def parent(self):
        return self._parent() if self._parent else None

    @parent.setter
    def parent(self, node):
        self._parent = weakref.ref(node) if node else None

root = NodeSafe("root")
child = NodeSafe("child")
child.parent = root   # doesn't increment root's refcount

del root
print(child.parent)  # → None immediately. No GC needed.

I use this pattern in any linked/tree data structure. Also useful for caches where you want entries to disappear automatically when nothing else holds the value.

7. The GIL — The Most Misunderstood Thing in Python

I spent years being confused by the GIL. Let me save you the confusion.

The GIL (Global Interpreter Lock) is a mutex. It enforces one invariant: only one thread executes Python bytecode at a given moment.

// Python/ceval_gil.c
static _Py_atomic_int eval_breaker;   // "yield now" flag
static _Py_atomic_int gil_locked;     // 0 = free, 1 = held

// sys.getswitchinterval() = 5ms by default
// This is a CHECK INTERVAL — not a guaranteed switch time.
// Real switching only happens when a waiting thread requests
// the GIL within that window.

Why It Exists

Reference counting is not thread-safe. Two threads touching ob_refcnt concurrently is a data race. Without synchronization, you'd get _Py_Dealloc firing twice on the same object — use-after-free, crash. The GIL is the "one big lock" solution to this. For 1991 when CPython was first written, it was a reasonable engineering choice.

The trade-off: simplicity and safety in exchange for CPU parallelism.

What It Doesn't Block

This is the part most people miss.

The GIL is released during I/O. Every network call, every disk read, every time.sleep — the GIL drops for the duration. While one thread waits on the network, other threads run bytecode. Real concurrency.

In my Telegram bots, I run hundreds of concurrent polling connections. Threading works great for this because 99% of the time each thread is blocked waiting on an API response — GIL is free for whoever needs it.

import threading, urllib.request

def fetch(url):
    urllib.request.urlopen(url)  # GIL released during entire network wait

threads = [threading.Thread(target=fetch, args=(url,)) for url in urls]
for t in threads: t.start()
for t in threads: t.join()
# Actual concurrent I/O across all threads

C extensions can also explicitly drop the GIL during computation. That's how numpy, pandas, and Pillow achieve real multi-core performance — Python layer holds the GIL, C computation layer drops it.

❗ The Rule to Remember

The GIL doesn't make Python slow — it just means CPython can't use multiple CPU cores in parallel.

CPU-bound task  →  multiprocessing  (each process = its own GIL)
I/O-bound task  →  threading/asyncio (GIL releases on every blocking call)
C extension     →  threading + Py_BEGIN_ALLOW_THREADS

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ProcessPoolExecutor(max_workers=4) as exe:
    results = list(exe.map(heavy, [1_000_000] * 8))
# 4 real cores, no GIL contention

In my automation work: API calls, web scraping, Telegram polling — threading works perfectly. Parsing HTML, running ML inference, crunching numbers — multiprocessing or a C extension.

8. The Eval Loop — The Heart of the Machine

Python/ceval.c. This is where every Python instruction you've ever executed actually ran.

~3,000 lines of C. One for(;;) loop. One switch statement inside with a case for each opcode. That's it.

// Simplified — the real version is much more complex
PyObject *
_PyEval_EvalFrameDefault(PyThreadState *tstate, PyFrameObject *f, int exc)
{
    for (;;) {
        if (_Py_atomic_load_relaxed(&eval_breaker)) {
            // Handle signals, check whether to yield the GIL, etc.
        }

        opcode = NEXTOPARG();

        switch (opcode) {
            case LOAD_FAST: {
                PyObject *value = GETLOCAL(oparg);
                Py_INCREF(value);
                PUSH(value);
                break;
            }
            case BINARY_OP: {
                PyObject *right = POP();
                PyObject *left  = POP();
                // Reality: dispatches through _PyNumber_BinaryOp()
                // → left->ob_type->tp_as_number->nb_add
                // → PyNumber_Add → slot lookup
                // Simplified here for readability.
                PyObject *res = binary_op(left, right, oparg);
                Py_DECREF(left); Py_DECREF(right);
                PUSH(res);
                break;
            }
            case RETURN_VALUE: {
                retval = POP();
                goto return_or_yield;
            }
        }
    }
}

Python is a stack machine. Instructions push values onto a virtual stack and pop them to perform operations:

[LOAD_FAST a]  → Stack: [42]
[LOAD_FAST b]  → Stack: [42, 8]
[BINARY_OP +]  → Stack: [50]
[RETURN_VALUE] → returns 50, frame destroyed

Frame Objects

Each function call creates a PyFrameObject — locals, current bytecode position, pointer to the calling frame. The f_back chain is the call stack you see in tracebacks:

import sys

def inner():
    frame = sys._getframe()
    print(frame.f_code.co_name)         # 'inner'
    print(frame.f_back.f_code.co_name)  # 'outer'

def outer():
    inner()
outer()

Since Python 3.11, frames are often C-stack allocated rather than heap allocated. Deep recursion got noticeably faster from this alone.

The Adaptive Interpreter (Python 3.11+)

This is genuinely clever. The Eval Loop now watches what types your code uses. After seeing the same instruction with the same types enough times, it rewrites that instruction in place with a specialized version:

BINARY_OP   →  BINARY_OP_ADD_INT      (both operands are int — skip type checks)
LOAD_ATTR   →  LOAD_ATTR_MODULE       (attribute is from a module — direct pointer)
CALL        →  CALL_PY_EXACT_ARGS     (argument count matches exactly)

Python rewrites its own bytecode at runtime based on what it observes your code doing. A specialized opcode skips the entire type dispatch chain and goes straight to the optimized C path.

Result: +25% throughput versus Python 3.10 with no changes to your code. This is also the foundation the experimental JIT in Python 3.13+ builds on.

9. C Extensions — What Actually Made My Scraper 13x Faster

Back to the original question.

Any .so on Linux/macOS or .pyd on Windows is a dynamic library built against the Python C API. This is the mechanism behind numpy, lxml, ujson, orjson, Pillow — Python shell, C engine.

Here's what my scraper change looked like conceptually:

# Before: pure Python, 1.2s per document
def extract_links(html: str) -> list[str]:
    from bs4 import BeautifulSoup
    soup = BeautifulSoup(html, "html.parser")
    return [a["href"] for a in soup.find_all("a", href=True)]

What Python was doing under the hood for each document:

Create dozens of PyObject instances for DOM nodes
Run refcount increments/decrements on every attribute access
All HTML traversal going through the Eval Loop opcode by opcode
GIL held for the entire duration — no other thread could run

The C extension path:

static PyObject *
fast_extract_links(PyObject *self, PyObject *args)
{
    const char *html;
    PyArg_ParseTuple(args, "s", &html);

    Py_BEGIN_ALLOW_THREADS   // ← GIL released
    // ... libxml2 parsing in pure C, no PyObjects, no refcounts ...
    Py_END_ALLOW_THREADS     // ← GIL re-acquired

    return result_list;
}

What changed: the HTML traversal went from "thousands of opcode dispatches + PyObject allocations" to "C function calls with direct memory access." No Eval Loop overhead. No per-step refcount churn. GIL released so other threads ran in parallel. 13x.

Building a Minimal C Extension

// mymodule.c
#define PY_SSIZE_T_CLEAN
#include <Python.h>

static PyObject *
double_it(PyObject *self, PyObject *args)
{
    int n;
    if (!PyArg_ParseTuple(args, "i", &n))
        return NULL;
    return PyLong_FromLong(n * 2);
}

static PyMethodDef methods[] = {
    {"double_it", double_it, METH_VARARGS, "Double an integer"},
    {NULL, NULL, 0, NULL}
};

static struct PyModuleDef module = {
    PyModuleDef_HEAD_INIT, "mymodule", NULL, -1, methods
};

PyMODINIT_FUNC PyInit_mymodule(void) {
    return PyModule_Create(&module);
}

python setup.py build_ext --inplace

import mymodule
mymodule.double_it(21)  # → 42

Reference Management — Where Things Go Wrong

The most common bug in C extensions: getting refcounts wrong. Leak one INCREF → memory leak. Miss one DECREF → use-after-free. The standard cleanup pattern:

static PyObject *
safe_build_list(PyObject *self, PyObject *args)
{
    PyObject *list = NULL, *item = NULL;

    list = PyList_New(3);
    if (!list) goto error;

    for (int i = 0; i < 3; i++) {
        item = PyLong_FromLong(i);
        if (!item) goto error;

        PyList_SET_ITEM(list, i, item);  // list steals the reference
        item = NULL;                      // we don't own it anymore
    }

    return list;

error:
    Py_XDECREF(item);  // XDECREF is NULL-safe
    Py_XDECREF(list);
    return NULL;
}

The Modern Toolkit

Nobody writes raw C extensions unless they have a specific reason. The usual choices:

Tool	When
ctypes	Call existing `.dll`/`.so` without compiling anything
cffi	More robust ctypes for complex C APIs
Cython	Python-like syntax → compiles to C; great for loops
pybind11	Wrapping existing C++ code — industry standard
mypyc	Type-annotated Python → C extension, zero new syntax

10. PyPy, Python 3.13 Free-Threading, and What's Coming

PyPy

PyPy is an alternative Python interpreter with a tracing JIT compiler. It watches what your code does and compiles hot paths to machine code. Loops that take seconds in CPython: milliseconds in PyPy, no source changes needed.

The trade-off: PyPy uses tracing GC instead of reference counting. No deterministic __del__. Objects die when the GC decides, not when the last reference disappears.

pypy3 script.py  # often faster with zero changes

Use it for: pure Python algorithms, number crunching without numpy, long-running services with well-defined hot paths.

Avoid it when: you depend heavily on C extensions like numpy, pandas, Playwright. They're built for the CPython ABI and range from slower to broken under PyPy.

CPython 3.13: Free-Threading (PEP 703)

Python 3.13 shipped an experimental build with the GIL removed:

pyenv install 3.13t  # 't' = free-threaded
python3.13t -c "import sys; print(sys._is_gil_enabled())"  # False

Removing the GIL required solving the thread-safety problem differently:

Biased Reference Counting — fast per-thread counters for thread-owned objects; slower shared counters for cross-thread objects
Immortal Objects (PEP 683) — None, True, False get a fixed sentinel refcount and are never modified, eliminating all races on the most common objects
Per-object locks on dict, list, and other mutable containers

Currently opt-in — most C extensions aren't compatible yet. But this is the most significant architectural change to CPython in its history.

Architecture Map

Your code (.py)
      │
      ▼
 [PEG Parser]  →  AST  →  [Compiler]  →  Bytecode
                                               │
                                               ▼
                                        [Eval Loop]
                                         ceval.c
                                               │
                              ┌────────────────┼────────────────┐
                              │                │                │
                              ▼                ▼                ▼
                        [PyObject]      [Memory Mgmt]    [C Extensions]
                        ob_refcnt       pymalloc           .so / .pyd
                        ob_type         arenas/pools       Python C API
                              │                │
                              ▼                ▼
                        [Ref Counting]   [GC (cyclic)]
                        Py_INCREF/       Mark & Sweep
                        Py_DECREF        3 generations

11. What Actually Changed in My Code

This is the part nobody writes. Not "here are optimizations" — but what specifically changed in how I write real production code after understanding CPython internals.

1. I use __slots__ on any class I instantiate at volume.
My Playwright bot creates thousands of event objects and DOM node wrappers per session. Before: ~280 bytes per object. After __slots__: ~56 bytes. On a 24-hour run, that difference is significant.

2. I bind module functions before tight loops.
Bots and scrapers have hot loops. json.loads in a loop → _loads = json.loads above the loop. This is LOAD_FAST vs LOAD_GLOBAL on every iteration. In a loop over 100k items, it's measurable.

3. I stopped using threading for CPU-bound work.
Sounds obvious in hindsight. But I had a pipeline where I was running threading.Thread for parallel HTML parsing, wondering why I wasn't getting faster. Now I know: parsing is CPU-bound, GIL doesn't release, threads compete instead of parallelize. Switched to ProcessPoolExecutor, got the expected speedup.

4. I reach for C extensions or Cython for genuine hot paths.
Not prematurely — only after profiling. But knowing that the cost difference is "thousands of Eval Loop opcode dispatches" vs "direct C function calls" gives me a clear mental model for when the overhead matters enough to pay the engineering cost.

5. I use weakref in data structures with back-references.
My bots have conversation state graphs where nodes reference parent nodes. Without weakref, the GC has to find and clean up the cycles. With weakref, parent references are free — no cycle, no GC involvement, immediate cleanup.

6. When a memory leak appears, I know where to look first.
Before: guessing, adding del everywhere, restarting services. Now: is it a cycle? Is it a module-level cache growing unbounded? Is pymalloc holding freed arenas? Is a C extension leaking a refcount? I have a mental model that makes the diagnostic systematic instead of random.

Conclusion

Python isn't slow because it's poorly designed. It's slow because it does an enormous amount of work on your behalf.

Every x = [] is an allocator call, a refcount, a type pointer, a GC registration. Every a + b is a type dispatch chain. Every function call is a new frame object. Python hides this so you can think about the problem. The hiding is the point.

Understanding it doesn't mean you write lower-level code. It means you know when the abstraction is costing you something and when it isn't. You stop cargo-culting optimizations and start making decisions based on what's actually happening in the runtime.

That's the difference.

Next steps if you want to go further:

Clone CPython: git clone https://github.com/python/cpython
Read Objects/listobject.c — list implementation, including list.sort() (timsort)
Read Objects/longobject.c — why integers work the way they do
Run dis.dis() on your hot functions and look at what Python actually executes
python -m cProfile -s cumulative your_script.py before optimizing anything

If you're preparing for senior/lead interviews, Part 2 will cover how to talk about CPython internals in a technical interview — what's worth knowing, what questions to expect, how to frame deep knowledge without sounding like you're reciting a textbook.

I write about Python automation, bots, and infrastructure — came from 6 years in business, now 2 years deep in the engineering side.

python #cpython #performance #programming #automation #webdev