Python

Python — Advanced Patterns

Language-level patterns for writing correct, readable, performant Python.

When to Activate

• Choosing between dataclasses, TypedDict, NamedTuple, or Pydantic models
• Designing type annotations with generics, Protocols, or TypeVar
• Writing or debugging async code (async/await, event loops, asyncio)
• Building decorators, context managers, or generators
• Using itertools, functools, or comprehensions effectively
• Applying structural pattern matching (match/case)
• Performance micro-optimisations (__slots__, lru_cache, generators vs lists)

Type Hints

Built-in generics (Python 3.10+)

# Use built-in types directly — no need to import from typing
def process(items: list[str]) -> dict[str, int]: ...
def fetch(ids: set[int]) -> tuple[str, ...]: ...
def map_fn(data: dict[str, list[int]]) -> None: ...

# Union with | (3.10+)
def parse(value: str | int | None) -> str: ...

# TypeAlias
type UserId = str           # Python 3.12+
UserId = NewType("UserId", str)  # Python 3.10+

TypeVar and Generics

from typing import TypeVar, Generic

T = TypeVar("T")
K = TypeVar("K")
V = TypeVar("V")

def first(items: list[T]) -> T | None:
    return items[0] if items else None

class Repository(Generic[T]):
    async def get(self, id: str) -> T | None: ...
    async def save(self, entity: T) -> T: ...

class UserRepository(Repository[User]): ...  # T = User

Protocol — structural subtyping (duck typing with types)

from typing import Protocol, runtime_checkable

@runtime_checkable
class Closeable(Protocol):
    def close(self) -> None: ...

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict: ...
    @classmethod
    def from_dict(cls, data: dict) -> "Serializable": ...

# Any class with these methods satisfies the protocol — no inheritance needed
def cleanup(resource: Closeable) -> None:
    resource.close()

Annotated — attach metadata to types

from typing import Annotated
from pydantic import Field

# Reusable constrained types
PositiveInt = Annotated[int, Field(gt=0)]
EmailStr = Annotated[str, Field(pattern=r".+@.+")]
UserId = Annotated[str, Field(min_length=36, max_length=36)]

class User(BaseModel):
    id: UserId
    age: PositiveInt
    email: EmailStr

Literal and TypeGuard

from typing import Literal, TypeGuard

Status = Literal["active", "inactive", "banned"]

def is_active(status: Status) -> TypeGuard[Literal["active"]]:
    return status == "active"

# TypedDict for dict shapes
from typing import TypedDict

class UserDict(TypedDict):
    id: str
    name: str
    email: str

class PartialUserDict(TypedDict, total=False):
    name: str
    email: str

Dataclasses vs Pydantic vs TypedDict

from dataclasses import dataclass, field

@dataclass
class Point:
    x: float
    y: float
    tags: list[str] = field(default_factory=list)  # mutable default must use field()

@dataclass(frozen=True)   # immutable, hashable
class Color:
    r: int; g: int; b: int

@dataclass(slots=True)    # __slots__ automatically (Python 3.10+)
class FastModel:
    name: str
    value: int

Async / Await

Pitfalls

# BAD: sync sleep blocks the entire event loop
async def handler():
    time.sleep(1)           # freezes all other coroutines

# GOOD: async sleep yields control
async def handler():
    await asyncio.sleep(1)  # other coroutines run while waiting

# BAD: sync I/O in async code (blocks event loop)
async def read_file():
    return open("file.txt").read()  # blocking

# GOOD: use aiofiles or run in executor
import aiofiles
async def read_file():
    async with aiofiles.open("file.txt") as f:
        return await f.read()

# Run blocking code in thread pool (for libraries you can't change)
import asyncio
result = await asyncio.get_event_loop().run_in_executor(None, blocking_function, arg)

Concurrency patterns

import asyncio

# Run independent coroutines concurrently
results = await asyncio.gather(
    fetch_user(id),
    fetch_orders(id),
    fetch_profile(id),
)
user, orders, profile = results

# gather with error handling
results = await asyncio.gather(fetch_a(), fetch_b(), return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        handle_error(r)

# TaskGroup (Python 3.11+) — cancels all tasks if one fails
async with asyncio.TaskGroup() as tg:
    task_a = tg.create_task(fetch_a())
    task_b = tg.create_task(fetch_b())
# both results available after the block

# Timeout
try:
    result = await asyncio.wait_for(slow_operation(), timeout=5.0)
except asyncio.TimeoutError:
    handle_timeout()

# Semaphore — limit concurrency (e.g. max 10 concurrent HTTP requests)
sem = asyncio.Semaphore(10)
async def rate_limited_fetch(url):
    async with sem:
        return await httpx.get(url)

Async generators and context managers

# Async generator
async def paginate(url: str):
    page = 1
    while True:
        data = await fetch(f"{url}?page={page}")
        if not data:
            break
        yield data
        page += 1

async for batch in paginate("/api/items"):
    process(batch)

# Async context manager
class AsyncDB:
    async def __aenter__(self):
        self.conn = await connect()
        return self.conn

    async def __aexit__(self, *args):
        await self.conn.close()

async with AsyncDB() as conn:
    await conn.execute("SELECT 1")

Context Managers

from contextlib import contextmanager, asynccontextmanager, suppress

# Synchronous
@contextmanager
def timer(label: str):
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.3f}s")

with timer("query"):
    result = db.execute(query)

# Async
@asynccontextmanager
async def db_transaction(session):
    async with session.begin():
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

# Suppress specific exceptions
with suppress(FileNotFoundError):
    os.remove("tmp.txt")

# ExitStack — combine multiple context managers dynamically
from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(f)) for f in file_list]
    process(files)

Decorators

from functools import wraps
import time

# Basic decorator with functools.wraps (preserves __name__, __doc__)
def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await fn(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    await asyncio.sleep(delay * (attempt + 1))
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5)
async def fetch_data(url: str) -> dict: ...

# Class-based decorator (useful when the decorator needs state)
class RateLimit:
    def __init__(self, calls: int, period: float):
        self.calls = calls
        self.period = period
        self.timestamps: list[float] = []

    def __call__(self, fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            now = time.time()
            self.timestamps = [t for t in self.timestamps if now - t < self.period]
            if len(self.timestamps) >= self.calls:
                raise Exception("Rate limit exceeded")
            self.timestamps.append(now)
            return await fn(*args, **kwargs)
        return wrapper

Generators and itertools

import itertools

# Generator — lazy, memory-efficient
def read_chunks(path: str, size: int = 4096):
    with open(path, "rb") as f:
        while chunk := f.read(size):
            yield chunk

# Generator expression
total = sum(x ** 2 for x in range(1_000_000))  # no list in memory

# itertools
list(itertools.islice(range(100), 10))          # first 10
list(itertools.chain([1,2], [3,4], [5,6]))      # flatten iterables
list(itertools.batched([1..9], 3))              # [[1,2,3],[4,5,6],[7,8,9]] (3.12+)
list(itertools.groupby(sorted_items, key=lambda x: x.category))
list(itertools.takewhile(lambda x: x < 5, items))
list(itertools.dropwhile(lambda x: x < 5, items))
list(itertools.pairwise([1,2,3,4]))             # [(1,2),(2,3),(3,4)] (3.10+)

# functools
from functools import lru_cache, cached_property, reduce, partial

@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    return n if n < 2 else fibonacci(n-1) + fibonacci(n-2)

class Config:
    @cached_property        # computed once, cached on instance
    def parsed_rules(self) -> list[Rule]:
        return parse_rules(self.raw)

double = partial(operator.mul, 2)   # partial application

Structural Pattern Matching (Python 3.10+)

def handle_event(event: dict):
    match event:
        case {"type": "user_created", "data": {"id": user_id, "email": email}}:
            create_user(user_id, email)

        case {"type": "order_placed", "data": {"total": total}} if total > 1000:
            flag_high_value_order(event)

        case {"type": str(t)} if t.startswith("payment_"):
            handle_payment(event)

        case _:
            log_unknown(event)

# Match on types
def process(value):
    match value:
        case int(n) if n > 0:   return f"positive int: {n}"
        case str(s):            return f"string: {s}"
        case [*items]:          return f"list of {len(items)}"
        case {"key": v}:        return f"dict with key: {v}"
        case None:              return "nothing"

__slots__

Reduces memory per instance by ~40-60% for classes with many instances. Prevents arbitrary attribute assignment.

class Point:
    __slots__ = ("x", "y")   # no __dict__, no __weakref__ by default

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

# With inheritance: each class only declares its own new slots
class Point3D(Point):
    __slots__ = ("z",)        # inherits x, y slots from Point

Use slots=True on dataclasses: @dataclass(slots=True).

Common Gotchas

# Mutable default argument — shared across all calls
def bad(items=[]):      items.append(1)  # BAD — list is shared
def good(items=None):   items = items or []  # GOOD

# Late binding in closures
fns = [lambda x, i=i: x + i for i in range(3)]  # i=i captures current value

# is vs ==
a = 256; b = 256; a is b   # True  — small ints are cached
a = 257; b = 257; a is b   # False — large ints are not
# Always use == for value equality; is only for None/True/False/singletons

# Walrus operator := (Python 3.8+)
while chunk := file.read(8192):
    process(chunk)

if m := re.search(pattern, text):
    print(m.group(0))

# Exception chaining
try:
    result = parse(data)
except ValueError as e:
    raise ServiceError("Parse failed") from e  # preserves original traceback

# f-string debugging (Python 3.8+)
x = 42
print(f"{x=}")  # prints: x=42

Performance Tips

Red Flags

• Mutable default arguments (def f(items=[])) — the default list is created once at definition time and shared across all calls; use None as the default and initialize inside the function body
• Bare except: or except Exception: — catching all exceptions hides bugs and swallows KeyboardInterrupt; catch the narrowest specific exception type you actually expect and handle
• asyncio.run() inside an already-running event loop — calling asyncio.run() from within an async context (FastAPI, Jupyter) raises RuntimeError; use await directly or loop.run_until_complete()
• Threads for CPU-bound work — Python's GIL prevents true thread parallelism for CPU tasks; use multiprocessing or ProcessPoolExecutor for CPU-bound parallelism
• from module import * in __init__.py — star imports pollute the namespace and make name origins untraceable; always import explicitly
• @dataclass fields with mutable defaults — field: list = [] shares the same list object across all instances; use field(default_factory=list) for mutable defaults
• is to compare values — x is 1 or x is "hello" relies on CPython interning that is not guaranteed across Python versions; use == for value comparison and is only for None, True, False

Checklist

• Type hints on all public functions and method signatures
• Protocol used instead of ABC when only structural compatibility matters
• Mutable defaults use field(default_factory=...) in dataclasses
• Async code uses asyncio.sleep, aiofiles, or run_in_executor — never time.sleep
• Independent async calls use asyncio.gather or TaskGroup
• Decorators use @functools.wraps to preserve function metadata
• @lru_cache on pure, frequently-called functions
• __slots__ on dataclasses or classes with many instances
• suppress / ExitStack from contextlib instead of try/finally boilerplate