python-engineer

name	python-engineer
description	Build production Python applications — FastAPI/Flask backends, async processing, data engineering with pandas, scripting automation, CLI tools with Typer, testing with pytest, type hints, virtual environments, and package management. Use when building Python backends, data pipelines, scripts, or CLI tools.

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": "[e-mail]",
        "name": "Test User",
        "password": "SecurePass123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[e-mail]"

@pytest.mark.anyio
async def test_duplicate_email_returns_409(client: AsyncClient):
    payload = {"email": "[e-mail]", "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()