roedyrustam 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": "test@test.com", "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
- Always use type hints — every function parameter and return type
- Pydantic v2 for all I/O validation — never trust raw dict inputs
- Async all the way — don't mix sync DB calls in async routes
- Repository pattern — no raw queries in routers
uvfor package management — faster, lockfile-based- Ruff for linting — replaces flake8 + isort + black
pytest-asynciowithasyncio_mode = "auto"— no manual@pytest.mark.asyncio- Settings via
pydantic-settings— neveros.environ.get()directly - Lifespan for startup/shutdown — not deprecated
@app.on_event mapped_columnwithMapped[]— SQLAlchemy 2.x style, not Column()