Python programming expert

Python Programming Expert (3.12+)

Modern Python patterns emphasizing type safety, async-first design, and production quality.

---

Core Stack

Layer	Tool
Runtime	Python 3.12+
Package manager	uv (preferred) or Poetry
Type checking	mypy (strict) or pyright
Linting/formatting	Ruff
Web framework	FastAPI
ORM	SQLAlchemy 2.x or SQLModel
Validation	Pydantic v2
Testing	pytest + pytest-asyncio
Task queue	Celery or ARQ (async)

---

Project Setup with uv

# Create new project
uv init my-app
cd my-app

# Add dependencies
uv add fastapi "uvicorn[standard]" pydantic sqlalchemy alembic

# Add dev dependencies
uv add --dev ruff mypy pytest pytest-asyncio httpx

# Run
uv run uvicorn app.main:app --reload

# pyproject.toml is the single source of truth

pyproject.toml (full config)

[project]
name = "my-app"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "uvicorn[standard]>=0.32",
    "pydantic>=2.9",
    "sqlalchemy>=2.0",
    "alembic>=1.14",
]

[project.optional-dependencies]
dev = ["ruff", "mypy", "pytest", "pytest-asyncio", "httpx"]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "TCH"]
ignore = ["E501"]

[tool.mypy]
python_version = "3.12"
strict = true
ignore_missing_imports = true

[tool.pytest.ini_options]
asyncio_mode = "auto"

---

Type Safety Patterns

Modern Type Hints (Python 3.12)

# Use built-in generics — no need to import from typing
def process(items: list[str]) -> dict[str, int]:
    return {item: len(item) for item in items}

# New type alias syntax
type UserId = str
type PostSlug = str

# TypeVar with bounds
from typing import TypeVar
T = TypeVar("T", bound="BaseModel")

# ParamSpec for decorator typing
from typing import ParamSpec, Callable
P = ParamSpec("P")

def retry(fn: Callable[P, T]) -> Callable[P, T]: ...

Pydantic v2 Models

from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import datetime
from uuid import UUID, uuid4

class UserCreate(BaseModel):
    email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(..., ge=0, le=150)

    @field_validator("email")
    @classmethod
    def email_lowercase(cls, v: str) -> str:
        return v.lower().strip()

class User(BaseModel):
    id: UUID = Field(default_factory=uuid4)
    email: str
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = True

    model_config = {"from_attributes": True}  # replaces orm_mode

class UserWithPosts(User):
    posts: list["Post"] = []

    @model_validator(mode="after")
    def check_active_has_posts(self) -> "UserWithPosts":
        if not self.is_active and self.posts:
            raise ValueError("Inactive users should not have posts")
        return self

---

Async / Asyncio Patterns

TaskGroups (Python 3.11+)

import asyncio
from asyncio import TaskGroup

async def fetch_user(user_id: str) -> dict:
    await asyncio.sleep(0.1)  # simulate DB call
    return {"id": user_id, "name": "Alice"}

async def fetch_posts(user_id: str) -> list[dict]:
    await asyncio.sleep(0.1)
    return [{"title": "Post 1"}]

async def get_user_dashboard(user_id: str) -> dict:
    # Run concurrently — both start immediately
    async with TaskGroup() as tg:
        user_task = tg.create_task(fetch_user(user_id))
        posts_task = tg.create_task(fetch_posts(user_id))
    # Both done here — exceptions propagate automatically
    return {
        "user": user_task.result(),
        "posts": posts_task.result(),
    }

Async Context Managers

from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

---

FastAPI Patterns

App Structure

app/
├── main.py
├── core/
│   ├── config.py      # Settings via pydantic-settings
│   └── database.py    # Engine + session factory
├── models/            # SQLAlchemy models
├── schemas/           # Pydantic schemas
├── repositories/      # DB access layer
├── routers/           # FastAPI routers
└── dependencies/      # Reusable FastAPI deps

Settings with pydantic-settings

# core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    debug: bool = False
    cors_origins: list[str] = ["http://localhost:3000"]

settings = Settings()

Router with Dependencies

# routers/posts.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_session
from app.schemas.post import PostCreate, PostResponse
from app.repositories.post import PostRepository
from app.dependencies.auth import get_current_user

router = APIRouter(prefix="/posts", tags=["posts"])

@router.get("/", response_model=list[PostResponse])
async def list_posts(
    page: int = 1,
    limit: int = 10,
    session: AsyncSession = Depends(get_session),
) -> list[PostResponse]:
    repo = PostRepository(session)
    return await repo.find_published(limit=limit, offset=(page - 1) * limit)

@router.post("/", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
async def create_post(
    data: PostCreate,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
) -> PostResponse:
    repo = PostRepository(session)
    post = await repo.create(user_id=current_user.id, **data.model_dump())
    return PostResponse.model_validate(post)

@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(
    post_id: str,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
) -> None:
    repo = PostRepository(session)
    post = await repo.find_by_id(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    if post.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Forbidden")
    await repo.delete(post_id)

Lifespan (replaces on_event)

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.core.database import engine
from app.core.config import settings

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown
    await engine.dispose()

app = FastAPI(
    title="My API",
    lifespan=lifespan,
    debug=settings.debug,
)

---

SQLAlchemy 2.x (Async)

# models/post.py
from sqlalchemy import String, Text, Boolean, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID
import uuid

class Base(DeclarativeBase):
    pass

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(UUID, ForeignKey("users.id"), nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str | None] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    user: Mapped["User"] = relationship(back_populates="posts")

---

Testing with pytest

# tests/test_posts.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

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

@pytest.fixture
async def auth_headers(client: AsyncClient) -> dict:
    resp = await client.post("/auth/token", data={"username": "[email protected]", "password": "pass"})
    token = resp.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

async def test_create_post(client: AsyncClient, auth_headers: dict) -> None:
    resp = await client.post(
        "/posts/",
        json={"title": "Hello", "content": "World"},
        headers=auth_headers,
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "Hello"
    assert "id" in data

---

Key Rules

1. Always use type hints — every function parameter and return type
2. Pydantic v2 for all I/O validation — never trust raw dict inputs
3. Async all the way — don't mix sync DB calls in async routes
4. Repository pattern — no raw queries in routers
5. uv for package management — faster, lockfile-based
6. Ruff for linting — replaces flake8 + isort + black
7. pytest-asyncio with asyncio_mode = "auto" — no manual @pytest.mark.asyncio
8. Settings via pydantic-settings — never os.environ.get() directly
9. Lifespan for startup/shutdown — not deprecated @app.on_event
10. mapped_column with Mapped[] — SQLAlchemy 2.x style, not Column()