# Python

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

---

## Modern Python (3.11+)

### Type Hints (Mandatory for New Code)
```python
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+)
```python
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+)
```python
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+)

| Category | Recommended | Alternatives |
|---|---|---|
| **Package Manager** | **uv** (fast, Rust-based) | Poetry, PDM, pip-tools |
| **Virtual Env** | uv venv / uv pip | venv, virtualenv |
| **Linting** | **ruff** (fast, replaces flake8/isort/black) | flake8 + black + isort |
| **Type Checking** | **mypy** (strict) | pyright, pyre |
| **Formatting** | **ruff format** (Black-compatible) | Black |
| **Testing** | **pytest** + pytest-asyncio | unittest |
| **Test Coverage** | pytest-cov | coverage.py |
| **Pre-commit** | **pre-commit** + ruff + mypy | Husky (JS-style) |
| **Dependency Security** | pip-audit, Safety | Snyk, Dependabot |

### pyproject.toml (Single Config File)
```toml
[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)
```python
# 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)
```python
# 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
```bash
alembic revision --autogenerate -m "add users table"
alembic upgrade head
```

---

## Async Patterns

### Structured Concurrency (TaskGroup)
```python
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)
```python
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
```python
# 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)
```python
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)
```python
# 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

- [[Languages/Language-Agnostic-Principles]] — Universal patterns
- [[Languages/TypeScript-JavaScript]] — JS/TS patterns
- [[Languages/Go]] — Go patterns
- [[Languages/Rust]] — Rust patterns
- [[Architecture/Domain-Driven-Design]] — DDD in Python
- [[Patterns/Design-Patterns]] — Patterns in Python
- [[Testing/Test-Pyramid]] — Testing strategy
- [[DevOps/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*