---
name: python-patterns
description: "Modern Python 3.12+: type hints, dataclasses, Pydantic v2, async/await, context managers, generators, pattern matching. Use when writing or reviewing modern Python: typing, dataclasses, Pydantic, async, generators."
tags: [python, type-hints, pydantic, asyncio, dataclasses, backend]
version: "2025.1"
---

# Modern Python Patterns (3.12+)

## Core Principles

Python 3.12+ brings improved performance, better error messages, type parameter syntax (PEP 695),
and f-string improvements. Always use type hints, prefer dataclasses or Pydantic for data
modeling, and use async/await for I/O-bound operations.

## Type Hints

```python
from typing import TypeAlias, TypeVar, Protocol
from collections.abc import Sequence, Mapping, Callable, Iterator

# Basic annotations
def greet(name: str, times: int = 1) -> str:
    return f"Hello, {name}! " * times

# Union types (Python 3.10+ syntax)
def process(value: str | int | None) -> str:
    if value is None:
        return "empty"
    return str(value)

# Generic functions (Python 3.12+ syntax  -  PEP 695)
type NumberList[T: (int, float)] = list[T]

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

# Type aliases
type UserId = str
type Headers = Mapping[str, str]
type Handler = Callable[[str, int], bool]

# Protocol (structural subtyping)
class Printable(Protocol):
    def to_string(self) -> str: ...

def display(item: Printable) -> None:
    print(item.to_string())

# TypeGuard
from typing import TypeGuard

def is_str_list(val: list[object]) -> TypeGuard[list[str]]:
    return all(isinstance(x, str) for x in val)

# TypedDict
from typing import TypedDict, NotRequired

class UserDict(TypedDict):
    id: str
    name: str
    email: str
    age: NotRequired[int]
```

## Dataclasses

```python
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum

class Role(StrEnum):
    ADMIN = "admin"
    USER = "user"
    VIEWER = "viewer"

@dataclass(frozen=True, slots=True)
class User:
    id: str
    name: str
    email: str
    role: Role = Role.USER
    tags: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)

    @property
    def display_name(self) -> str:
        return f"{self.name} ({self.role.value})"

    def has_tag(self, tag: str) -> bool:
        return tag in self.tags

# Usage
user = User(id="1", name="Alice", email="alice@example.com")
print(user.display_name)  # "Alice (user)"

# frozen=True makes it hashable and immutable
user_set: set[User] = {user}

# slots=True improves memory and attribute access speed
```

## Pydantic v2

```python
from pydantic import BaseModel, Field, field_validator, model_validator, ConfigDict
from datetime import datetime

class CreateUserRequest(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, frozen=True)

    name: str = Field(min_length=1, max_length=100)
    email: str = Field(pattern=r'^[\w.-]+@[\w.-]+\.\w+$')
    age: int = Field(ge=0, le=150)
    role: Role = Role.USER
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator('email')
    @classmethod
    def normalize_email(cls, v: str) -> str:
        return v.lower()

    @model_validator(mode='after')
    def validate_admin_age(self) -> 'CreateUserRequest':
        if self.role == Role.ADMIN and self.age < 18:
            raise ValueError("Admin must be 18 or older")
        return self

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    role: Role
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

# Validation
try:
    user = CreateUserRequest(name="Alice", email="ALICE@example.com", age=25)
    print(user.email)  # "alice@example.com" (normalized)
except ValidationError as e:
    print(e.errors())

# Serialization
user_dict = user.model_dump()
user_json = user.model_dump_json()

# From ORM object
db_user = db.query(UserModel).first()
response = UserResponse.model_validate(db_user)
```

## Async/Await with asyncio

```python
import asyncio
from collections.abc import AsyncIterator
import httpx

# Async function
async def fetch_user(user_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

# Parallel execution
async def fetch_all_users(user_ids: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [
            client.get(f"https://api.example.com/users/{uid}")
            for uid in user_ids
        ]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return [
            r.json() for r in responses
            if not isinstance(r, Exception) and r.status_code == 200
        ]

# Async context manager
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_db_session() -> AsyncIterator[AsyncSession]:
    session = AsyncSession(engine)
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

# Async generator
async def stream_events(url: str) -> AsyncIterator[dict]:
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    yield json.loads(line[6:])

# Semaphore for rate limiting
async def fetch_with_limit(urls: list[str], max_concurrent: int = 10) -> list[str]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url: str) -> str:
        async with semaphore:
            async with httpx.AsyncClient() as client:
                resp = await client.get(url)
                return resp.text

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

## Context Managers

```python
from contextlib import contextmanager, suppress
from typing import Generator
import time

@contextmanager
def timer(label: str) -> Generator[None, None, None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

# Usage
with timer("database query"):
    results = db.execute(query)

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

# Class-based context manager
class ManagedResource:
    def __init__(self, name: str) -> None:
        self.name = name

    def __enter__(self) -> 'ManagedResource':
        print(f"Acquiring {self.name}")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        print(f"Releasing {self.name}")
        return False  # Do not suppress exceptions
```

## Generators

```python
from collections.abc import Generator, Iterator

# Generator function
def chunked[T](items: list[T], size: int) -> Generator[list[T], None, None]:
    for i in range(0, len(items), size):
        yield items[i:i + size]

# Usage
for chunk in chunked(large_list, batch_size=100):
    process_batch(chunk)

# Generator expression (lazy evaluation)
total = sum(item.price for item in orders if item.status == "completed")

# Infinite generator
def counter(start: int = 0) -> Iterator[int]:
    n = start
    while True:
        yield n
        n += 1

# Pipeline of generators
def read_lines(path: str) -> Iterator[str]:
    with open(path) as f:
        yield from f

def parse_csv(lines: Iterator[str]) -> Iterator[dict[str, str]]:
    headers = next(lines).strip().split(",")
    for line in lines:
        values = line.strip().split(",")
        yield dict(zip(headers, values))

def filter_active(records: Iterator[dict[str, str]]) -> Iterator[dict[str, str]]:
    for record in records:
        if record.get("status") == "active":
            yield record

# Compose the pipeline (lazy  -  processes one record at a time)
active_users = filter_active(parse_csv(read_lines("users.csv")))
```

## Pattern Matching (match/case)

```python
from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

def describe_point(point: Point) -> str:
    match point:
        case Point(x=0, y=0):
            return "origin"
        case Point(x=0, y=y):
            return f"on y-axis at y={y}"
        case Point(x=x, y=0):
            return f"on x-axis at x={x}"
        case Point(x=x, y=y) if x == y:
            return f"on diagonal at {x}"
        case _:
            return f"at ({point.x}, {point.y})"

# Matching on type and structure
def handle_response(response: dict) -> str:
    match response:
        case {"status": 200, "data": data}:
            return f"Success: {data}"
        case {"status": 404}:
            return "Not found"
        case {"status": status} if 500 <= status < 600:
            return f"Server error: {status}"
        case {"error": str(message)}:
            return f"Error: {message}"
        case _:
            return "Unknown response"

# Matching sequences
def process_command(command: list[str]) -> None:
    match command:
        case ["quit" | "exit"]:
            sys.exit(0)
        case ["go", direction]:
            move(direction)
        case ["pick", "up", item]:
            pick_up(item)
        case ["drop", *items] if items:
            for item in items:
                drop(item)
        case _:
            print(f"Unknown command: {' '.join(command)}")
```

## Do's

- Use type hints on all function signatures and class attributes
- Use `dataclass(frozen=True, slots=True)` for simple value objects
- Use Pydantic `BaseModel` for input validation and serialization
- Use `async/await` for I/O-bound operations (network, file, database)
- Use generators for lazy processing of large datasets
- Use `StrEnum` instead of plain strings for known constant sets
- Use `match/case` for complex conditional logic with structured data
- Use `pathlib.Path` instead of `os.path` for file operations

## Don'ts

- Do not use `Any` type annotation unless absolutely necessary
- Do not use bare `except:`  -  always specify exception types
- Do not use mutable default arguments (`def f(items=[])`)
- Do not use `global` variables for state management
- Do not use `asyncio.run()` inside an already-running event loop
- Do not catch `Exception` when you mean a specific error type
- Do not use `type: ignore` without explaining why
- Do not mix sync and async code without proper bridges (`asyncio.to_thread`)

## Troubleshooting

| Problem | Cause | Solution |
|---------|-------|----------|
| "RuntimeError: event loop already running" | Calling `asyncio.run()` inside async context | Use `await` directly or `asyncio.create_task()` |
| Pydantic validation error not helpful | Missing field validators | Add `@field_validator` with clear error messages |
| Generator exhausted | Iterating generator twice | Convert to list if needed multiple times, or create new generator |
| Type checker error with Protocol | Missing method or wrong signature | Match Protocol method signatures exactly |
| Slow async code | Sequential awaits instead of parallel | Use `asyncio.gather()` or `asyncio.TaskGroup` for concurrency |
| "TypeError: unhashable type" | Using mutable object as dict key | Use `frozen=True` dataclass or tuple instead |