Python

Pythonic patterns, modern tooling, and best practices for maintainable Python systems.

Modern Python (3.11+)

Type Hints (Mandatory for New Code)

from __future__ import annotations
from typing import Protocol, TypeVar, Generic, Optional
from dataclasses import dataclass
from datetime import datetime

T = TypeVar('T')

@dataclass(frozen=True, slots=True)
class UserId:
    value: str

@dataclass(frozen=True, slots=True)
class User:
    id: UserId
    email: str
    created_at: datetime

class UserRepository(Protocol):
    async def find_by_id(self, id: UserId) -> Optional[User]: ...
    async def save(self, user: User) -> None: ...

# Generic Result type
@dataclass(frozen=True)
class Result(Generic[T]):
    ok: bool
    value: Optional[T] = None
    error: Optional[Exception] = None

    @classmethod
    def ok(cls, value: T) -> Result[T]:
        return cls(ok=True, value=value)

    @classmethod
    def err(cls, error: Exception) -> Result[T]:
        return cls(ok=False, error=error)

Structural Pattern Matching (3.10+)

match event:
    case UserCreated(user_id, email):
        await send_welcome_email(email)
    case OrderPlaced(order_id, items) if len(items) > 10:
        await flag_large_order(order_id)
    case OrderPlaced(order_id, items):
        await process_order(order_id, items)
    case _:
        logger.warning("Unknown event", event_type=type(event).__name__)

Exception Groups (3.11+)

try:
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process(item))
except* ValidationError as eg:
    for exc in eg.exceptions:
        logger.error("Validation failed", exc_info=exc)
except* TransientError as eg:
    await retry_failed(eg.exceptions)

Tooling Stack (2024+)

pyproject.toml (Single Config File)

[project]
name = "my-service"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.109",
    "pydantic>=2.6",
    "sqlalchemy>=2.0",
    "asyncpg>=0.29",
    "pydantic-settings>=2.2",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "pytest-cov>=4.1",
    "ruff>=0.3",
    "mypy>=1.8",
    "pre-commit>=3.6",
]

[tool.ruff]
target-version = "py311"
line-length = 100
select = ["E", "F", "I", "UP", "B", "C4", "PTH", "TRY", "EM", "PIE", "T20"]
ignore = ["E501"]  # line length handled by formatter

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

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_ignores = true
disallow_untyped_defs = true
no_implicit_optional = true

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
addopts = "-v --cov=src --cov-report=term-missing"

Project Structure (Clean Architecture)

src/
├── domain/
│   ├── entities/
│   ├── value_objects/
│   ├── services/
│   ├── events/
│   └── repositories/      # Protocols (interfaces)
├── application/
│   ├── commands/
│   ├── queries/
│   ├── dto/
│   └── ports/             # Outbound port interfaces
├── infrastructure/
│   ├── persistence/       # SQLAlchemy, asyncpg impls
│   ├── http/              # FastAPI routers
│   ├── messaging/         # Redis, Kafka, RabbitMQ
│   └── config.py          # Pydantic Settings
└── main.py                # Composition root (DI)

Dependency Injection (FastAPI + DI Container)

# infrastructure/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str
    class Config:
        env_file = ".env"

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: init DB pool, Redis, etc.
    yield
    # Shutdown: close pools

app = FastAPI(lifespan=lifespan)

# Dependency providers
async def get_user_repo() -> UserRepository:
    return SQLAlchemyUserRepository(session_factory)

@app.get("/users/{user_id}")
async def get_user(user_id: str, repo: UserRepository = Depends(get_user_repo)):
    user = await repo.find_by_id(UserId(user_id))
    if not user:
        raise HTTPException(404)
    return user

Database Access

SQLAlchemy 2.0 (Async)

# Async session + typed models
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class UserModel(Base):
    __tablename__ = "users"
    id: Mapped[str] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

# Repository implementation
class SQLAlchemyUserRepository:
    def __init__(self, session_factory: async_sessionmaker[AsyncSession]):
        self._session_factory = session_factory

    async def find_by_id(self, id: UserId) -> Optional[User]:
        async with self._session_factory() as session:
            result = await session.get(UserModel, id.value)
            return self._to_domain(result) if result else None

    async def save(self, user: User) -> None:
        async with self._session_factory() as session:
            model = self._to_model(user)
            session.add(model)
            await session.commit()

Migrations: Alembic

alembic revision --autogenerate -m "add users table"
alembic upgrade head

Async Patterns

Structured Concurrency (TaskGroup)

async def process_batch(items: list[Item]) -> list[Result]:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process_item(item)) for item in items]
    return [task.result() for task in tasks]

# With timeout
async def with_timeout(coro, timeout: float):
    try:
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        raise TimeoutError(f"Operation timed out after {timeout}s")

Retry with Backoff (tenacity)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    wait=wait_exponential(multiplier=1, min=1, max=10),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(TransientError),
)
async def call_external_api(payload: dict) -> Response:
    ...

Testing

Pytest Patterns

# tests/unit/test_pricing.py
import pytest
from unittest.mock import AsyncMock
from domain.services.pricing import PricingService

class TestPricingService:
    @pytest.fixture
    def repo(self):
        return AsyncMock(spec=UserRepository)

    @pytest.fixture
    def service(self, repo):
        return PricingService(repo)

    async def test_premium_discount(self, service, repo):
        repo.find_by_id.return_value = User(id=UserId("1"), tier="premium")
        price = await service.calculate(UserId("1"), Decimal("100"))
        assert price == Decimal("90")

Property-Based Testing (Hypothesis)

from hypothesis import given, strategies as st

@given(st.integers(min_value=1, max_value=1000))
def test_invariant_holds(n):
    assert expensive_computation(n) == optimized_computation(n)

Contract Testing (Pact)

# Consumer-driven contract test
def test_user_api_contract(pact):
    pact.given("user exists").upon_receiving("request for user").with_request(
        method="GET", path="/users/123"
    ).will_respond_with(status=200, body=like({"id": "123", "email": "test@example.com"}))

Performance

• Async I/O — Use asyncpg, aiohttp, aiofiles; avoid sync libs in async code
• Connection pooling — SQLAlchemy async engine + pool_size, max_overflow
• Caching — aiocache (Redis/Memcached) with TTL + invalidation
• Profiling — py-spy, scalene, cProfile + snakeviz
• Memory — __slots__ on dataclasses; frozen=True for immutability
• Pydantic v2 — Rust-backed; 10-50x faster validation

Security

• Dependencies — pip-audit in CI; uv pip compile --generate-hashes
• Secrets — Pydantic Settings from env; never hardcode
• Input validation — Pydantic models on all boundaries (API, messaging, config)
• SQL injection — SQLAlchemy ORM/parameterized queries only
• CORS/CSRF — FastAPI CORSMiddleware; fastapi-csrf-protect for forms

Related Notes

• Language Agnostic Principles — Universal patterns
• TypeScript JavaScript — JS/TS patterns
• Go — Go patterns
• Rust — Rust patterns
• Domain Driven Design — DDD in Python
• Design Patterns — Patterns in Python
• Test Pyramid — Testing strategy
• CI CD Best Practices — Pipeline for Python

Sources: Python Docs (3.11+), Effective Python (Brett Slatkin), Real Python, FastAPI Docs, SQLAlchemy 2.0 Docs, Pydantic v2 Docs, ruff, mypy, pytest, uv, Astral (ruff/uv), Python Typing Best Practices