DEV Community: Shayan Holakouee

A Systematic Approach to AI Agent Engineering

Shayan Holakouee — Thu, 04 Jun 2026 13:48:41 +0000

You've shipped microservices that handle millions of requests. You've designed event-driven systems that wake you up at 3am. Now your CTO is asking you to "add AI agents." This is that guide written for engineers who know what a race condition is and have opinions about retry logic.

The Mental Model Shift

Most AI agent frameworks are demos dressed up as infrastructure. Before touching any library, internalize this:

An AI agent is a control loop with a probabilistic step.

That's it. The "intelligence" is just one step in a workflow usually a call to an LLM that produces a decision. Everything around it: state management, tool dispatch, error handling, observability that's engineering. And that's where senior engineers have an actual edge.

while not done:
    observation = perceive(environment)
    action = llm_decide(observation, memory, tools)  # the probabilistic step
    result = execute(action)
    memory.update(result)

Once you see it this way, your existing intuitions about fault tolerance, idempotency, and distributed systems apply directly.

Architecture Patterns That Actually Scale

1. The Supervisor Pattern (Multi-Agent Orchestration)

For complex workflows, a single agent becomes a god object. Instead, build a supervisor that routes to specialized sub-agents.

from dataclasses import dataclass
from typing import Callable, Any
import asyncio

@dataclass
class AgentTask:
    task_id: str
    task_type: str
    payload: dict
    priority: int = 5

class SupervisorAgent:
    def __init__(self):
        self._registry: dict[str, Callable] = {}
        self._queue: asyncio.PriorityQueue = asyncio.PriorityQueue()

    def register(self, task_type: str):
        """Decorator to register specialized sub-agents."""
        def decorator(fn: Callable):
            self._registry[task_type] = fn
            return fn
        return decorator

    async def dispatch(self, task: AgentTask) -> Any:
        handler = self._registry.get(task.task_type)
        if not handler:
            raise ValueError(f"No agent registered for task type: {task.task_type}")
        return await handler(task.payload)

    async def run_parallel(self, tasks: list[AgentTask]) -> list[Any]:
        """Dispatch multiple tasks concurrently with priority ordering."""
        sorted_tasks = sorted(tasks, key=lambda t: t.priority, reverse=True)
        return await asyncio.gather(*[self.dispatch(t) for t in sorted_tasks])


supervisor = SupervisorAgent()

@supervisor.register("code_review")
async def code_review_agent(payload: dict) -> dict:
    # Specialized agent with its own system prompt and tools
    ...

@supervisor.register("test_generation")
async def test_generation_agent(payload: dict) -> dict:
    ...

The key insight: each sub-agent is stateless from the supervisor's perspective. State lives in your message store, not in the agent object.

2. Pipelines vs. Agents: Know Which One You Need

A pipeline is deterministic every input follows the same steps. An agent is non-deterministic the LLM decides what to do next. Don't use an agent when a pipeline will do.

Use a Pipeline when...	Use an Agent when...
Steps are known at design time	Steps depend on runtime context
Latency matters (low variance needed)	Flexibility matters
You need 100% auditability	Best-effort is acceptable
Input/output schema is fixed	Open-ended task completion

# Pipeline: deterministic, predictable, fast
class DocumentPipeline:
    steps = [extract_text, chunk_document, embed_chunks, store_vectors]

    async def run(self, document: bytes) -> list[str]:
        result = document
        for step in self.steps:
            result = await step(result)
        return result

# Agent: flexible, non-deterministic, slower
class ResearchAgent:
    async def run(self, query: str) -> str:
        # May call web_search 0-5 times, then summarize
        # The LLM decides
        ...

Advanced Python Patterns for AI Systems

Structured Outputs with Pydantic + Runtime Validation

Never parse LLM output with regex. Define your schema, enforce it, validate at runtime.

from pydantic import BaseModel, field_validator, model_validator
from typing import Literal
import instructor
import anthropic

class ActionPlan(BaseModel):
    reasoning: str
    action: Literal["search", "summarize", "delegate", "terminate"]
    confidence: float
    tool_args: dict | None = None

    @field_validator("confidence")
    @classmethod
    def confidence_must_be_probability(cls, v: float) -> float:
        if not 0.0 <= v <= 1.0:
            raise ValueError("Confidence must be between 0 and 1")
        return round(v, 3)

    @model_validator(mode="after")
    def tool_args_required_for_search(self) -> "ActionPlan":
        if self.action == "search" and not self.tool_args:
            raise ValueError("tool_args required when action is 'search'")
        return self

# Using instructor for structured extraction
client = instructor.from_anthropic(anthropic.Anthropic())

def get_action(observation: str, available_tools: list[str]) -> ActionPlan:
    return client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        response_model=ActionPlan,
        messages=[{
            "role": "user",
            "content": f"Observation: {observation}\nAvailable tools: {available_tools}\nWhat action should we take?"
        }]
    )

Memory Architecture: Don't Reinvent the Wheel

Production agents need layered memory. Model it explicitly:

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import hashlib

class MemoryLayer(ABC):
    @abstractmethod
    async def store(self, key: str, value: dict) -> None: ...

    @abstractmethod
    async def retrieve(self, query: str, top_k: int = 5) -> list[dict]: ...


class WorkingMemory:
    """Short-term, in-process. Cleared between runs."""
    def __init__(self, max_tokens: int = 8000):
        self._buffer: list[dict] = []
        self._max_tokens = max_tokens

    def add(self, message: dict) -> None:
        self._buffer.append(message)
        self._trim()

    def _trim(self) -> None:
        # Sliding window keep most recent context under token budget
        # In practice, count tokens properly with tiktoken or equivalent
        while len(self._buffer) > 20:
            self._buffer.pop(0)

    def snapshot(self) -> list[dict]:
        return list(self._buffer)


class EpisodicMemory(MemoryLayer):
    """Medium-term. Key episodes from past runs. Stored in vector DB."""

    async def store(self, key: str, value: dict) -> None:
        embedding = await embed(value["content"])
        await vector_db.upsert(id=key, vector=embedding, metadata=value)

    async def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        query_embedding = await embed(query)
        results = await vector_db.query(vector=query_embedding, top_k=top_k)
        return [r.metadata for r in results]


class AgentMemoryManager:
    """Orchestrates all memory layers."""
    def __init__(self):
        self.working = WorkingMemory()
        self.episodic = EpisodicMemory()

    async def get_context(self, query: str) -> list[dict]:
        working_context = self.working.snapshot()
        episodic_context = await self.episodic.retrieve(query, top_k=3)
        # Merge, deduplicate, and rank
        return self._merge(working_context, episodic_context)

    def _merge(self, working: list, episodic: list) -> list:
        seen = set()
        merged = []
        for item in working + episodic:
            content_hash = hashlib.md5(str(item).encode()).hexdigest()
            if content_hash not in seen:
                seen.add(content_hash)
                merged.append(item)
        return merged

Tool Use: The Sharp Edge

Tool execution is where most agent systems fail in production. Every tool is an external side effect treat it like a database write.

Idempotent Tool Design

import functools
import hashlib
import json
from typing import TypeVar, Callable

T = TypeVar("T")

def idempotent_tool(cache_ttl_seconds: int = 300):
    """
    Makes tool calls idempotent by caching results keyed on input.
    Critical for retry scenarios where the LLM re-calls a tool that already succeeded.
    """
    def decorator(fn: Callable[..., T]) -> Callable[..., T]:
        _cache: dict[str, tuple[T, datetime]] = {}

        @functools.wraps(fn)
        async def wrapper(*args, **kwargs) -> T:
            cache_key = hashlib.sha256(
                json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True).encode()
            ).hexdigest()

            if cache_key in _cache:
                result, cached_at = _cache[cache_key]
                if datetime.utcnow() - cached_at < timedelta(seconds=cache_ttl_seconds):
                    return result

            result = await fn(*args, **kwargs)
            _cache[cache_key] = (result, datetime.utcnow())
            return result

        return wrapper
    return decorator


@idempotent_tool(cache_ttl_seconds=60)
async def web_search(query: str, max_results: int = 5) -> list[dict]:
    """Tool that won't hammer the search API on LLM retries."""
    ...

Tool Registry with Schema Validation

from typing import get_type_hints
import inspect

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, dict] = {}

    def register(self, description: str):
        def decorator(fn: Callable):
            hints = get_type_hints(fn)
            sig = inspect.signature(fn)

            schema = {
                "name": fn.__name__,
                "description": description,
                "input_schema": {
                    "type": "object",
                    "properties": {
                        param: {"type": self._python_type_to_json(hints.get(param))}
                        for param in sig.parameters
                        if param != "return"
                    },
                    "required": [
                        p for p, v in sig.parameters.items()
                        if v.default is inspect.Parameter.empty
                    ]
                }
            }
            self._tools[fn.__name__] = {"fn": fn, "schema": schema}
            return fn
        return decorator

    def get_schemas(self) -> list[dict]:
        return [t["schema"] for t in self._tools.values()]

    async def call(self, name: str, **kwargs) -> Any:
        if name not in self._tools:
            raise KeyError(f"Tool '{name}' not found in registry")
        return await self._tools[name]["fn"](**kwargs)

    @staticmethod
    def _python_type_to_json(t) -> str:
        mapping = {str: "string", int: "integer", float: "number", bool: "boolean"}
        return mapping.get(t, "string")

Observability Is Not Optional

An agent you can't observe is a black box your on-call rotation will hate. Instrument everything.

import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field

@dataclass
class AgentSpan:
    span_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    trace_id: str = ""
    parent_id: str | None = None
    name: str = ""
    start_time: float = field(default_factory=time.time)
    end_time: float | None = None
    attributes: dict = field(default_factory=dict)
    events: list[dict] = field(default_factory=list)
    status: str = "ok"

    def add_event(self, name: str, attrs: dict = None):
        self.events.append({
            "name": name,
            "timestamp": time.time(),
            "attributes": attrs or {}
        })

    def finish(self, status: str = "ok"):
        self.end_time = time.time()
        self.status = status
        return self

class AgentTracer:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self._spans: list[AgentSpan] = []
        self._current_span: AgentSpan | None = None

    @asynccontextmanager
    async def span(self, name: str, **attrs):
        s = AgentSpan(
            trace_id=self.trace_id,
            parent_id=self._current_span.span_id if self._current_span else None,
            name=name,
            attributes=attrs
        )
        prev = self._current_span
        self._current_span = s
        try:
            yield s
            s.finish("ok")
        except Exception as e:
            s.add_event("exception", {"message": str(e), "type": type(e).__name__})
            s.finish("error")
            raise
        finally:
            self._spans.append(s)
            self._current_span = prev

    def export(self) -> list[dict]:
        return [
            {
                "trace_id": s.trace_id,
                "span_id": s.span_id,
                "parent_id": s.parent_id,
                "name": s.name,
                "duration_ms": round((s.end_time - s.start_time) * 1000, 2) if s.end_time else None,
                "status": s.status,
                "attributes": s.attributes,
                "events": s.events,
            }
            for s in self._spans
        ]


# Usage inside your agent loop
async def agent_step(observation: str, tracer: AgentTracer) -> dict:
    async with tracer.span("llm_call", model="claude-opus-4-5", input_tokens=len(observation)):
        action = await call_llm(observation)

    async with tracer.span("tool_execution", tool=action.tool_name):
        result = await tool_registry.call(action.tool_name, **action.args)

    return result

Feed spans into your existing observability stack (Datadog, Grafana, Honeycomb). AI agents are distributed systems trace them like one.

Failure Modes That Will Bite You

1. The Infinite Loop

LLMs can get stuck in thought loops calling the same tool repeatedly, or alternating between two actions.

from collections import Counter

class LoopDetector:
    def __init__(self, window: int = 10, threshold: int = 3):
        self._window = window
        self._threshold = threshold
        self._action_history: list[str] = []

    def check(self, action: str) -> bool:
        """Returns True if a loop is detected."""
        self._action_history.append(action)
        if len(self._action_history) > self._window:
            self._action_history.pop(0)

        counts = Counter(self._action_history)
        return any(count >= self._threshold for count in counts.values())

2. Context Window Overflow

For long-running agents, the context window fills up. Build compaction in from the start.

async def compact_context(messages: list[dict], llm_client) -> list[dict]:
    """Summarize older messages to free context space."""
    if len(messages) < 20:
        return messages

    old_messages = messages[:-10]
    recent_messages = messages[-10:]

    summary_response = await llm_client.messages.create(
        model="claude-haiku-4-5",  # Use cheaper model for summarization
        max_tokens=512,
        messages=[
            *old_messages,
            {"role": "user", "content": "Summarize the key facts and decisions from this conversation in under 200 words."}
        ]
    )

    summary_message = {
        "role": "assistant",
        "content": f"[Context Summary]: {summary_response.content[0].text}"
    }

    return [summary_message] + recent_messages

3. Prompt Injection in Tool Outputs

External data (web pages, user files, database rows) can contain adversarial instructions. Sanitize before feeding back to the LLM.

import re

INJECTION_PATTERNS = [
    r"ignore\s+previous\s+instructions",
    r"new\s+system\s+prompt",
    r"you\s+are\s+now\s+",
    r"disregard\s+all",
]

def sanitize_tool_output(raw_output: str) -> str:
    """Basic prompt injection mitigation for tool outputs."""
    sanitized = raw_output
    for pattern in INJECTION_PATTERNS:
        sanitized = re.sub(pattern, "[REDACTED]", sanitized, flags=re.IGNORECASE)

    # Wrap in a context boundary
    return f"<tool_result>\n{sanitized}\n</tool_result>"

Putting It Together: A Production Agent Loop

async def run_agent(
    task: str,
    max_iterations: int = 20,
    tracer: AgentTracer | None = None,
) -> str:
    tracer = tracer or AgentTracer()
    memory = AgentMemoryManager()
    loop_detector = LoopDetector()

    memory.working.add({"role": "user", "content": task})

    for iteration in range(max_iterations):
        async with tracer.span("agent_iteration", iteration=iteration):

            # Build context
            context = await memory.get_context(task)
            if len(context) > 15:
                context = await compact_context(context, llm_client)

            # Get action from LLM
            async with tracer.span("llm_decision"):
                action = get_action(
                    observation=json.dumps(context[-5:]),
                    available_tools=tool_registry.get_schemas()
                )

            # Loop detection
            if loop_detector.check(action.action):
                tracer._current_span.add_event("loop_detected")
                return "Agent halted: repetitive action loop detected."

            # Terminal condition
            if action.action == "terminate":
                async with tracer.span("final_answer"):
                    return await generate_final_answer(context)

            # Execute tool
            async with tracer.span("tool_call", tool=action.action, confidence=action.confidence):
                tool_result = await tool_registry.call(action.action, **(action.tool_args or {}))
                sanitized_result = sanitize_tool_output(str(tool_result))

            # Update memory
            memory.working.add({"role": "assistant", "content": str(action)})
            memory.working.add({"role": "tool", "content": sanitized_result})

    return "Agent halted: max iterations reached."

Framework Landscape (2025 State of Play)

Don't build everything from scratch. Know when to reach for existing tools:

Framework	Best For	Watch Out For
LangGraph	Stateful, graph-based workflows	Steep learning curve; abstraction leaks
CrewAI	Multi-agent role-playing	Can be too "magic" for complex logic
Pydantic AI	Type-safe, structured agent outputs	Younger ecosystem
Temporal + custom	Long-running, durable workflows	Infrastructure overhead
Raw API + your code	Full control, complex routing	You own the bugs

The right answer for most teams: use LangGraph for the graph/state machine, instructor for structured outputs, Pydantic for validation, and write tool execution yourself.

Closing Thoughts

AI agents are not magic. They're non-deterministic state machines wrapped around probabilistic function calls and they demand the same rigor as any other distributed system.

Your unfair advantage as a senior engineer isn't knowing the hottest framework. It's knowing that:

Idempotency matters. Tools get retried.
Observability is load-bearing. You will debug this at midnight.
Failure modes compound. Design for them explicitly, not as an afterthought.
Simpler is almost always better. A pipeline beats an agent when you know the steps.

The teams shipping reliable AI systems aren't the ones with the most sophisticated prompts. They're the ones who treated AI components like infrastructure with circuit breakers, retries, schema validation, and on-call runbooks.

Build accordingly.

Found a bug or a better pattern? Drop a comment. This is production code, not a tutorial corrections welcome.

If this was useful, follow for more on AI systems engineering, Python architecture, and the sharp edges that don't make it into the demos.

Building a net/http Middleware Chain From Scratch in Go

Shayan Holakouee — Wed, 03 Jun 2026 17:10:51 +0000

Every Go web framework ships with middleware. Chi has it. Gin has it. Echo has it. And every tutorial shows you how to use it.

But almost none of them show you how it actually works.

This article builds a production-grade middleware chain using nothing but the Go standard library. No framework, no magic. By the end you will have written request logging, authentication, panic recovery, and timeout middleware from scratch, and you will understand exactly why the pattern works the way it does.

The Foundation: What Is http.Handler?

Everything in Go's HTTP world revolves around one interface:

type Handler interface {
    ServeHTTP(ResponseWriter, *Request)
}

That's it. One method. Any type that implements ServeHTTP is an HTTP handler. This simplicity is intentional, and it's what makes middleware composable.

There's also http.HandlerFunc, a function type that implements Handler:

type HandlerFunc func(ResponseWriter, *Request)

func (f HandlerFunc) ServeHTTP(w ResponseWriter, r *Request) {
    f(w, r)
}

This adapter lets you write a plain function and treat it as a Handler. You'll use this constantly.

The Middleware Signature

A middleware is a function that takes a Handler and returns a Handler:

type Middleware func(http.Handler) http.Handler

That's the whole pattern. Middleware wraps a handler, intercepts the request, does something before or after calling the next handler, and passes control along.

Here's the skeleton every middleware follows:

func MyMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // do something before
        next.ServeHTTP(w, r)
        // do something after
    })
}

The next handler is the thing being wrapped. You call it explicitly. This is the key insight: Go middleware is not magic. It's just nested function calls, composed at setup time.

Building the Chain

Managing middleware manually gets messy fast:

// This is readable but doesn't scale
handler := LoggingMiddleware(AuthMiddleware(RecoveryMiddleware(myHandler)))

Let's write a proper Chain that takes a slice of middleware and applies them in order:

package middleware

import "net/http"

type Middleware func(http.Handler) http.Handler

func Chain(middlewares ...Middleware) Middleware {
    return func(final http.Handler) http.Handler {
        for i := len(middlewares) - 1; i >= 0; i-- {
            final = middlewares[i](final)
        }
        return final
    }
}

The reverse iteration matters: if you want request logging to run first (outermost), it needs to wrap everything else. Applying in reverse order ensures the first middleware in your list is the first one that sees the request.

Usage:

chain := middleware.Chain(
    RecoveryMiddleware,
    LoggingMiddleware,
    AuthMiddleware,
)

http.Handle("/api/", chain(myAPIHandler))

Clean, ordered, easy to reason about.

Middleware 1: Request Logging

The most common middleware. We want to log the method, path, status code, and duration of every request.

The tricky part: http.ResponseWriter doesn't expose the status code after it's been written. We need to wrap it:

package middleware

import (
    "log/slog"
    "net/http"
    "time"
)

type responseWriter struct {
    http.ResponseWriter
    status int
    written bool
}

func (rw *responseWriter) WriteHeader(code int) {
    if !rw.written {
        rw.status = code
        rw.written = true
        rw.ResponseWriter.WriteHeader(code)
    }
}

func (rw *responseWriter) statusCode() int {
    if !rw.written {
        return http.StatusOK // default if WriteHeader was never called
    }
    return rw.status
}

func LoggingMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()

        wrapped := &responseWriter{ResponseWriter: w}
        next.ServeHTTP(wrapped, r)

        slog.Info("request",
            "method", r.Method,
            "path", r.URL.Path,
            "status", wrapped.statusCode(),
            "duration", time.Since(start),
            "remote_addr", r.RemoteAddr,
        )
    })
}

A few things worth noting here:

We embed http.ResponseWriter in our wrapper struct so all other methods (Header(), Write(), etc.) delegate to the original automatically.
We guard against double calls to WriteHeader, which handlers can do accidentally.
We use log/slog (Go 1.21+) for structured logging. If you're on an older version, swap in your preferred logger.

Middleware 2: Panic Recovery

A panic in a goroutine serving a request will crash the entire server if not recovered. This middleware catches panics and turns them into 500 responses:

package middleware

import (
    "log/slog"
    "net/http"
    "runtime/debug"
)

func RecoveryMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        defer func() {
            if err := recover(); err != nil {
                slog.Error("panic recovered",
                    "error", err,
                    "stack", string(debug.Stack()),
                )
                http.Error(w, "internal server error", http.StatusInternalServerError)
            }
        }()
        next.ServeHTTP(w, r)
    })
}

debug.Stack() gives you the full goroutine stack trace at the point of the panic, which is invaluable for debugging production incidents.

Put RecoveryMiddleware first (outermost) in your chain. It needs to wrap everything so nothing escapes.

Middleware 3: Authentication

This one reads a Bearer token, validates it, and puts the claims into the request context so downstream handlers can read them without re-doing the work.

package middleware

import (
    "context"
    "net/http"
    "strings"
)

type contextKey string

const UserIDKey contextKey = "userID"

func AuthMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        header := r.Header.Get("Authorization")
        if !strings.HasPrefix(header, "Bearer ") {
            http.Error(w, "unauthorized", http.StatusUnauthorized)
            return
        }

        token := strings.TrimPrefix(header, "Bearer ")
        userID, err := validateToken(token) // your validation logic
        if err != nil {
            http.Error(w, "unauthorized", http.StatusUnauthorized)
            return
        }

        ctx := context.WithValue(r.Context(), UserIDKey, userID)
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

// Reading it in a handler
func myHandler(w http.ResponseWriter, r *http.Request) {
    userID, ok := r.Context().Value(UserIDKey).(string)
    if !ok {
        http.Error(w, "missing user", http.StatusInternalServerError)
        return
    }
    // use userID
}

A few design choices worth explaining:

Why a custom contextKey type? Context keys are compared by value and type. Using a plain string like "userID" means any package could accidentally read or overwrite your value. A private contextKey type scoped to your package makes collisions impossible.

Why return early on auth failure? The middleware should short-circuit: if auth fails, call http.Error and return without calling next.ServeHTTP. Forgetting the return after the error write is a classic bug that leads to handlers running even when auth failed.

Middleware 4: Request Timeout

Long-running requests can exhaust your server's connection pool. This middleware cancels the request context after a deadline:

package middleware

import (
    "context"
    "net/http"
    "time"
)

func TimeoutMiddleware(timeout time.Duration) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ctx, cancel := context.WithTimeout(r.Context(), timeout)
            defer cancel()

            done := make(chan struct{})
            go func() {
                next.ServeHTTP(w, r.WithContext(ctx))
                close(done)
            }()

            select {
            case <-done:
                // handler finished in time
            case <-ctx.Done():
                http.Error(w, "request timeout", http.StatusGatewayTimeout)
            }
        })
    }
}

Note that TimeoutMiddleware takes a time.Duration and returns a Middleware. This is the factory pattern: when a middleware needs configuration, wrap it in a function that accepts that config and returns the actual middleware. Clean and composable.

Usage:

chain := middleware.Chain(
    RecoveryMiddleware,
    LoggingMiddleware,
    middleware.TimeoutMiddleware(5*time.Second),
    AuthMiddleware,
)

Putting It All Together

Here's a full, working example:

package main

import (
    "fmt"
    "net/http"
    "time"

    "yourmodule/middleware"
)

func helloHandler(w http.ResponseWriter, r *http.Request) {
    userID, _ := r.Context().Value(middleware.UserIDKey).(string)
    fmt.Fprintf(w, "Hello, %s\n", userID)
}

func main() {
    chain := middleware.Chain(
        middleware.RecoveryMiddleware,
        middleware.LoggingMiddleware,
        middleware.TimeoutMiddleware(5*time.Second),
        middleware.AuthMiddleware,
    )

    mux := http.NewServeMux()
    mux.Handle("/hello", chain(http.HandlerFunc(helloHandler)))

    http.ListenAndServe(":8080", mux)
}

The execution order for a request to /hello:

RecoveryMiddleware (enter)
  LoggingMiddleware (enter, start timer)
    TimeoutMiddleware (enter, start context)
      AuthMiddleware (enter, validate token)
        helloHandler (run)
      AuthMiddleware (exit)
    TimeoutMiddleware (exit)
  LoggingMiddleware (exit, log result)
RecoveryMiddleware (exit)

Each middleware wraps the next. The chain is just deeply nested function calls, resolved at request time.

When to Reach for a Framework

This pattern scales well. But there are cases where the standard library genuinely falls short:

Path parameters: net/http's ServeMux gained basic wildcard support in Go 1.22 (/users/{id}), but complex routing (regex, optional segments) still requires a router like Chi or httprouter.
Middleware on route groups: applying middleware to /api/ but not /healthz requires careful mux setup or a router that supports route groups natively.

For most APIs, the standard library plus a lightweight router like Chi (which uses this exact middleware pattern under the hood) is the right call.

What You Built

From scratch, using only the standard library:

A reusable Chain function for composing middleware in order
Structured request logging with response code capture
Panic recovery with stack traces
Token authentication with context propagation
Configurable request timeouts

And more importantly: you now know what a framework's middleware stack actually is. It's this. Just wrapped in a nicer API.

If this was useful, the obvious next step is reading the source of Chi's middleware package. It follows exactly this pattern and is clean, readable Go. Link in the tags above.

Your Python Pre-commit Hook Took 18 Seconds. Ruff Does It in 0.3.

Shayan Holakouee — Wed, 03 Jun 2026 15:43:52 +0000

Here is a pre-commit hook I used to have:

black .
isort .
flake8 .
pylint src/

Four tools. Four configs. Four things that could break. And on a mid-sized project, about 18 seconds of waiting every time I committed.

Now I have this:

ruff check . --fix
ruff format .

Two commands. One tool. 0.3 seconds. Same coverage, fewer headaches.

That tool is Ruff, and if you're still running the old stack, this article will convince you to switch.

What Is Ruff?

Ruff is a Python linter and formatter written in Rust. It was built by Astral, the same team behind uv. The pitch is simple: replace Flake8, Black, isort, pyupgrade, and autoflake with one binary that runs 10 to 100 times faster than any of them individually.

It's not a wrapper around those tools. Ruff re-implements their rules directly in Rust, which is why it's so fast and why it has no Python dependencies at all.

As of 2025, Ruff ships with over 900 lint rules. It's used in production by NumPy, Pandas, FastAPI, Airflow, and thousands of other projects.

Installing Ruff

# Via pip
pip install ruff

# Via uv (recommended)
uv add --dev ruff

# Standalone binary (no Python required)
curl -LsSf https://astral.sh/ruff/install.sh | sh

Check it works:

ruff --version
# ruff 0.x.x

The Two Commands You'll Use Every Day

Ruff has two main modes: check (linting) and format (formatting).

# Lint your code
ruff check .

# Lint and auto-fix what can be fixed
ruff check . --fix

# Format your code (Black-compatible)
ruff format .

# Check formatting without changing files
ruff format . --check

That's the whole surface area for day-to-day use. Everything else is configuration.

What Does It Actually Catch?

Let's take a messy file:

import os
import sys
import json

def greet(name):
    x = 42
    if name == None:
        return "nobody"
    msg = "Hello, %s" % name
    return msg

Run ruff check messy.py:

messy.py:3:1: F401 `json` imported but unused
messy.py:6:5: F841 Local variable `x` is assigned to but never used
messy.py:7:12: E711 Comparison to `None` (use `is None`)
messy.py:8:11: UP031 Use f-string instead of `%`-formatting
Found 4 errors.

Now run ruff check messy.py --fix and look at the result:

import os
import sys


def greet(name):
    if name is None:
        return "nobody"
    msg = f"Hello, {name}"
    return msg

Unused import removed. == None fixed to is None. %-formatting upgraded to an f-string. All automatically.

The x = 42 unused variable is flagged but not auto-fixed because removing a variable assignment could theoretically change behavior in edge cases. Ruff is conservative about what it touches automatically.

Replacing isort

Import sorting is built in. No separate tool, no separate config.

# Before
import sys
import os
from collections import defaultdict
import json
from typing import Optional

# After ruff check . --fix
import json
import os
import sys
from collections import defaultdict
from typing import Optional

Standard library, third-party, and local imports get grouped and sorted correctly. Same behavior as isort, no extra setup.

Enable it explicitly in your config if it's not running by default:

[tool.ruff.lint]
select = ["I"]  # isort rules

Configuration in pyproject.toml

Ruff reads from pyproject.toml, which means no separate .flake8, .isort.cfg, or setup.cfg. One file rules everything.

A solid starting config:

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = [
    "E",    # pycodestyle errors
    "W",    # pycodestyle warnings
    "F",    # pyflakes
    "I",    # isort
    "B",    # flake8-bugbear
    "UP",   # pyupgrade
    "RUF",  # ruff-specific rules
]
ignore = [
    "E501",  # line too long (handled by formatter)
]

[tool.ruff.format]
quote-style = "double"
indent-style = "space"

The select field is where you pick which rule categories to enable. By default Ruff only runs E and F rules, which is already solid coverage. Adding B (bugbear) and UP (pyupgrade) gets you a lot more value with very little noise.

Rule Categories Worth Knowing

Ruff organizes its 900+ rules into prefixed categories. Here are the ones worth enabling on most projects:

Prefix	Origin	What It Catches
`F`	Pyflakes	Unused imports, undefined names, redefined vars
`E` / `W`	pycodestyle	PEP 8 style issues
`I`	isort	Import ordering
`B`	flake8-bugbear	Common bugs and design issues
`UP`	pyupgrade	Outdated syntax (pre-f-strings, old union types)
`N`	pep8-naming	Naming conventions
`RUF`	Ruff-native	Rules with no upstream equivalent
`SIM`	flake8-simplify	Code that can be simplified
`ANN`	flake8-annotations	Missing type annotations

You can look up any rule at docs.astral.sh/ruff/rules. Each one has an explanation and an example.

Ignoring Rules Inline

Sometimes you genuinely need to break a rule. The syntax is the same as Flake8:

import os  # noqa: F401

Or disable for an entire block:

# ruff: noqa: E501
some_very_long_string = "this line is intentionally long because it has to be"

Or skip formatting for a section:

# fmt: off
matrix = [
    1, 0, 0,
    0, 1, 0,
    0, 0, 1,
]
# fmt: on

Editor Integration

VS Code

Install the Ruff extension and add this to your settings:

{
  "editor.formatOnSave": true,
  "editor.defaultFormatter": "charliermarsh.ruff",
  "[python]": {
    "editor.codeActionsOnSave": {
      "source.fixAll.ruff": "explicit",
      "source.organizeImports.ruff": "explicit"
    }
  }
}

Format on save, import sort on save, lint feedback in the gutter. Done.

Neovim

Via conform.nvim and nvim-lint, or through the LSP with nvim-lspconfig. The Ruff LSP server (ruff server) ships with the Ruff binary.

Pre-commit Integration

This is where Ruff's speed makes the biggest practical difference. Replace your old hooks with:

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.9.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format

Two hooks, no Python dependencies, sub-second feedback. Your teammates will stop disabling pre-commit.

Migrating From the Old Stack

If you're coming from Black + isort + Flake8, the migration path is low-friction.

Ruff's formatter is designed to be Black-compatible, so your diffs should be minimal after switching. Start by running both and comparing output:

black --check .
ruff format --check .

For linting, start conservative: just enable E, F, and I. Fix the violations, then expand select to add more rule categories as your team is ready.

You can also silence existing violations in bulk without fixing them using # noqa comments or per-file-ignores, which lets you enforce Ruff on new code while leaving legacy files alone.

One Honest Caveat

Ruff does not do type checking. It does not replace mypy or pyright. It can catch some type-annotation issues (via the ANN rules) and enforce annotation style, but it does not perform type inference.

If you want static type checking, you still need mypy or pyright alongside Ruff. They solve different problems.

The Bottom Line

Ruff is not an incremental improvement on Python linting. It's a full reset: one tool, one config file, no dependency conflicts, no wrapper scripts, and feedback fast enough that you stop dreading the pre-commit hook.

If you're starting a new project, add it as a dev dependency on day one and configure it in pyproject.toml. If you're on an existing project, swap in the formatter first since it's Black-compatible and then layer in linting rules from there.

The Python ecosystem moved slowly on tooling for a long time. Ruff is what it looks like when someone decides to stop accepting that.

Enjoyed this? I wrote a similar piece on uv, the package manager from the same team, if you want to go further down the Astral rabbit hole. Drop a reaction or share it with someone still waiting 18 seconds for their pre-commit to finish.

Stop Juggling 5 Tools , Python's uv Does It All (And It's Blazing Fast)

Shayan Holakouee — Tue, 02 Jun 2026 21:38:43 +0000

If you've been writing Python for more than a year, you know the ritual. A new project. A fresh terminal. And then:

pyenv install 3.12.3
pyenv local 3.12.3
python -m venv .venv
source .venv/bin/activate
pip install pip --upgrade
pip install -r requirements.txt

Six commands before you've written a single line of code. And that's if nothing breaks.

Enter uv a single binary that replaces pip, virtualenv, pip-tools, pyenv, and pipx. Written in Rust. 10–100x faster than pip. And honestly, one of the most pleasant tools I've used in the Python ecosystem in years.

Let's dig into it.

What Even Is `uv`?

uv is a Python package and project manager built by Astral, the same team behind ruff, the linter that everyone switched to and never looked back. The goal is simple: be the Cargo for Python. One tool, one lockfile, no friction.

It's a standalone binary with zero Python dependencies, which means it works even before Python is installed.

Installing uv

# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# Or via pip if you prefer
pip install uv

Verify:

uv --version
# uv 0.9.x

The Speed Claim Is It Real?

Yes. Embarrassingly so.

Here's a timed comparison on Apple Silicon (Python 3.14):

Operation	`pip` / `venv`	`uv`
Create virtual env	~2 seconds	35 milliseconds
Install FastAPI + deps (cold)	~12s	~1.2s
Install with warm cache	~8s	~0.1s

The warm cache case is where uv really shines it uses a global cache and hard-links packages into environments instead of copying them. If you've installed requests in any previous project, your next project gets it nearly instantly.

Starting a New Project

This is where uv feels like a completely different world:

uv init my-api
cd my-api

That single command gives you:

my-api/
├── .git/
├── .venv/          ← already created
├── .python-version
├── pyproject.toml
├── README.md
└── main.py

No separate python -m venv, no git init, no template copying. It's all there.

Managing Python Versions

Before uv, this meant pyenv. Now:

# Install any Python version
uv python install 3.12

# Pin a version for the current project
uv python pin 3.11

# List installed versions
uv python list

uv manages its own Python installations, independently of whatever system Python you have. No more which python3 confusion.

Adding and Removing Dependencies

# Add a package (updates pyproject.toml + uv.lock)
uv add fastapi

# Add a dev-only dependency
uv add --dev pytest ruff

# Remove a package
uv remove requests

# Sync your environment to match the lockfile
uv sync

Notice what's missing: no pip install, no pip freeze > requirements.txt, no manually editing pyproject.toml. uv add handles all of it atomically.

The uv.lock file is cross-platform and cross-Python-version one lockfile that works for your whole team regardless of OS.

Running Commands Without Activating the Venv

This is a small thing that makes a big difference in practice:

# Instead of: source .venv/bin/activate && python main.py
uv run main.py

# Run a dev tool
uv run pytest

# Run ruff
uv run ruff check .

uv run automatically uses the project's virtual environment. You never have to think about activation again.

Replacing `pipx` Running One-off Tools

pipx lets you run CLI tools in isolated environments. uv does this too:

# Run a tool without installing it globally
uvx ruff check .
uvx black --check src/

# Or install it as a persistent tool
uv tool install httpie

uvx is the npx of Python. Run a tool once, don't pollute your environment.

Inline Script Dependencies

This is a lesser-known feature that's actually great for scripting:

# /// script
# requires-python = ">=3.11"
# dependencies = ["httpx", "rich"]
# ///

import httpx
from rich import print

response = httpx.get("https://api.github.com/zen")
print(f"[bold green]{response.text}[/bold green]")

Run it with:

uv run my_script.py

uv reads the inline metadata, creates a temporary isolated environment, installs httpx and rich, and runs the script. No manual setup. Perfect for one-off automation scripts you share with teammates.

Migrating an Existing Project

Already have a requirements.txt or a Poetry project? Migration is low-friction:

# From requirements.txt
uv init
uv add $(cat requirements.txt)

# Or import directly
uv pip install -r requirements.txt  # pip-compatible mode

uv also has a pip-compatible interface (uv pip install, uv pip freeze, etc.) for teams that want the speed gains without changing their workflow immediately.

The Mental Model Shift

Here's the thing that took me a moment to internalize. With traditional Python tooling, you manage layers:

pyenv → Python version
venv  → Isolated environment
pip   → Package installation
pip-tools → Lockfile generation
pipx  → Global tool isolation

Each layer has its own state, its own config, its own failure modes.

With uv, the mental model is much flatter:

uv → everything

One tool manages Python versions, environments, dependencies, lockfiles, and tool execution. The cognitive overhead just... goes away.

When to Still Be Careful

uv is production-ready but worth knowing its quirks:

Conda environments: uv doesn't replace conda. If you're in data science and relying on conda for native packages or CUDA builds, uv isn't a full substitute yet.
Complex build requirements: Packages that need special compilers or system libs work fine, but you'll need those system deps present just like with pip.
It's still pre-1.0: uv is at ~0.9.x. The API is stable for daily use, but if you're building enterprise tooling around it, keep an eye on the changelog.

The Bottom Line

uv is the Python tooling reset the ecosystem needed. It's not experimental it's fast, pragmatic, and genuinely replaces a stack of tools that have been awkward to use together for years.

For new projects: start with uv init. You won't miss the old way.

For existing projects: try swapping in uv pip install for pip first. It's a one-character change and you'll immediately notice the speed difference. From there, adopting the full workflow is a natural next step.

If you found this useful, drop a ❤️ or share it with someone still writing six commands to start a project. The uv docs at docs.astral.sh/uv are genuinely excellent if you want to go deeper.

FastAPI Patterns You Should Be Using in Production

Shayan Holakouee — Tue, 26 May 2026 07:43:59 +0000

FastAPI is already fast to build with — but are you using it to its full potential?

Introduction

Most FastAPI tutorials stop at CRUD routes, Pydantic models, and uvicorn main:app --reload. That's fine for getting started, but production applications demand more: dependency injection at scale, background tasks, custom middleware, event-driven design, and performance tuning.

This article dives deep into the patterns that separate a toy project from a production-grade FastAPI service.

Prerequisites: Comfortable with FastAPI basics, async Python, and Pydantic v2.

1. Dependency Injection — The Right Way

FastAPI's Depends() system is incredibly powerful, but most people only scratch the surface with database sessions. Let's go deeper.

Layered Dependencies

from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_token(credentials: HTTPAuthorizationCredentials = Security(security)) -> str:
    return credentials.credentials

async def get_current_user(token: str = Depends(get_token)) -> User:
    user = await decode_jwt(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def require_admin(user: User = Depends(get_current_user)) -> User:
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admins only")
    return user

You now have a clean chain: token → user → admin. Each layer is independently testable and composable.

Class-Based Dependencies

For stateful dependencies (rate limiters, feature flags, DB repos), classes are superior:

class PaginationParams:
    def __init__(self, page: int = 1, size: int = 20, max_size: int = 100):
        if size > max_size:
            raise HTTPException(400, f"Page size cannot exceed {max_size}")
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@router.get("/items")
async def list_items(pagination: PaginationParams = Depends()):
    return await Item.find_all(offset=pagination.offset, limit=pagination.size)

Dependency Overrides in Tests

This is where FastAPI truly shines for testing:

# In your test file
app.dependency_overrides[get_current_user] = lambda: User(id=1, role="admin")

async def test_admin_endpoint(client: AsyncClient):
    response = await client.get("/admin/dashboard")
    assert response.status_code == 200

No mocking. No patching. Just clean overrides.

2. Lifespan Events — Replacing `@app.on_event`

The old @app.on_event("startup") is deprecated. Use the lifespan context manager instead — it's cleaner and handles errors properly:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize resources
    app.state.db_pool = await create_db_pool()
    app.state.redis = await create_redis_client()
    app.state.http_client = httpx.AsyncClient()
    print("✅ Application started")

    yield  # Application runs here

    # Shutdown: release resources
    await app.state.db_pool.close()
    await app.state.redis.close()
    await app.state.http_client.aclose()
    print("🛑 Application shut down cleanly")

app = FastAPI(lifespan=lifespan)

Access these anywhere via request.app.state.redis.

3. Custom Middleware — Beyond `CORSMiddleware`

Request ID Middleware

Every request in production should have a traceable ID:

import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id

        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

app.add_middleware(RequestIDMiddleware)

Timing Middleware

import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000
        response.headers["X-Process-Time-Ms"] = f"{duration_ms:.2f}"
        return response

⚠️ Gotcha: BaseHTTPMiddleware adds overhead. For ultra-high throughput, use pure ASGI middleware or starlette.routing directly.

4. Background Tasks vs. Celery — When to Use What

FastAPI's Built-in Background Tasks

Perfect for fire-and-forget operations that don't need retries or monitoring:

from fastapi import BackgroundTasks

async def send_welcome_email(email: str, name: str):
    await email_client.send(
        to=email,
        subject=f"Welcome, {name}!",
        body="Thanks for signing up."
    )

@router.post("/users", status_code=201)
async def create_user(data: UserCreate, background_tasks: BackgroundTasks):
    user = await User.create(**data.model_dump())
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    return user  # Returns immediately; email sends in background

When to Use Celery (or ARQ/Dramatiq)

Use a proper task queue when you need:

Retry logic on failure
Scheduled tasks (cron-like)
Progress tracking
Rate limiting per worker
Distributed processing across multiple machines

# With ARQ (async-native, pairs beautifully with FastAPI)
async def process_video(ctx, video_id: int):
    video = await Video.get(video_id)
    await transcode(video.path)
    await video.update(status="processed")

# Enqueue from your route
@router.post("/videos/{video_id}/process")
async def trigger_processing(video_id: int, redis: Redis = Depends(get_redis)):
    await redis.enqueue_job("process_video", video_id)
    return {"status": "queued"}

5. Advanced Pydantic v2 Patterns

Computed Fields

from pydantic import BaseModel, computed_field

class Order(BaseModel):
    items: list[OrderItem]
    discount_pct: float = 0.0

    @computed_field
    @property
    def subtotal(self) -> float:
        return sum(item.price * item.qty for item in self.items)

    @computed_field
    @property
    def total(self) -> float:
        return self.subtotal * (1 - self.discount_pct / 100)

Model Validators

from pydantic import model_validator

class DateRange(BaseModel):
    start_date: date
    end_date: date

    @model_validator(mode="after")
    def check_dates(self) -> "DateRange":
        if self.end_date <= self.start_date:
            raise ValueError("end_date must be after start_date")
        return self

    @computed_field
    @property
    def duration_days(self) -> int:
        return (self.end_date - self.start_date).days

Custom Serialization

from pydantic import field_serializer
from datetime import datetime

class Event(BaseModel):
    name: str
    created_at: datetime

    @field_serializer("created_at")
    def serialize_dt(self, dt: datetime) -> str:
        return dt.strftime("%Y-%m-%d %H:%M UTC")

6. Structured Error Handling

Don't let exceptions leak into your responses. Build a global error system:

from fastapi import Request
from fastapi.responses import JSONResponse

class AppError(Exception):
    def __init__(self, message: str, code: str, status: int = 400):
        self.message = message
        self.code = code
        self.status = status

class NotFoundError(AppError):
    def __init__(self, resource: str, id: int):
        super().__init__(
            message=f"{resource} with id {id} not found",
            code="RESOURCE_NOT_FOUND",
            status=404
        )

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "request_id": getattr(request.state, "request_id", None)
            }
        }
    )

# Usage in routes
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await User.get(user_id)
    if not user:
        raise NotFoundError("User", user_id)
    return user

7. Streaming Responses

Perfect for LLMs, large file downloads, or server-sent events:

from fastapi.responses import StreamingResponse
import asyncio

async def event_stream(topic: str):
    async for event in subscribe_to_events(topic):
        yield f"data: {event.model_dump_json()}\n\n"

@router.get("/events/{topic}")
async def stream_events(topic: str):
    return StreamingResponse(
        event_stream(topic),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Disable nginx buffering
        }
    )

For file streaming:

@router.get("/exports/{report_id}")
async def download_report(report_id: int):
    async def generate():
        async for chunk in generate_csv_chunks(report_id):
            yield chunk

    return StreamingResponse(
        generate(),
        media_type="text/csv",
        headers={"Content-Disposition": f"attachment; filename=report_{report_id}.csv"}
    )

8. Router Organization at Scale

As your app grows, structure matters:

app/
├── main.py
├── core/
│   ├── config.py       # Settings via pydantic-settings
│   ├── security.py     # Auth dependencies
│   └── database.py     # DB pool setup
├── api/
│   ├── __init__.py
│   ├── v1/
│   │   ├── router.py   # Aggregates all v1 routes
│   │   ├── users.py
│   │   ├── orders.py
│   │   └── products.py
│   └── v2/
│       └── router.py
├── models/
├── schemas/
└── services/

# api/v1/router.py
from fastapi import APIRouter
from .users import router as users_router
from .orders import router as orders_router

v1_router = APIRouter(prefix="/v1")
v1_router.include_router(users_router, prefix="/users", tags=["Users"])
v1_router.include_router(orders_router, prefix="/orders", tags=["Orders"])

# main.py
app.include_router(v1_router, prefix="/api")

9. Performance: Async Done Right

The #1 Async Mistake

Running blocking I/O in async routes kills your throughput. Use run_in_executor or asyncio.to_thread:

import asyncio
from PIL import Image  # Blocking library

@router.post("/images/resize")
async def resize_image(file: UploadFile):
    contents = await file.read()

    # ✅ Offload blocking CPU work to thread pool
    result = await asyncio.to_thread(process_image_sync, contents)
    return {"size": len(result)}

def process_image_sync(data: bytes) -> bytes:
    img = Image.open(BytesIO(data))
    img = img.resize((800, 600))
    buf = BytesIO()
    img.save(buf, format="JPEG")
    return buf.getvalue()

Response Caching with Redis

import json
from functools import wraps

def cache(key_prefix: str, ttl: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, redis=None, **kwargs):
            cache_key = f"{key_prefix}:{':'.join(str(v) for v in kwargs.values())}"
            cached = await redis.get(cache_key)
            if cached:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await redis.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@router.get("/products/{product_id}")
@cache("product", ttl=300)
async def get_product(product_id: int, redis=Depends(get_redis)):
    return await Product.get(product_id)

10. Testing Async FastAPI Properly

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.core.database import get_db

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.fixture(autouse=True)
async def override_db(test_db_session):
    app.dependency_overrides[get_db] = lambda: test_db_session
    yield
    app.dependency_overrides.clear()

async def test_create_user(client: AsyncClient):
    response = await client.post("/api/v1/users", json={
        "email": "test@example.com",
        "name": "Test User"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "id" in data

Wrapping Up

Here's what we covered:

Dependency injection at scale with layered and class-based dependencies
Lifespan events for proper resource management
Custom middleware for observability
Background tasks and when to reach for a proper queue
Pydantic v2 advanced features: computed fields, validators, serializers
Structured error handling with custom exception hierarchies
Streaming responses for real-time and large data
Router organization for growing codebases
Performance patterns: async threading and response caching
Async testing with httpx.AsyncClient

FastAPI gives you incredible primitives. The difference between a fragile prototype and a resilient service is how deliberately you compose them.

What advanced FastAPI patterns have you found indispensable? Drop them in the comments!

Building a Custom DSL in Python, From Tokenizer to Interpreter

Shayan Holakouee — Sat, 09 May 2026 16:40:14 +0000

Most backend engineers reach for an existing query language or config format and call it done. That works until it doesn't. Maybe your team keeps writing thousand-line YAML files that nobody wants to touch. Maybe you need a rule engine where non-engineers can define logic without deploying code. Maybe you just want to understand what actually happens when Python parses 1 + 2 * 3.

Building a domain-specific language (DSL) forces you to confront parsing, evaluation, and language design in ways that make you a sharper engineer. This article walks through building a real one from scratch: a small expression language for defining data transformation pipelines.

No parser-combinator libraries. No ANTLR grammars. Just Python.

What We Are Building

Our DSL will let users write transformation rules like this:

FILTER age > 18
MAP name -> upper(name)
REDUCE count(*)

By the end, you will have a working interpreter that can parse this syntax, build an AST, and evaluate it against a list of dictionaries. The concepts generalize directly to anything from config languages to query engines.

Step 1: The Lexer (Tokenizer)

The lexer's job is to turn raw text into a stream of typed tokens. Think of it as the part of the pipeline that stops caring about characters and starts caring about meaning.

from dataclasses import dataclass
from enum import Enum, auto
import re

class TokenType(Enum):
    KEYWORD    = auto()
    IDENTIFIER = auto()
    NUMBER     = auto()
    STRING     = auto()
    OPERATOR   = auto()
    ARROW      = auto()
    LPAREN     = auto()
    RPAREN     = auto()
    COMMA      = auto()
    NEWLINE    = auto()
    EOF        = auto()

@dataclass
class Token:
    type: TokenType
    value: str
    line: int

KEYWORDS = {"FILTER", "MAP", "REDUCE"}

TOKEN_PATTERNS = [
    (TokenType.ARROW,      r'->'),
    (TokenType.OPERATOR,   r'[><=!]+|and|or|not'),
    (TokenType.NUMBER,     r'\d+(\.\d+)?'),
    (TokenType.STRING,     r'"[^"]*"'),
    (TokenType.LPAREN,     r'\('),
    (TokenType.RPAREN,     r'\)'),
    (TokenType.COMMA,      r','),
    (TokenType.NEWLINE,    r'\n'),
    (TokenType.IDENTIFIER, r'[A-Za-z_][A-Za-z0-9_]*'),
]

class Lexer:
    def __init__(self, source: str):
        self.source = source
        self.pos = 0
        self.line = 1

    def tokenize(self) -> list[Token]:
        tokens = []
        while self.pos < len(self.source):
            if self.source[self.pos] in ' \t':
                self.pos += 1
                continue

            matched = False
            for token_type, pattern in TOKEN_PATTERNS:
                regex = re.compile(pattern)
                m = regex.match(self.source, self.pos)
                if m:
                    value = m.group(0)
                    if token_type == TokenType.IDENTIFIER and value in KEYWORDS:
                        token_type = TokenType.KEYWORD
                    if token_type == TokenType.NEWLINE:
                        self.line += 1
                    tokens.append(Token(token_type, value, self.line))
                    self.pos = m.end()
                    matched = True
                    break

            if not matched:
                raise SyntaxError(
                    f"Unexpected character {self.source[self.pos]!r} at line {self.line}"
                )

        tokens.append(Token(TokenType.EOF, '', self.line))
        return tokens

A few things worth calling out here. The order of TOKEN_PATTERNS matters: -> has to appear before > or it will never match. The keyword check happens after identifier matching so we don't need a separate pattern for each keyword. Both are common gotchas.

Step 2: The AST Nodes

Before writing the parser, define the data structures it will produce. The Abstract Syntax Tree (AST) is the clean, structured representation of your program. Each node type corresponds to a concept in your language.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class NumberLiteral:
    value: float

@dataclass
class StringLiteral:
    value: str

@dataclass
class Identifier:
    name: str

@dataclass
class BinaryOp:
    left: Any
    op: str
    right: Any

@dataclass
class FunctionCall:
    name: str
    args: list = field(default_factory=list)

@dataclass
class FilterStatement:
    condition: Any

@dataclass
class MapStatement:
    source: str
    target: Any

@dataclass
class ReduceStatement:
    aggregation: Any

@dataclass
class Program:
    statements: list = field(default_factory=list)

Keep AST nodes as dumb data containers. Resist the urge to put evaluation logic inside them. That belongs in the interpreter, and keeping them separate makes the whole system easier to extend and test.

Step 3: The Parser

The parser consumes a flat list of tokens and produces the nested AST. We will use a recursive descent parser, which is the most approachable style and perfectly adequate for small languages.

class Parser:
    def __init__(self, tokens: list[Token]):
        self.tokens = tokens
        self.pos = 0

    def current(self) -> Token:
        return self.tokens[self.pos]

    def peek(self, offset=1) -> Token:
        idx = self.pos + offset
        return self.tokens[idx] if idx < len(self.tokens) else self.tokens[-1]

    def consume(self, expected_type: TokenType = None, expected_value: str = None) -> Token:
        token = self.current()
        if expected_type and token.type != expected_type:
            raise SyntaxError(
                f"Expected {expected_type}, got {token.type} ({token.value!r}) at line {token.line}"
            )
        if expected_value and token.value != expected_value:
            raise SyntaxError(
                f"Expected {expected_value!r}, got {token.value!r} at line {token.line}"
            )
        self.pos += 1
        return token

    def skip_newlines(self):
        while self.current().type == TokenType.NEWLINE:
            self.pos += 1

    def parse(self) -> Program:
        statements = []
        self.skip_newlines()
        while self.current().type != TokenType.EOF:
            stmt = self.parse_statement()
            statements.append(stmt)
            self.skip_newlines()
        return Program(statements)

    def parse_statement(self):
        token = self.current()
        if token.type != TokenType.KEYWORD:
            raise SyntaxError(f"Expected keyword at line {token.line}, got {token.value!r}")

        if token.value == "FILTER":
            return self.parse_filter()
        elif token.value == "MAP":
            return self.parse_map()
        elif token.value == "REDUCE":
            return self.parse_reduce()
        else:
            raise SyntaxError(f"Unknown keyword: {token.value}")

    def parse_filter(self) -> FilterStatement:
        self.consume(TokenType.KEYWORD, "FILTER")
        condition = self.parse_expression()
        return FilterStatement(condition)

    def parse_map(self) -> MapStatement:
        self.consume(TokenType.KEYWORD, "MAP")
        source = self.consume(TokenType.IDENTIFIER).value
        self.consume(TokenType.ARROW)
        target = self.parse_expression()
        return MapStatement(source, target)

    def parse_reduce(self) -> ReduceStatement:
        self.consume(TokenType.KEYWORD, "REDUCE")
        aggregation = self.parse_expression()
        return ReduceStatement(aggregation)

    def parse_expression(self):
        left = self.parse_primary()
        if self.current().type == TokenType.OPERATOR:
            op = self.consume(TokenType.OPERATOR).value
            right = self.parse_primary()
            return BinaryOp(left, op, right)
        return left

    def parse_primary(self):
        token = self.current()

        if token.type == TokenType.NUMBER:
            self.pos += 1
            return NumberLiteral(float(token.value))

        if token.type == TokenType.STRING:
            self.pos += 1
            return StringLiteral(token.value.strip('"'))

        if token.type == TokenType.IDENTIFIER:
            # Check if this is a function call
            if self.peek().type == TokenType.LPAREN:
                return self.parse_function_call()
            self.pos += 1
            return Identifier(token.value)

        raise SyntaxError(f"Unexpected token {token.value!r} at line {token.line}")

    def parse_function_call(self) -> FunctionCall:
        name = self.consume(TokenType.IDENTIFIER).value
        self.consume(TokenType.LPAREN)
        args = []
        while self.current().type != TokenType.RPAREN:
            args.append(self.parse_expression())
            if self.current().type == TokenType.COMMA:
                self.consume(TokenType.COMMA)
        self.consume(TokenType.RPAREN)
        return FunctionCall(name, args)

Recursive descent parsers mirror the grammar almost 1:1. Each parse_* method corresponds to a grammar rule. Once you see that structure, the parser stops feeling like magic.

Step 4: The Interpreter

Now the fun part. The interpreter walks the AST and does real work. We will use the visitor pattern via singledispatch, which is a cleaner alternative to giant isinstance chains.

from functools import singledispatch
from typing import Callable

# Built-in functions available inside the DSL
BUILTINS: dict[str, Callable] = {
    "upper":  lambda s: s.upper(),
    "lower":  lambda s: s.lower(),
    "count":  lambda *args: len(args[0]) if args else 0,
    "length": lambda s: len(s),
    "str":    lambda x: str(x),
}

OPERATORS = {
    ">":  lambda a, b: a > b,
    "<":  lambda a, b: a < b,
    ">=": lambda a, b: a >= b,
    "<=": lambda a, b: a <= b,
    "==": lambda a, b: a == b,
    "!=": lambda a, b: a != b,
    "and": lambda a, b: a and b,
    "or":  lambda a, b: a or b,
}

class Interpreter:
    def __init__(self, builtins: dict = None):
        self.builtins = {**BUILTINS, **(builtins or {})}

    def eval_node(self, node, record: dict):
        """Dispatch to the right evaluation method based on node type."""
        method_name = f"eval_{type(node).__name__}"
        method = getattr(self, method_name, None)
        if method is None:
            raise RuntimeError(f"No evaluator for node type: {type(node).__name__}")
        return method(node, record)

    def eval_NumberLiteral(self, node: NumberLiteral, record: dict):
        return node.value

    def eval_StringLiteral(self, node: StringLiteral, record: dict):
        return node.value

    def eval_Identifier(self, node: Identifier, record: dict):
        if node.name not in record:
            raise NameError(f"Undefined field: {node.name!r}")
        return record[node.name]

    def eval_BinaryOp(self, node: BinaryOp, record: dict):
        left = self.eval_node(node.left, record)
        right = self.eval_node(node.right, record)
        op_fn = OPERATORS.get(node.op)
        if op_fn is None:
            raise RuntimeError(f"Unknown operator: {node.op!r}")
        return op_fn(left, right)

    def eval_FunctionCall(self, node: FunctionCall, record: dict):
        fn = self.builtins.get(node.name)
        if fn is None:
            raise NameError(f"Unknown function: {node.name!r}")
        args = [self.eval_node(arg, record) for arg in node.args]
        return fn(*args)

    def run(self, program: Program, data: list[dict]) -> list[dict]:
        result = list(data)
        for stmt in program.statements:
            result = self.execute_statement(stmt, result)
        return result

    def execute_statement(self, stmt, data: list[dict]) -> list[dict]:
        if isinstance(stmt, FilterStatement):
            return [
                record for record in data
                if self.eval_node(stmt.condition, record)
            ]

        if isinstance(stmt, MapStatement):
            output = []
            for record in data:
                new_record = dict(record)
                new_record[stmt.source] = self.eval_node(stmt.target, record)
                output.append(new_record)
            return output

        if isinstance(stmt, ReduceStatement):
            # For count(*), pass the whole dataset as a single record context
            count = self.eval_node(stmt.aggregation, {"*": data})
            return [{"result": count}]

        raise RuntimeError(f"Unknown statement type: {type(stmt).__name__}")

Step 5: Wiring It All Together

def run_pipeline(source: str, data: list[dict]) -> list[dict]:
    lexer = Lexer(source)
    tokens = lexer.tokenize()

    parser = Parser(tokens)
    program = parser.parse()

    interpreter = Interpreter()
    return interpreter.run(program, data)

Let's run it:

pipeline = """
FILTER age > 18
MAP name -> upper(name)
"""

records = [
    {"name": "alice", "age": 22},
    {"name": "bob", "age": 15},
    {"name": "carol", "age": 31},
]

result = run_pipeline(pipeline, records)
print(result)
# [{'name': 'ALICE', 'age': 22}, {'name': 'CAROL', 'age': 31}]

It works.

Where To Go From Here

This implementation handles the core pipeline, but a production-grade DSL would need a few more things:

Operator precedence. Right now age > 18 and role == "admin" would fail because parse_expression only handles one operator. Pratt parsing (top-down operator precedence) is the standard fix and worth learning.

Better error messages. Error messages are the UX of a programming language. Include the line, the column, and a snippet of the source. Users will thank you.

A type system or validator. An AST visitor that checks field names against a schema before evaluation catches errors early and makes the DSL much safer to expose to end users.

Extensible functions. The builtins dict already accepts custom functions. You can expose a registration decorator to make it feel like a proper plugin system:

dsl = Interpreter()

@dsl.register_function("slugify")
def slugify(s: str) -> str:
    return s.lower().replace(" ", "-")

Caching parsed programs. Parsing is the expensive part. Cache Program objects keyed by source hash so repeated runs of the same pipeline skip the lexer and parser entirely.

Why This Matters Beyond the Exercise

The pattern you just built -- tokenize, parse, evaluate -- shows up everywhere. Template engines, feature flag rule engines, database query builders, workflow automation tools. Once you have built one end to end, you start recognizing the same skeleton in every tool that lets users express logic without writing code.

The implementation here clocks in at around 250 lines. That is small enough to own completely, test thoroughly, and extend confidently. For a lot of internal tooling, that is exactly the right size.

Full Source

The complete working code is available as a single file. Clone it, break it on purpose, and add a feature. That is the best way to actually internalize what is happening.

https://gist.github.com/your-handle/python-dsl-from-scratch

If you end up extending this with a type checker or a Pratt parser, drop it in the comments. Curious what people build on top of it.

Python's AsyncIO Internals, What Actually Happens When You Write await

Shayan Holakouee — Wed, 29 Apr 2026 03:31:01 +0000

You have written async def and await. You know the surface. At some point something deadlocked, or performed worse than expected, or behaved in a way that did not fit your mental model. This article is about building the mental model that makes those situations predictable rather than surprising.

Coroutines Are Not Threads and Not Callbacks

The two common async models before asyncio were threads and callbacks. Threads give you apparent concurrency with shared state and all the locking that entails. Callbacks give you non-blocking IO with inverted control flow that becomes unreadable at scale.

Coroutines are a third model. A coroutine is a function that can suspend its execution at specific points and resume from exactly that point later. The suspension is cooperative: the coroutine decides when to yield control, not a scheduler. The interpreter does not preempt it.

This is the first thing to internalize. Async Python is not parallel. It is concurrent in the sense that multiple coroutines can be in progress at the same time, but only one runs at any given moment. If a coroutine never yields, nothing else runs.

What async def Actually Produces

An async def function does not execute its body when called. It returns a coroutine object.

async def fetch(url):
    print(f"fetching {url}")
    return "data"

coro = fetch("https://example.com")
print(type(coro))   # <class 'coroutine'>
print(coro)         # <coroutine object fetch at 0x...>
# nothing printed yet, the body has not run

The coroutine object implements the coroutine protocol: it has send(), throw(), and close() methods. The event loop drives it by calling send(None) repeatedly until it raises StopIteration, at which point the return value is carried in the exception's value attribute.

You can drive a simple coroutine manually to see this:

async def simple():
    return 42

coro = simple()
try:
    coro.send(None)
except StopIteration as e:
    print(e.value)  # 42

This is exactly what the event loop does, wrapped in scheduling logic. The await keyword and the event loop are two sides of the same mechanism.

What await Actually Does

await can only appear inside async def. It takes an awaitable object (a coroutine, a Task, a Future, or any object implementing __await__) and suspends the current coroutine until the awaitable completes.

At the bytecode level, await expr compiles to GET_AWAITABLE followed by YIELD_FROM. YIELD_FROM repeatedly sends values into the inner coroutine and yields values out to the outer caller until the inner coroutine raises StopIteration. The value of the StopIteration exception becomes the result of the await expression.

The chain goes all the way down. When you await a coroutine that awaits a Future, the YIELD_FROM instructions chain together. The innermost Future is what actually suspends execution by yielding a non-StopIteration value up through the chain to the event loop. The event loop receives that value (typically the Future itself), registers a callback to resume the coroutine when the Future completes, and moves on to other work.

async def inner():
    await asyncio.sleep(1)  # eventually yields a Future to the event loop
    return "done"

async def outer():
    result = await inner()  # chains the yield through
    print(result)

The event loop never sees inner or outer directly during the sleep. It sees a Future. When the timer fires, it resolves the Future, which triggers the callback that resumes inner, which completes and triggers the callback that resumes outer.

The Event Loop Is a Scheduling Loop

The event loop is simpler than most people expect. At its core it is a loop that does three things: run ready callbacks, poll IO for completeness, and schedule callbacks for completed IO.

A simplified version of CPython's event loop logic looks like this:

while True:
    # Run all callbacks that are ready right now
    while self._ready:
        handle = self._ready.popleft()
        handle._run()

    # Poll IO with a timeout
    # (uses select/epoll/kqueue depending on platform)
    timeout = self._compute_timeout()
    event_list = self._selector.select(timeout)

    # Schedule callbacks for completed IO events
    self._process_events(event_list)

    # Run scheduled callbacks whose time has come
    self._run_once_scheduled()

asyncio.sleep(1) works by creating a Future, scheduling a callback to resolve it after one second via call_later, and returning the Future for the coroutine to await. When the event loop's timer fires, it resolves the Future, which schedules the coroutine's resumption as a ready callback. On the next iteration of the loop, the coroutine resumes.

No thread is involved. The OS timer is polled during selector.select(). The event loop handles the rest.

Tasks vs Futures vs Coroutines

These three are related but distinct and conflating them causes confusion.

A coroutine is a generator-like object produced by an async def call. It does nothing on its own. It must be driven by something calling send() on it.

A Future is a low-level object representing a value that will exist at some point. It has states: pending, cancelled, done. When done, it holds either a result or an exception. You can attach callbacks that fire when it completes. Most application code never creates Future objects directly.

A Task is a subclass of Future that wraps a coroutine and drives it. When you call asyncio.create_task(coro), you get a Task that is immediately scheduled on the event loop. The Task calls coro.send(None) to start the coroutine, catches StopIteration when it completes, and resolves itself with the result.

async def work():
    await asyncio.sleep(0.1)
    return 42

async def main():
    # coroutine: not scheduled yet
    coro = work()

    # Task: immediately scheduled, starts running on the next event loop iteration
    task = asyncio.create_task(work())

    # awaiting the coroutine directly: runs inline, no separate task
    result = await coro

    # awaiting the task: waits for the already-running task
    result = await task

The difference between await coro and await asyncio.create_task(coro) is whether the coroutine runs as a separate scheduled unit or inline in the current coroutine's execution. For concurrency, you need tasks. Two await statements in sequence run sequentially. Two tasks run concurrently.

Blocking the Event Loop Is the Canonical Mistake

Because async Python is single-threaded cooperative concurrency, any code that does not yield blocks everything else. There is no preemption.

import time

async def bad():
    time.sleep(2)  # blocks the OS thread for 2 seconds
    return "done"

async def also_running():
    await asyncio.sleep(0)
    print("I run eventually")

async def main():
    await asyncio.gather(bad(), also_running())
    # also_running does not run until bad() returns
    # because time.sleep never yields to the event loop

time.sleep is a syscall that parks the OS thread. The event loop cannot run while the thread is parked. asyncio.sleep schedules a timer and yields a Future to the event loop, allowing other tasks to run during the wait.

The same problem applies to CPU-heavy code. A tight loop does not yield. A large file read through blocking IO does not yield. Any call to a synchronous library that does IO internally does not yield.

The fix is asyncio.to_thread (Python 3.9+) or loop.run_in_executor, which run blocking code in a thread pool. The coroutine awaits the thread pool result, yielding to the event loop while the thread runs:

async def process_file(path):
    # runs in a thread pool, event loop remains free
    content = await asyncio.to_thread(open(path).read)
    return content

await and Custom Awaitables

Any object can be awaitable by implementing __await__, which must return an iterator. This is the extension point for writing your own low-level async primitives.

class SleepUntilNextSecond:
    def __await__(self):
        loop = asyncio.get_event_loop()
        future = loop.create_future()

        import time
        delay = 1.0 - (time.time() % 1.0)
        loop.call_later(delay, future.set_result, None)

        yield from future.__await__()
        return "woke up"


async def main():
    result = await SleepUntilNextSecond()
    print(result)  # "woke up"

yield from future.__await__() chains into the Future's awaitable protocol, which ultimately yields the Future object up to the event loop. This is the same mechanism asyncio.sleep uses internally. Writing custom awaitables is how libraries like aiohttp and asyncpg integrate with the event loop at a level below async def.

Cancellation Is Cooperative

task.cancel() does not kill a task. It schedules a CancelledError to be thrown into the coroutine at its next await point. The coroutine receives it, and if it does not catch and suppress it, the task is cancelled.

async def careful_work():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        # do cleanup here
        raise  # re-raise to actually cancel

async def main():
    task = asyncio.create_task(careful_work())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task was cancelled")

The raise after cleanup is important. A coroutine that catches CancelledError and does not re-raise it has suppressed the cancellation. The task will appear to complete normally rather than being cancelled. This breaks asyncio.wait_for timeouts and structured concurrency patterns that depend on cancellation propagating correctly.

The Mental Model

Async Python is a single-threaded scheduler running coroutines that explicitly yield at await points. The event loop runs callbacks in order. IO completion and timers register callbacks. Coroutines are suspended at await and resumed when their awaited object resolves.

Everything that works correctly in async Python works because coroutines yield frequently. Everything that breaks does so because something held the thread without yielding. The event loop cannot fix that. It can only schedule what yields to it.

Python's Data Model Is an API. Here Is How to Use It Properly

Shayan Holakouee — Wed, 29 Apr 2026 03:24:13 +0000

Most Python developers know that __len__ makes len() work and __add__ makes + work. That is the surface. The actual story is that Python's data model is a coherent, documented protocol through which user-defined objects can participate in the language itself: not just operators, but truthiness, hashing, iteration, context management, attribute access, memory layout, and more. Using it well means understanding what CPython actually calls, in what order, and why.

The Interpreter Calls These, Not You

The first thing to internalize: dunder methods are called by the interpreter, not by user code. When you write len(obj), Python calls type(obj).__len__(obj). Not obj.__len__(). The lookup goes through the type, not the instance.

This matters for two reasons.

First, defining a dunder on an instance rather than the class does not work:

class MyClass:
    pass

obj = MyClass()
obj.__len__ = lambda: 42

len(obj)  # TypeError: object of type 'MyClass' has no len()

The interpreter checked type(obj).__len__, found nothing, and raised. The instance attribute was ignored entirely.

Second, it means metaclasses can define dunders that apply to the class itself as an object. __len__ on a metaclass makes len(MyClass) work. __iter__ on a metaclass makes for x in MyClass work. The class is an instance of the metaclass, so the metaclass's dunders are the class's dunders.

Truthiness, Equality, and Hashing Are a Triangle

These three are deeply connected and breaking the contract between them is one of the more common ways to introduce subtle bugs.

Truthiness is determined by __bool__. If not defined, Python falls back to __len__: an object with length zero is falsy. If neither is defined, the object is always truthy.

class Container:
    def __init__(self, items):
        self.items = items

    def __len__(self):
        return len(self.items)

c = Container([])
if not c:
    print("empty")  # prints, because __bool__ fell back to __len__

Equality is __eq__. The default is identity comparison (same as is). Override it to define value equality.

Hashing is __hash__. Here is the contract: objects that compare equal must have the same hash. If you define __eq__, Python automatically sets __hash__ to None, making your object unhashable. This is intentional. Mutable objects with value equality should not be hashable because their hash could change when they are mutated, silently breaking set and dict behavior.

class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __eq__(self, other):
        return isinstance(other, Point) and self.x == other.x and self.y == other.y

    # Python set __hash__ = None automatically
    # Point is now unhashable

p = Point(1, 2)
{p}  # TypeError: unhashable type: 'Point'

If you want a hashable value type, define both:

    def __hash__(self):
        return hash((self.x, self.y))

The tuple hash is stable, consistent with equality, and composes well. Never define __hash__ without thinking about what makes two instances equal, and never define __eq__ without thinking about whether you also need __hash__.

Comparison Operators and the Reflected Protocol

__lt__, __le__, __gt__, __ge__, __eq__, __ne__ cover the six comparison operators. Python's mechanism for resolving them is more nuanced than most people realize.

When you write a < b, Python first tries type(a).__lt__(a, b). If that returns NotImplemented, Python tries the reflected operation: type(b).__gt__(b, a). This gives the right operand a chance to handle the comparison when the left operand does not know how.

class Celsius:
    def __init__(self, temp):
        self.temp = temp

    def __lt__(self, other):
        if isinstance(other, Fahrenheit):
            return self.temp < (other.temp - 32) * 5 / 9
        if isinstance(other, Celsius):
            return self.temp < other.temp
        return NotImplemented  # not NotImplemented(), the singleton


class Fahrenheit:
    def __init__(self, temp):
        self.temp = temp

    def __gt__(self, other):
        if isinstance(other, Celsius):
            return self.temp > other.temp * 9 / 5 + 32
        return NotImplemented

NotImplemented is a singleton, not an exception. Returning it tells Python to try the reflected operation. Raising TypeError or returning False are both wrong here: they prevent the fallback.

functools.total_ordering is worth knowing. Define __eq__ and one of the four ordering methods, and it derives the rest. It is slower than defining all four manually but saves considerable boilerplate for types where ordering is well-defined.

Arithmetic Operators and In-Place Variants

The arithmetic protocol has three layers: forward, reflected, and in-place.

a + b tries type(a).__add__(a, b) first. If that returns NotImplemented, it tries type(b).__radd__(b, a). The reflected methods (__radd__, __rsub__, __rmul__, etc.) exist so that user-defined types can work with built-in types on both sides of an operator.

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __add__(self, other):
        if isinstance(other, Vector):
            return Vector(self.x + other.x, self.y + other.y)
        return NotImplemented

    def __radd__(self, other):
        # called when other + self and other.__add__ returned NotImplemented
        return self.__add__(other)

    def __mul__(self, scalar):
        if isinstance(scalar, (int, float)):
            return Vector(self.x * scalar, self.y * scalar)
        return NotImplemented

    def __rmul__(self, scalar):
        return self.__mul__(scalar)

v = Vector(1, 2)
print(3 * v)  # works because int.__mul__(3, v) returns NotImplemented,
              # then Vector.__rmul__(v, 3) is tried

In-place operators (+=, -=, *=) call __iadd__, __isub__, __imul__ and so on. If the in-place method is not defined, Python falls back to the regular method and rebinds the name. For mutable types, define the in-place variants to mutate in place and return self. For immutable types, do not define them and let Python handle the fallback.

def __iadd__(self, other):
    if isinstance(other, Vector):
        self.x += other.x
        self.y += other.y
        return self  # must return self for in-place operations
    return NotImplemented

Attribute Access Is More Controllable Than You Think

The attribute access protocol has four hooks: __getattribute__, __getattr__, __setattr__, and __delattr__.

__getattribute__ is called on every attribute access, including dunders. Overriding it without calling super().__getattribute__ will break your class entirely. It is rarely the right hook.

__getattr__ is called only when normal attribute lookup has failed. It is the right hook for lazy loading, proxying, and dynamic attributes:

class LazyLoader:
    def __init__(self, module_name):
        self._module_name = module_name
        self._module = None

    def __getattr__(self, name):
        if self._module is None:
            import importlib
            self._module = importlib.import_module(self._module_name)
        return getattr(self._module, name)

__setattr__ is called on every attribute assignment, including in __init__. If you override it, you must use object.__setattr__(self, name, value) or super().__setattr__(name, value) to actually store the value, otherwise you get infinite recursion:

class Validated:
    def __setattr__(self, name, value):
        if name == "age" and not isinstance(value, int):
            raise TypeError("age must be int")
        super().__setattr__(name, value)  # not self.name = value

The Container Protocol in Full

To make a proper sequence, you need __len__ and __getitem__. Python derives a surprising amount from these two:

class Fibonacci:
    def __len__(self):
        return 100

    def __getitem__(self, index):
        if isinstance(index, slice):
            return [self[i] for i in range(*index.indices(len(self)))]
        if index < 0:
            index += len(self)
        if not 0 <= index < len(self):
            raise IndexError(index)
        a, b = 0, 1
        for _ in range(index):
            a, b = b, a + b
        return a

fib = Fibonacci()
print(fib[10])       # 55
print(fib[2:5])      # [1, 2, 3]
print(10 in fib)     # works: Python iterates using __getitem__
for n in fib:        # works: iteration falls back to __getitem__
    pass

in and for both fall back to __getitem__-based iteration if __contains__ and __iter__ are not defined. Python calls __getitem__ with increasing integer indices starting at 0 until it gets an IndexError. This fallback exists for backward compatibility but you should define __iter__ explicitly for anything you intend to be iterable.

For mappings, define __getitem__, __setitem__, __delitem__, __len__, and __iter__. Inheriting from collections.abc.MutableMapping and implementing those five methods gives you keys(), values(), items(), get(), update(), pop(), and the rest for free through mixin implementations.

slots: Memory Layout as a Data Model Feature

__slots__ is part of the data model. Defining it on a class prevents the creation of __dict__ on instances and instead allocates a fixed set of descriptors for the named attributes.

class Point:
    __slots__ = ("x", "y")

    def __init__(self, x, y):
        self.x = x
        self.y = y

p = Point(1, 2)
p.z = 3  # AttributeError: 'Point' object has no attribute 'z'

The memory savings are significant for classes with many instances. A regular instance carries a __dict__ (a hash table) and a __weakref__ slot. A slotted instance carries only the declared attributes. For a class like Point that might be instantiated millions of times, this matters.

__slots__ interacts with inheritance in ways that require care. If a subclass does not define __slots__, instances get a __dict__ again, defeating the purpose. Every class in the hierarchy needs to define __slots__ for the optimization to hold all the way down.

The One Rule That Ties Everything Together

Dunder methods are a protocol, not a feature list. Every method participates in a documented contract. __eq__ implies __hash__ constraints. __iter__ implies __next__. __enter__ implies __exit__. __get__ participating in the descriptor protocol implies understanding when obj is None.

The mistakes come from implementing part of a protocol and ignoring the rest. Define __eq__ without thinking about __hash__ and your objects silently break in sets. Define __iter__ without StopIteration handling and your loops never end. Define __enter__ without a matching __exit__ and resources leak.

Read the protocol documentation for any dunder you implement. The contracts are specified, they are not long, and violating them produces bugs that are difficult to reproduce and harder to trace. The data model is the most reliable part of Python. It rewards the developers who actually read it.

Python Metaclasses, What's Actually Happening When You Write class

Shayan Holakouee — Wed, 29 Apr 2026 03:19:01 +0000

Every Python developer knows that everything is an object. What fewer know is that classes are objects too, and like every other object, they are instances of something. That something is a metaclass. Understanding metaclasses means understanding how Python builds classes in the first place, which turns out to explain a lot of behavior that otherwise looks like magic.

Classes Are Instances of type

Start here:

class MyClass:
    pass

print(type(MyClass))    # <class 'type'>
print(type(int))        # <class 'type'>
print(type(str))        # <class 'type'>
print(type(type))       # <class 'type'>

type is the metaclass of every class you have ever written unless you said otherwise. It is the class of classes. It is also an instance of itself, which is one of those things that is true and also slightly uncomfortable to think about for too long.

Since type is a class, you can call it to create new classes at runtime. The three-argument form of type does exactly this:

# type(name, bases, namespace)
MyClass = type("MyClass", (object,), {"x": 42, "greet": lambda self: "hello"})

obj = MyClass()
print(obj.x)        # 42
print(obj.greet())  # hello

This is not a trick or an edge case. This is literally what happens when Python processes a class statement. The class syntax is a shorthand for calling type (or a custom metaclass) with the class name, base classes, and namespace.

The Class Creation Protocol, Step by Step

When Python encounters a class block, it does not just evaluate the body and hand it to type. It follows a specific protocol.

Step 1: Determine the metaclass.

Python looks for a metaclass keyword argument in the class definition. If not found, it checks the base classes for their metaclass. If no bases specify one, it defaults to type.

class Meta(type):
    pass

class MyClass(metaclass=Meta):
    pass

print(type(MyClass))  # <class '__main__.Meta'>

Step 2: Prepare the namespace.

Python calls metaclass.__prepare__(name, bases, **kwargs). This returns the dictionary that will be used as the class namespace while the body executes. By default it returns a plain dict, but you can return an OrderedDict or a custom mapping to do things like track definition order or intercept attribute assignments.

class OrderedMeta(type):
    @classmethod
    def __prepare__(mcs, name, bases, **kwargs):
        return {}  # default, but you can return anything dict-like

Step 3: Execute the class body.

The class body runs as a function, with the namespace from step 2 as its local scope. Every name defined in the body becomes a key in that namespace. This is why you can do things like:

class MyClass:
    items = []
    for i in range(3):
        items.append(i)  # this runs at class definition time

Step 4: Call the metaclass.

Python calls metaclass(name, bases, namespace) with the populated namespace. This is where the class object is actually created. This is type.__call__, which internally calls type.__new__ to create the class object and type.__init__ to initialize it.

The full chain: Meta(name, bases, ns) → Meta.__call__ → Meta.__new__ + Meta.__init__.

Writing a Metaclass That Does Something Useful

The canonical example of a metaclass that earns its complexity is one that enforces an interface. Suppose you want an abstract base pattern without the abc module, where subclasses that forget to implement required methods fail at class definition time rather than at instantiation.

class RequiredMethodsMeta(type):
    required = []

    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)

        # skip the check for the base class itself
        if bases:
            missing = [
                method for method in mcs.required
                if method not in namespace
            ]
            if missing:
                raise TypeError(
                    f"{name} must implement: {', '.join(missing)}"
                )
        return cls


class Shape(metaclass=RequiredMethodsMeta):
    RequiredMethodsMeta.required = ["area", "perimeter"]


class Circle(Shape):
    def area(self):
        return 3.14 * self.radius ** 2

    def perimeter(self):
        return 2 * 3.14 * self.radius

# This class raises TypeError at definition time, not at use time
class Square(Shape):
    def area(self):
        return self.side ** 2
    # missing perimeter
# => TypeError: Square must implement: perimeter

The error happens when the class statement for Square is processed, before any instance is ever created. This is the key advantage of metaclass-level enforcement over runtime checks.

new vs init in a Metaclass

When you subclass type, the two methods you override are __new__ and __init__. The distinction matters.

__new__ creates and returns the class object. This is where you modify the class before it exists as a complete object. You can alter the namespace, add or remove attributes, change the bases. The class has not been fully constructed yet.

__init__ receives the already-created class object and can perform post-construction setup. The class exists at this point but has not been returned to the caller yet.

class TrackedMeta(type):
    registry = []

    def __new__(mcs, name, bases, namespace):
        # Modify namespace before class is created
        namespace["_created_by"] = "TrackedMeta"
        cls = super().__new__(mcs, name, bases, namespace)
        return cls

    def __init__(cls, name, bases, namespace):
        # Class exists, register it
        TrackedMeta.registry.append(cls)
        super().__init__(name, bases, namespace)


class Base(metaclass=TrackedMeta):
    pass

class Child(Base):
    pass

print(TrackedMeta.registry)  # [<class 'Base'>, <class 'Child'>]
print(Base._created_by)      # TrackedMeta

The practical rule: use __new__ when you need to alter what the class is. Use __init__ when you just need to register or observe it after it is created.

prepare and Definition Order

__prepare__ is underused and often the cleaner solution to problems people solve with __new__. Because it controls the namespace used during class body execution, it can observe the order in which things are defined.

A plain dict preserves insertion order in Python 3.7+, so you get definition order for free now. But __prepare__ lets you go further: you can use a custom mapping that intercepts assignments, rejects duplicates, or transforms values as they are added.

class NoDuplicatesMeta(type):
    @classmethod
    def __prepare__(mcs, name, bases, **kwargs):
        return NoDuplicatesDict()

    def __new__(mcs, name, bases, namespace):
        return super().__new__(mcs, name, bases, dict(namespace))


class NoDuplicatesDict(dict):
    def __setitem__(self, key, value):
        if key in self and not key.startswith("_"):
            raise TypeError(f"Duplicate attribute: {key}")
        super().__setitem__(key, value)


class MyClass(metaclass=NoDuplicatesMeta):
    x = 1
    y = 2
    x = 3  # raises TypeError: Duplicate attribute: x

This catches the duplicate at class body execution time, not after. The error message points at the right place.

Metaclass Conflict and the MRO Problem

If you use multiple inheritance and the base classes have different metaclasses, Python raises a TypeError unless one metaclass is a subclass of all the others.

class MetaA(type): pass
class MetaB(type): pass

class A(metaclass=MetaA): pass
class B(metaclass=MetaB): pass

class C(A, B): pass
# TypeError: metaclass conflict: the metaclass of a derived class
# must be a subclass of the metaclasses of all its bases

The fix is to create a combined metaclass that inherits from both:

class MetaC(MetaA, MetaB): pass

class C(A, B, metaclass=MetaC): pass  # works

In practice, metaclass conflicts are a strong signal that the design has gotten complicated. If you are writing three metaclasses that need to be composed, consider whether __init_subclass__ or a class decorator solves the same problem more simply.

init_subclass: The Metaclass You Did Not Have to Write

Python 3.6 added __init_subclass__, which handles the most common metaclass use case without requiring a metaclass at all. It is called on the base class whenever a subclass is created.

class Plugin:
    _registry = {}

    def __init_subclass__(cls, plugin_name=None, **kwargs):
        super().__init_subclass__(**kwargs)
        if plugin_name:
            Plugin._registry[plugin_name] = cls


class JSONPlugin(Plugin, plugin_name="json"):
    def process(self, data): ...

class XMLPlugin(Plugin, plugin_name="xml"):
    def process(self, data): ...

print(Plugin._registry)
# {'json': <class 'JSONPlugin'>, 'xml': <class 'XMLPlugin'>}

No metaclass, no __new__, no type subclass. __init_subclass__ is called automatically at class creation time with whatever keyword arguments were passed in the class definition. For registration patterns, interface enforcement, and configuration at definition time, this handles most cases cleanly.

When a Metaclass Is Actually the Right Tool

Metaclasses are the right tool in a narrow set of situations. They make sense when you need to transform or validate the class namespace before the class object exists, when you need behavior that __init_subclass__ cannot express because it does not run at all for the base class itself, and when you are building framework infrastructure that other developers will subclass without knowing the mechanics.

Django's ORM uses metaclasses. SQLAlchemy uses metaclasses. The abc module uses a metaclass. These are all framework-level concerns where the metaclass is invisible to the end user.

If you are writing application code and reaching for a metaclass, stop and ask whether a class decorator or __init_subclass__ solves the problem. They almost always do, and they are significantly easier to understand, test, and debug. A metaclass that the next developer has to spend an hour understanding is a liability. The best metaclass is usually the one you did not have to write.

The BEAM Is Not Like Other Runtimes (And That's Why Elixir Scales the Way It Does)

Shayan Holakouee — Mon, 27 Apr 2026 20:35:48 +0000

You have been writing Elixir for a while. You know GenServers. You know supervisors. You have hit the BEAM's concurrency model enough times to trust it. But there is a level below that which most Elixir developers never look at, and it explains a lot of behavior that otherwise seems like magic.

Why does spawning a hundred thousand processes not kill your system? Why is a process crash isolated but not silent? Why does garbage collection in Elixir not stop the world? The answers are all in the BEAM, and they are worth understanding properly.

The BEAM Is Not the JVM With a Different Language

This comparison comes up constantly and it misleads people. Both are virtual machines that run bytecode. That is roughly where the similarity ends.

The JVM was designed around threads. Concurrency on the JVM means OS threads, shared mutable state, locks, and a garbage collector that operates across the entire heap. The BEAM was designed around processes. Not OS processes. Not OS threads. Its own lightweight processes, each with isolated memory, independent garbage collection, and message passing as the only communication mechanism.

This is not an implementation detail. It is the fundamental design decision that every other property of the BEAM follows from.

Processes Are Cheaper Than You Think

When people hear "process" they think OS process: expensive, slow to start, heavy on memory. BEAM processes are none of those things.

A BEAM process starts with around 2KB of stack space. It grows as needed. Spawning one takes microseconds. The BEAM scheduler manages thousands of them across a small pool of OS threads, one per CPU core by default. You can run a million concurrent processes on a single machine without the OS knowing about most of them.

# This is not dangerous. It is idiomatic.
pids = Enum.map(1..100_000, fn i ->
  spawn(fn ->
    Process.sleep(:timer.seconds(10))
    IO.puts("Process #{i} done")
  end)
end)

IO.puts("Spawned #{length(pids)} processes")
# => Spawned 100000 processes

On most systems this runs without issue. Try the equivalent with OS threads in any language and you will run out of resources before you hit ten thousand. The difference is not hardware. It is the scheduler.

The Scheduler: Preemptive and Reduction-Based

The BEAM uses a preemptive scheduler, but not in the way most runtimes do it. It does not preempt based on time slices measured in milliseconds. It preempts based on reductions.

A reduction is roughly one unit of work: a function call, a pattern match, a message send. Each process gets a budget of 2000 reductions per scheduling turn. When the budget runs out, the scheduler suspends the process and runs another one. This happens regardless of what the process is doing. A tight loop does not starve other processes.

# This does not block the scheduler
defmodule Spinner do
  def spin(n) do
    spin(n + 1)  # each call consumes a reduction
  end
end

spawn(fn -> Spinner.spin(0) end)
spawn(fn -> IO.puts("I still run") end)
# => I still run

In Node.js, a synchronous tight loop blocks the event loop entirely. In the BEAM, the scheduler preempts it after 2000 reductions. Every other process continues running. The misbehaving process gets its turn again shortly, but it cannot monopolize the system.

This is why latency in Elixir systems tends to be consistent rather than spiky. No single process can hold the scheduler hostage.

Memory Isolation and Per-Process GC

Every BEAM process has its own heap. No sharing. When a process dies, its heap is reclaimed immediately. No coordination with other processes, no waiting for a global GC cycle.

This is the source of one of the BEAM's most important properties: garbage collection does not stop the world. There is no world to stop. Each process collects its own garbage independently, on its own schedule, without affecting any other process.

# Each process allocates and collects independently
spawn(fn ->
  large_list = Enum.to_list(1..1_000_000)
  # large_list is only in this process's heap
  # GC here affects nothing else
  :ok
end)

# This process is unaffected by the GC above
spawn(fn ->
  IO.puts("Running without pause")
end)

Compare this to the JVM's stop-the-world pauses, or Go's GC which, while incremental, still coordinates across the entire heap. In the BEAM, a process doing heavy allocation and collection does not introduce latency spikes in other processes. The isolation is complete.

The tradeoff is message passing. Because processes share no memory, sending data between them requires copying. Large messages are expensive not because of the send itself but because of the copy. This is why Elixir idioms tend toward small, frequent messages rather than large data transfers. The design pushes you toward the pattern that performs well.

How Message Passing Actually Works

Every process has a mailbox. Sending a message puts a copy of the message into the recipient's mailbox. The recipient reads from its mailbox using receive. This is asynchronous by default: sending does not block.

parent = self()

spawn(fn ->
  send(parent, {:result, 42})
end)

receive do
  {:result, value} -> IO.puts("Got: #{value}")
after
  5000 -> IO.puts("Timed out")
end

The receive block pattern matches against messages in the mailbox in order. If no message matches, the process blocks and waits. Unmatched messages stay in the mailbox. This is a subtle point: if you receive messages without handling all patterns, your mailbox grows unboundedly. A common bug in long-running processes is accumulating unmatched messages until memory pressure becomes a problem.

# Dangerous in a long-running process
receive do
  {:ok, value} -> handle(value)
  # no catch-all: anything else stays in mailbox forever
end

# Safer
receive do
  {:ok, value} -> handle(value)
  other -> Logger.warning("Unexpected message: #{inspect(other)}")
end

Preemption vs Cooperation: The NIF Problem

The BEAM's scheduler is preemptive for Elixir and Erlang code. It is not preemptive for NIFs.

A NIF (Native Implemented Function) is C code called directly from Elixir. When a NIF runs, the BEAM scheduler cannot preempt it. The OS thread running that NIF is blocked for the duration of the call. If your NIF takes 100ms, every process scheduled on that OS thread waits 100ms.

# If :my_nif.heavy_computation/1 takes 500ms,
# it blocks the scheduler thread for 500ms
result = :my_nif.heavy_computation(data)

This is why the BEAM documentation is so emphatic that NIFs must be fast. Anything over one millisecond should use a dirty scheduler: a separate pool of OS threads reserved for long-running native calls that runs outside the main scheduler.

# In your NIF definition (C side), mark it dirty:
# ERL_NIF_DIRTY_JOB_CPU_BOUND or ERL_NIF_DIRTY_JOB_IO_BOUND

Understanding this explains why some Elixir libraries that wrap C code (image processing, certain crypto operations, database drivers with native extensions) can introduce latency spikes that seem inconsistent with the BEAM's concurrency model. They are bypassing the model.

The Scheduler's Relationship With IO

IO in the BEAM is non-blocking at the scheduler level. When a process does a network read or a file operation, the BEAM does not block the OS thread. It suspends the process, registers the IO operation with an internal polling mechanism (built on epoll or kqueue depending on the OS), and runs other processes. When the IO completes, the process is rescheduled.

This is why Elixir handles tens of thousands of concurrent connections without a thread-per-connection model. Each connection maps to a process. Each process consumes a BEAM process worth of resources (a few kilobytes), not an OS thread. The IO multiplexing is handled by the runtime, not by the application code.

# Each of these connections is a lightweight process
# The BEAM handles the IO multiplexing
{:ok, socket} = :gen_tcp.accept(listen_socket)
pid = spawn(fn -> handle_connection(socket) end)

Phoenix handling millions of concurrent WebSocket connections is not a framework trick. It is the BEAM's process model applied to network IO with no special configuration required.

What This Means for How You Write Code

Once you understand the BEAM's process model, certain Elixir patterns stop looking like conventions and start looking like direct expressions of the runtime's capabilities.

Spawning a process per request is not wasteful, it is appropriate. Letting a process crash instead of handling every possible error is not lazy, it is correct: the process's isolated heap disappears cleanly, the supervisor restarts it in a known good state, and no other process was affected.

The things that are expensive in other runtimes (concurrency, isolation, failure recovery) are cheap in the BEAM because the runtime was built to make them cheap. The things that are expensive in the BEAM (large message copies, long-running NIFs) are expensive because they work against the model.

The more directly your code expresses the model, the more you get out of it.

Rust's Zero-Cost Abstractions, What Monomorphization Actually Does to Your Code

Shayan Holakouee — Mon, 27 Apr 2026 20:07:12 +0000

"Zero-cost abstractions" is one of Rust's core promises. You can write generic, composable, high-level code and the compiler will produce the same machine code as if you had written the low-level version by hand. Most people accept this claim and move on. It is worth actually understanding the mechanism behind it, because it has real consequences for compile times, binary size, and performance in ways that will eventually surprise you.

The Problem Abstractions Usually Create

In most languages with generics or polymorphism, the runtime pays for flexibility. Java's generics are erased at compile time and everything becomes Object, which means heap allocation and virtual dispatch. Python functions accept anything but pay for it with dynamic type lookups on every operation. Even C++ with templates avoids runtime cost, but the mechanism it uses is the same one Rust uses.

The question is: how do you write one function that works on many types, without inserting a layer of indirection at runtime?

What Monomorphization Is

Monomorphization means the compiler generates a separate, concrete copy of a generic function for each type it is actually called with. You write one function. The compiler writes many.

fn largest<T: PartialOrd>(a: T, b: T) -> T {
    if a > b { a } else { b }
}

fn main() {
    largest(1i32, 2i32);
    largest(1.0f64, 2.0f64);
}

After monomorphization, the compiler has effectively produced this:

fn largest_i32(a: i32, b: i32) -> i32 {
    if a > b { a } else { b }
}

fn largest_f64(a: f64, b: f64) -> f64 {
    if a > b { a } else { b }
}

Each version is fully concrete. The CPU never sees a generic type. There is no branching on type, no vtable lookup, no boxing. The generated assembly for largest_i32 is exactly what you would get if you had written it for i32 from the start.

You can verify this yourself. Run cargo build --release and inspect the output with cargo-show-asm or objdump. Each monomorphized version shows up as a distinct symbol in the binary.

How Traits Fit Into This

Traits are the constraint mechanism. When you write T: PartialOrd, you are telling the compiler what operations are available on T. At the call site, the compiler knows the concrete type, so it knows the exact implementation of PartialOrd to use. It inlines it directly.

This is what makes iterator chains fast:

let sum: i32 = (0..1000)
    .filter(|x| x % 2 == 0)
    .map(|x| x * x)
    .sum();

Each of filter, map, and sum is generic over the iterator type. After monomorphization, the entire chain collapses into a single loop with no intermediate allocation and no function call overhead. The compiler sees through every layer. It knows the concrete type at each step because you started with Range<i32>, and each adapter wraps the previous one in a new concrete type.

Compare this to how you would write the equivalent in a language without monomorphization. You would either materialize an intermediate collection at each step, or you would use some form of dynamic dispatch with its associated overhead.

The Actual Cost: Code Bloat and Compile Time

Zero-cost at runtime does not mean zero-cost overall. You pay elsewhere.

Binary size. Every unique combination of generic function and concrete type produces a new copy in the binary. If you call Vec::new::<String>(), Vec::new::<i32>(), and Vec::new::<MyStruct>(), you get three copies of the Vec::new implementation. In a large codebase with many generic types, this adds up. It is a real concern for embedded targets and WebAssembly where binary size matters.

Compile time. Monomorphization happens during code generation, which is one of the most expensive phases of Rust compilation. Every unique instantiation of a generic function is work the compiler has to do. This is a significant contributor to Rust's notoriously slow compile times. When you see a crate that takes thirty seconds to compile, a large part of that is the code generation phase expanding generics.

Instruction cache pressure. Multiple copies of similar functions means more machine code in memory. If those functions are all hot paths called in rapid succession, you can end up with instruction cache misses that would not exist with a single shared implementation.

Static Dispatch vs Dynamic Dispatch

Monomorphization is static dispatch. The call target is resolved at compile time. Rust also supports dynamic dispatch through trait objects, and the tradeoff is exactly what you would expect.

// static dispatch: monomorphized, zero runtime overhead
fn process_static<T: Draw>(item: T) {
    item.draw();
}

// dynamic dispatch: single function, vtable lookup at runtime
fn process_dynamic(item: &dyn Draw) {
    item.draw();
}

With &dyn Draw, the compiler generates one function. At runtime, a vtable pointer accompanies the reference, and the call goes through an indirect jump. This costs roughly one extra memory lookup per call. In exchange, you get a single copy of process_dynamic in the binary, and you can put different concrete types into the same collection.

The practical rule: use static dispatch (generics) for performance-critical code where all types are known at compile time. Use dynamic dispatch (dyn Trait) when you need heterogeneous collections, when the concrete type is only known at runtime, or when binary size and compile time matter more than the last bit of call overhead.

Where Monomorphization Gets Interesting: Trait Objects in Generic Contexts

A subtlety that catches people: trait objects themselves can be used as concrete types in generic contexts, which then get monomorphized.

fn process<T: Draw>(items: Vec<T>) { ... }

// This is one monomorphization
process::<Box<dyn Draw>>(vec![...]);

Here T is Box<dyn Draw>. The function is monomorphized for that concrete type. Inside the function, calls through T still go through the vtable because T is a trait object. You have static dispatch to the function, and dynamic dispatch inside it. Understanding this layering matters when you are trying to reason about where the overhead actually lives.

The impl Trait Shorthand

impl Trait in argument position is syntactic sugar for a generic parameter:

// these are equivalent
fn process(item: impl Draw) { ... }
fn process<T: Draw>(item: T) { ... }

Both are monomorphized. Each unique concrete type passed to process produces a new instantiation. impl Trait is not dynamic dispatch despite looking less generic at first glance.

impl Trait in return position is different:

fn make_drawable() -> impl Draw {
    Circle::new()
}

This tells the caller "you will get something that implements Draw, but I am not telling you the concrete type." The concrete type is fixed at compile time and the function is not generic over it. The caller cannot name the type but the compiler knows it, so there is no dynamic dispatch. It is a way to hide implementation details while keeping static dispatch.

Inspecting the Output

The best way to build intuition for what monomorphization produces is to look at actual assembly. The cargo-show-asm tool makes this straightforward:

cargo install cargo-show-asm
cargo asm --release your_crate::your_function

Run it on a generic function called with two different types and you will see two separate assembly listings with distinct symbol names. Run it on the iterator chain example and you will see a tight loop with no function calls. The gap between what you wrote and what the CPU executes is much smaller in Rust than in most languages, and this tool makes that concrete rather than theoretical.

The Mental Model to Carry Forward

When you write generic Rust code, think of yourself as writing a template that the compiler will stamp out once per concrete type. Each stamp is as fast as hand-written code for that type. The tradeoff you are making is compilation time and binary size in exchange for runtime performance and the ability to write the logic once.

This is why Rust can match C performance while supporting generics, iterators, and trait-based abstractions. The abstraction cost is paid at compile time, not runtime. Zero-cost does not mean free. It means the cost is moved to a place where it hurts less.

Python's Descriptor Protocol, The Feature Behind Everything You Use Daily

Shayan Holakouee — Mon, 27 Apr 2026 20:02:35 +0000

You use descriptors every day. Every time you write @property, every time you call a method, every time you use @staticmethod or @classmethod, you are using the descriptor protocol. Most Python developers have no idea it exists as a unified mechanism. Once you see it, you cannot unsee it.

What a Descriptor Is

A descriptor is any object that defines at least one of three methods: __get__, __set__, or __delete__. That is it. When an attribute lookup finds one of these objects sitting in a class's __dict__, Python hands control over to the descriptor instead of returning the object directly.

The minimal descriptor looks like this:

class MyDescriptor:
    def __get__(self, obj, objtype=None):
        print(f"__get__ called: obj={obj}, objtype={objtype}")
        return 42

class MyClass:
    attr = MyDescriptor()

instance = MyClass()
instance.attr   # prints: __get__ called: obj=<MyClass ...>, objtype=<class 'MyClass'>
MyClass.attr    # prints: __get__ called: obj=None, objtype=<class 'MyClass'>

When obj is None, the descriptor is being accessed from the class itself, not an instance. This distinction matters and you will use it constantly when writing real descriptors.

Data Descriptors vs Non-Data Descriptors

This split is the most important thing to understand about how descriptors interact with instance __dict__.

A data descriptor defines both __get__ and __set__ (or __delete__). It takes priority over the instance's own __dict__. You cannot shadow it by doing instance.attr = something because __set__ intercepts the assignment.

A non-data descriptor defines only __get__. The instance's __dict__ takes priority over it. If you assign to instance.attr, Python just stores the value in the instance dictionary and the descriptor is no longer called for that instance.

The lookup order for instance.attr is:

Data descriptors from the class (and its MRO)
Instance __dict__
Non-data descriptors and other class attributes

This ordering is not arbitrary. It is what makes @property work correctly. A property is a data descriptor because it defines __set__ (even if you only define a getter, the setter raises AttributeError), so it cannot be overridden by instance assignment. Functions, on the other hand, are non-data descriptors, which is why you can shadow a method by assigning to an instance attribute of the same name.

How Functions Become Methods

This is the part that clicks everything together. In Python, a plain function is a non-data descriptor. It only defines __get__. When you access a function through an instance, __get__ is called, and it returns a bound method object that already has the instance baked in as the first argument.

def greet(self):
    return f"Hello from {self}"

class Person:
    greet = greet

p = Person()

# These are equivalent
p.greet()
Person.greet(p)

# What actually happens
bound_method = Person.__dict__['greet'].__get__(p, Person)
bound_method()

function.__get__(instance, owner) returns a MethodType object that wraps the function and the instance together. There is no magic beyond this. The entire method binding mechanism in Python is the descriptor protocol applied to plain functions.

You can verify this:

print(type(greet))           # <class 'function'>
print(greet.__get__)         # <method-wrapper '__get__' of function object at ...>
print(type(p.greet))         # <class 'method'>

Writing a Real Descriptor: Typed Attributes

Here is where descriptors become practically useful. Suppose you want class attributes that enforce a type at assignment time without cluttering __init__ with validation logic.

class Typed:
    def __set_name__(self, owner, name):
        self.name = name
        self.private_name = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value):
        if not isinstance(value, self.expected_type):
            raise TypeError(
                f"{self.name} must be {self.expected_type.__name__}, "
                f"got {type(value).__name__}"
            )
        setattr(obj, self.private_name, value)

class Integer(Typed):
    expected_type = int

class String(Typed):
    expected_type = str

class Person:
    name = String()
    age = Integer()

    def __init__(self, name, age):
        self.name = name
        self.age = age

p = Person("Alice", 30)   # works
p.age = "thirty"          # raises TypeError: age must be int, got str

Notice __set_name__. This is a hook Python calls automatically when the class is created, passing the descriptor the owner class and the attribute name it was assigned to. Without it, the descriptor would not know its own name and you would have to pass it manually. It was added in Python 3.6 and makes descriptors significantly cleaner to write.

classmethod and staticmethod Are Just Descriptors

Once you understand the protocol, classmethod and staticmethod stop being mysterious decorators and become straightforward descriptor implementations.

staticmethod wraps a function and its __get__ simply returns the function unchanged, with no binding:

class StaticMethod:
    def __init__(self, f):
        self.f = f

    def __get__(self, obj, objtype=None):
        return self.f  # no binding, just return the raw function

classmethod wraps a function and its __get__ binds the class instead of the instance:

class ClassMethod:
    def __init__(self, f):
        self.f = f

    def __get__(self, obj, objtype=None):
        if objtype is None:
            objtype = type(obj)
        return self.f.__get__(objtype)  # bind the class, not the instance

These are simplified but functionally accurate. The actual CPython implementations are in C for performance, but the logic is identical. You could replace @classmethod with your own descriptor and everything would work the same way.

The Caching Descriptor Pattern

One underused pattern is using descriptors to compute expensive values once and then cache them on the instance, bypassing the descriptor entirely on subsequent access.

class cached_property:
    def __init__(self, func):
        self.func = func
        self.attrname = None

    def __set_name__(self, owner, name):
        self.attrname = name

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        val = self.func(obj)
        obj.__dict__[self.attrname] = val  # store directly on instance
        return val

This works because cached_property is a non-data descriptor (it only defines __get__). On first access, it calls the function and stores the result in the instance __dict__ under the same name. On second access, the instance __dict__ lookup wins and the descriptor is never called again.

Python ships functools.cached_property since 3.8 using exactly this technique. It is elegant because the caching mechanism falls directly out of the descriptor lookup order rather than requiring any explicit flag or sentinel.

When to Actually Use This

Descriptors are the right tool in specific situations. If you find yourself writing the same property logic across multiple attributes or classes (validation, unit conversion, access logging), a descriptor lets you define that logic once and attach it declaratively.

ORMs lean on this heavily. Django's Field classes and SQLAlchemy's Column objects are descriptors. When you write user.name, the ORM intercepts it through __get__ and can return a SQL expression rather than a plain value, depending on context.

Frameworks that do attribute access tracking for reactive programming (like Traitlets in Jupyter or any observable pattern) use descriptors to intercept reads and writes and fire callbacks.

If you are not building framework-level infrastructure, plain properties cover most cases. But understanding descriptors tells you exactly what @property is doing and why it behaves the way it does, which makes the edge cases much easier to reason about.

DEV Community: Shayan Holakouee

A Systematic Approach to AI Agent Engineering

The Mental Model Shift

Architecture Patterns That Actually Scale

1. The Supervisor Pattern (Multi-Agent Orchestration)

2. Pipelines vs. Agents: Know Which One You Need

Advanced Python Patterns for AI Systems

Structured Outputs with Pydantic + Runtime Validation

Memory Architecture: Don't Reinvent the Wheel

Tool Use: The Sharp Edge

Idempotent Tool Design

Tool Registry with Schema Validation

Observability Is Not Optional

Failure Modes That Will Bite You

1. The Infinite Loop

2. Context Window Overflow

3. Prompt Injection in Tool Outputs

Putting It Together: A Production Agent Loop

Framework Landscape (2025 State of Play)

Closing Thoughts

Building a net/http Middleware Chain From Scratch in Go

The Foundation: What Is http.Handler?

The Middleware Signature

Building the Chain

Middleware 1: Request Logging

Middleware 2: Panic Recovery

Middleware 3: Authentication

Middleware 4: Request Timeout

Putting It All Together

When to Reach for a Framework

What You Built

Your Python Pre-commit Hook Took 18 Seconds. Ruff Does It in 0.3.

What Is Ruff?

Installing Ruff

The Two Commands You'll Use Every Day

What Does It Actually Catch?

Replacing isort

Configuration in pyproject.toml

Rule Categories Worth Knowing

Ignoring Rules Inline

Editor Integration

VS Code

Neovim

Pre-commit Integration

Migrating From the Old Stack

One Honest Caveat

The Bottom Line

Stop Juggling 5 Tools , Python's uv Does It All (And It's Blazing Fast)

What Even Is uv?

Installing uv

The Speed Claim Is It Real?

Starting a New Project

Managing Python Versions

Adding and Removing Dependencies

Running Commands Without Activating the Venv

Replacing pipx Running One-off Tools

Inline Script Dependencies

Migrating an Existing Project

The Mental Model Shift

When to Still Be Careful

The Bottom Line

FastAPI Patterns You Should Be Using in Production

Introduction

1. Dependency Injection — The Right Way

Layered Dependencies

Class-Based Dependencies

Dependency Overrides in Tests

2. Lifespan Events — Replacing @app.on_event

3. Custom Middleware — Beyond CORSMiddleware

Request ID Middleware

Timing Middleware

4. Background Tasks vs. Celery — When to Use What

FastAPI's Built-in Background Tasks

When to Use Celery (or ARQ/Dramatiq)

5. Advanced Pydantic v2 Patterns

Computed Fields

Model Validators

Custom Serialization

6. Structured Error Handling

7. Streaming Responses

What Even Is `uv`?

Replacing `pipx` Running One-off Tools

2. Lifespan Events — Replacing `@app.on_event`

3. Custom Middleware — Beyond `CORSMiddleware`