python-programming-expert

name

python-programming-expert

description

Expert skill for Python programming (Python 3.12+). Use whenever the user writes, reviews, or debugs Python code — covering type safety (typing, Pydantic), async/await (asyncio, TaskGroups), web backends (FastAPI, SQLAlchemy, SQLModel), package management (uv, Poetry), code quality (Ruff), and testing with pytest. Trigger on any mention of Python, FastAPI, Pydantic, asyncio, uv, Poetry, SQLAlchemy, or pytest. Also trigger for data pipelines, CLI tools, or scripting tasks in Python.

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": "[e-mail]", "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()