python-engineer

Python Engineering

Project Setup

python -m venv .venv
source .venv/bin/activate  # Unix
.venv\Scripts\activate     # Windows
pip install fastapi uvicorn pydantic sqlalchemy alembic pytest httpx

Project Structure

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app + lifespan
│   ├── config.py             # Settings from env vars
│   ├── models/               # SQLAlchemy models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/               # Pydantic schemas
│   │   └── user.py
│   ├── routes/                # API route handlers
│   │   └── users.py
│   ├── services/              # Business logic
│   │   └── user_service.py
│   ├── repositories/          # Database queries
│   │   └── user_repo.py
│   └── utils/                 # Helpers
│       └── auth.py
├── tests/
│   ├── conftest.py
│   └── test_users.py
├── alembic/
├── alembic.ini
├── pyproject.toml
├── requirements.txt
└── Dockerfile

FastAPI Patterns

Application Setup

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

@asynccontextmanager
async def lifespan(app: FastAPI):
    await database.connect()
    yield
    await database.disconnect()

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

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Route Handler

from fastapi import APIRouter, HTTPException, Depends, status
from pydantic import BaseModel, EmailStr, Field

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

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)
    password: str = Field(min_length=8, max_length=128)

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

    model_config = {"from_attributes": True}

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(data: CreateUserRequest, db: Session = Depends(get_db)):
    existing = await user_service.get_by_email(db, data.email)
    if existing:
        raise HTTPException(status_code=409, detail="Email already registered")
    user = await user_service.create(db, data)
    return user

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: str, db: Session = Depends(get_db)):
    user = await user_service.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Dependency Injection

from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    jwt_secret: str
    debug: bool = False

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

@lru_cache
def get_settings() -> Settings:
    return Settings()

async def get_db():
    async with AsyncSession(engine) as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Testing with pytest

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

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

@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/api/users/", json={
        "email": "test@example.com",
        "name": "Test User",
        "password": "SecurePass123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"

@pytest.mark.anyio
async def test_duplicate_email_returns_409(client: AsyncClient):
    payload = {"email": "dup@test.com", "name": "User", "password": "SecurePass123"}
    await client.post("/api/users/", json=payload)
    response = await client.post("/api/users/", json=payload)
    assert response.status_code == 409

CLI Tools (Typer)

import typer
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="My CLI tool")
console = Console()

@app.command()
def greet(name: str, count: int = typer.Option(1, help="Number of greetings")):
    for _ in range(count):
        console.print(f"Hello, [bold magenta]{name}[/bold magenta]!")

@app.command()
def list_users():
    table = Table(title="Users")
    table.add_column("ID", style="dim")
    table.add_column("Name", style="bold")
    table.add_column("Email")
    for user in get_users():
        table.add_row(user.id, user.name, user.email)
    console.print(table)

if __name__ == "__main__":
    app()

Data Processing (pandas)

import pandas as pd

def process_csv(input_path: str, output_path: str) -> None:
    df = pd.read_csv(input_path)
    df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_")
    df = df.dropna(subset=["email"])
    df["email"] = df["email"].str.lower().str.strip()
    df = df.drop_duplicates(subset=["email"], keep="last")
    df.to_csv(output_path, index=False)

Best Practices

• Type hints on ALL function signatures
• Use pydantic for data validation — never validate manually
• Use async for I/O-bound operations
• Use ruff for linting and formatting
• Use pyproject.toml over setup.py
• Never use bare except: — always catch specific exceptions
• Use pathlib.Path over os.path
• Use f-strings — never % formatting or .format()