Python Development Cheatsheet

myproject/
├── src/
│   └── myproject/
│       ├── __init__.py
│       ├── main.py              # FastAPI app entry
│       ├── config.py            # settings / env vars
│       ├── models/
│       │   ├── __init__.py
│       │   └── user.py          # SQLAlchemy models
│       ├── schemas/
│       │   ├── __init__.py
│       │   └── user.py          # Pydantic schemas
│       ├── routers/
│       │   ├── __init__.py
│       │   └── users.py         # API routes
│       ├── services/
│       │   └── user_service.py  # business logic
│       ├── repositories/
│       │   └── user_repo.py     # data access
│       └── utils/
│           └── security.py
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_services.py
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env
└── README.md

# venv (built-in)
python -m venv .venv
source .venv/bin/activate        # macOS/Linux
.venv\Scripts\activate           # Windows
deactivate

# pip
pip install fastapi uvicorn
pip install -r requirements.txt
pip freeze > requirements.txt

# Poetry (modern)
poetry init
poetry add fastapi uvicorn
poetry add --group dev pytest httpx
poetry install
poetry run python main.py
poetry shell                     # activate venv

# uv (blazing fast, Rust-based)
uv venv
uv pip install fastapi
uv pip compile requirements.in -o requirements.txt
uv pip sync requirements.txt

from typing import Optional, Union, TypeAlias, TypeVar, Generic

# Basic types
name: str = "Bob"
age: int = 25
active: bool = True
score: float = 9.5

# Collections (Python 3.9+ — use built-in generics)
names: list[str] = ["a", "b"]
lookup: dict[str, int] = {"a": 1}
pair: tuple[str, int] = ("age", 25)
ids: set[int] = {1, 2, 3}

# Optional & Union (Python 3.10+ → use |)
email: str | None = None          # same as Optional[str]
value: int | str = 42             # same as Union[int, str]

# Function signatures
def greet(name: str, times: int = 1) -> str:
    return name * times

# Callable
from collections.abc import Callable
handler: Callable[[str, int], bool]

# TypeVar & Generic
T = TypeVar("T")
def first(items: list[T]) -> T:
    return items[0]

# TypeAlias
UserId: TypeAlias = int
JSON: TypeAlias = dict[str, any]

# Literal
from typing import Literal
def set_mode(mode: Literal["read", "write"]) -> None: ...

# TypedDict
from typing import TypedDict
class UserDict(TypedDict):
    name: str
    age: int
    email: str | None

# Protocol (structural subtyping)
from typing import Protocol
class HasName(Protocol):
    name: str

def greet(obj: HasName) -> str:
    return f"Hello {obj.name}"

from fastapi import FastAPI, HTTPException, Depends, Query, Path
from pydantic import BaseModel, Field, EmailStr

app = FastAPI(title="My API", version="1.0.0")

# Pydantic schemas
class CreateUser(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(ge=18, le=150)

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    model_config = {"from_attributes": True}  # enables ORM mode

# Routes
@app.get("/users", response_model=list[UserResponse])
async def get_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
):
    return await user_service.get_all(skip, limit)

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int = Path(..., gt=0)):
    user = await user_service.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(data: CreateUser):
    return await user_service.create(data)

@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
    await user_service.delete(user_id)

# Database session dependency
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Auth dependency
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# Use in route
@app.get("/me")
async def profile(
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    return user

# Custom middleware
@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logger.info(f"{request.method} {request.url.path} {response.status_code} {duration:.3f}s")
    return response

# Exception handler
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(status_code=400, content={"detail": str(exc)})

# CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"])

# Run
# uvicorn main:app --reload --port 8000

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker

# Engine & Session
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

# Modern declarative models (SQLAlchemy 2.0+)
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

# CRUD operations
def get_user(db: Session, user_id: int):
    return db.get(User, user_id)

def get_users(db: Session, skip=0, limit=10):
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, name: str, email: str):
    user = User(name=name, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

# Select statement (2.0 style)
from sqlalchemy import select
stmt = select(User).where(User.email == "bob@mail.com")
result = db.execute(stmt).scalar_one_or_none()

import asyncio
import httpx

# Basic async function
async def fetch_data(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.json()

# Run multiple concurrently
async def main():
    results = await asyncio.gather(
        fetch_data("https://api.example.com/a"),
        fetch_data("https://api.example.com/b"),
        fetch_data("https://api.example.com/c"),
    )
    return results

asyncio.run(main())

# Async generator
async def paginate(url):
    page = 1
    while True:
        data = await fetch_data(f"{url}?page={page}")
        if not data: break
        yield data
        page += 1

async for batch in paginate("/api/users"):
    process(batch)

# Semaphore (limit concurrency)
sem = asyncio.Semaphore(10)
async def limited_fetch(url):
    async with sem:
        return await fetch_data(url)

# Timeout
async with asyncio.timeout(5.0):
    result = await slow_operation()

import functools
import time

# Basic decorator
def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} took {time.time() - start:.3f}s")
        return result
    return wrapper

@timer
def slow_fn():
    time.sleep(1)

# Decorator with arguments
def retry(max_retries=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1: raise
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_retries=5, delay=2)
def flaky_api_call(): ...

# Class-based decorator
class Cache:
    def __init__(self, func):
        self.func = func
        self.cache = {}
    def __call__(self, *args):
        if args not in self.cache:
            self.cache[args] = self.func(*args)
        return self.cache[args]

# Built-in decorators
@property              # getter
@staticmethod          # no self/cls
@classmethod           # receives cls
@functools.lru_cache   # memoization
@functools.cached_property  # computed once
@dataclass             # auto __init__, __repr__, etc.

# Built-in
with open("file.txt") as f:
    content = f.read()

# Custom (class-based)
class DBConnection:
    def __enter__(self):
        self.conn = connect_db()
        return self.conn
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.conn.close()
        return False  # don't suppress exceptions

# Custom (decorator-based — simpler)
from contextlib import contextmanager

@contextmanager
def timer(label):
    start = time.time()
    yield
    print(f"{label}: {time.time() - start:.3f}s")

with timer("processing"):
    heavy_work()

# Async context manager
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # startup
    db = await connect_db()
    yield {"db": db}
    # shutdown
    await db.close()

# Install: pip install pytest pytest-asyncio httpx

# Basic test
def test_add():
    assert add(2, 3) == 5

# Parametrize
import pytest

@pytest.mark.parametrize("a, b, expected", [
    (1, 2, 3),
    (-1, 1, 0),
    (0, 0, 0),
])
def test_add_parametrized(a, b, expected):
    assert add(a, b) == expected

# Fixtures
@pytest.fixture
def sample_user():
    return User(name="Bob", email="bob@mail.com")

def test_user_name(sample_user):
    assert sample_user.name == "Bob"

# Exception testing
def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(1, 0)

# Mock
from unittest.mock import patch, MagicMock

@patch("myproject.services.user_service.user_repo")
def test_create_user(mock_repo):
    mock_repo.create.return_value = User(id=1, name="Bob")
    result = user_service.create({"name": "Bob"})
    assert result.name == "Bob"
    mock_repo.create.assert_called_once()

# Async test
@pytest.mark.asyncio
async def test_fetch():
    result = await fetch_data("/api/health")
    assert result["status"] == "ok"

# FastAPI test client
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_create_user_api():
    async with AsyncClient(app=app, base_url="http://test") as client:
        resp = await client.post("/users", json={"name": "Bob", "email": "bob@mail.com"})
        assert resp.status_code == 201

# Run: pytest -v
# Run: pytest -k "test_add" --tb=short
# Run: pytest --cov=myproject

import logging

# Basic setup
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

logger.debug("Debug message")
logger.info("User created: %s", user.name)
logger.warning("Deprecated endpoint called")
logger.error("Failed to connect", exc_info=True)
logger.critical("System is down!")

# Structured logging with structlog
import structlog
log = structlog.get_logger()
log.info("user.created", user_id=123, email="bob@mail.com")

# Pydantic Settings (recommended for FastAPI)
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My App"
    debug: bool = False
    database_url: str
    jwt_secret: str
    redis_url: str = "redis://localhost:6379"

    model_config = {"env_file": ".env"}

settings = Settings()

# .env file
# DATABASE_URL=postgresql://user:pass@localhost/db
# JWT_SECRET=mysecretkey

# os.environ (basic)
import os
db_url = os.environ.get("DATABASE_URL", "sqlite:///db.sqlite3")

# python-dotenv
from dotenv import load_dotenv
load_dotenv()

[project]
name = "myproject"
version = "0.1.0"
description = "My awesome project"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.110",
    "uvicorn[standard]>=0.27",
    "sqlalchemy>=2.0",
    "pydantic-settings>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "httpx>=0.27",
    "ruff>=0.3",
    "mypy>=1.8",
]

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

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP"]

[tool.mypy]
strict = true

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

# Multi-stage Dockerfile
FROM python:3.12-slim AS base
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM base AS production
COPY src/ ./src/
EXPOSE 8000
CMD ["uvicorn", "src.myproject.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
services:
  app:
    build: .
    ports: ["8000:8000"]
    env_file: .env
    depends_on: [db, redis]
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: mydb
      POSTGRES_PASSWORD: secret
    volumes: ["pgdata:/var/lib/postgresql/data"]
  redis:
    image: redis:7-alpine
volumes:
  pgdata:

# repository
class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def get_by_id(self, user_id: int) -> User | None:
        return self.db.get(User, user_id)

    def create(self, user: User) -> User:
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user

# service
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def get_user(self, user_id: int) -> UserResponse:
        user = self.repo.get_by_id(user_id)
        if not user:
            raise HTTPException(404, "Not found")
        return UserResponse.model_validate(user)

class Singleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance