martinholovsky/claude-skills-generator/fastapi-expert

# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
@pytest.fixture
async def async_client():
    """Async test client using httpx."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client
@pytest.mark.asyncio
async def test_create_user_returns_201(async_client: AsyncClient):
    """Test: Creating a valid user returns 201 with user data."""
    # Arrange
    user_data = {
        "email": "test@example.com",
        "username": "testuser",
        "password": "Test123!@#",
        "full_name": "Test User"
    }
    # Act
    response = await async_client.post("/api/v1/users/", json=user_data)
    # Assert
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert data["username"] == "testuser"
    assert "password" not in data  # Never expose password
    assert "id" in data
@pytest.mark.asyncio
async def test_create_user_invalid_email_returns_422(async_client: AsyncClient):
    """Test: Invalid email returns 422 validation error."""
    user_data = {
        "email": "not-an-email",
        "username": "testuser",
        "password": "Test123!@#",
        "full_name": "Test User"
    }
    response = await async_client.post("/api/v1/users/", json=user_data)
    assert response.status_code == 422
    assert "email" in str(response.json())
@pytest.mark.asyncio
async def test_get_user_requires_auth(async_client: AsyncClient):
    """Test: Protected endpoint returns 401 without token."""
    response = await async_client.get("/api/v1/users/me")
    assert response.status_code == 401
@pytest.mark.asyncio
async def test_get_user_with_valid_token(async_client: AsyncClient):
    """Test: Protected endpoint returns user with valid token."""
    # First login to get token
    login_response = await async_client.post(
        "/api/v1/auth/login",
        data={"username": "testuser", "password": "Test123!@#"}
    )
    token = login_response.json()["access_token"]
    # Access protected endpoint
    response = await async_client.get(
        "/api/v1/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    assert response.json()["username"] == "testuser"

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_db, get_current_user
from app.crud import user as user_crud
from app.schemas.user import UserCreate, UserResponse
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # Check if user exists
    existing = await user_crud.get_user_by_email(db, user_in.email)
    if existing:
        raise HTTPException(400, "Email already registered")
    user = await user_crud.create_user(db, user_in)
    return user
@router.get("/me", response_model=UserResponse)
async def get_current_user_info(
    current_user = Depends(get_current_user)
):
    return current_user

# Run all tests with coverage
pytest tests/ -v --cov=app --cov-report=term-missing
 
# Type checking
mypy app/
 
# Security audit
pip-audit
safety check
 
# Run linting
ruff check app/

# conftest.py - Full async test setup
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.main import app
from app.db.session import get_db
from app.db.models import Base
# Test database
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
@pytest_asyncio.fixture
async def test_db():
    """Create test database and tables."""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    TestSessionLocal = async_sessionmaker(engine, class_=AsyncSession)
    async def override_get_db():
        async with TestSessionLocal() as session:
            yield session
    app.dependency_overrides[get_db] = override_get_db
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    app.dependency_overrides.clear()
@pytest_asyncio.fixture
async def async_client(test_db):
    """Async client with test database."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client

# app/main.py - Production-ready structure
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.v1.router import api_router
app = FastAPI(
    title=settings.PROJECT_NAME,
    docs_url="/api/docs" if settings.ENVIRONMENT != "production" else None,
    openapi_url="/api/openapi.json",
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,  # Never ["*"] in production!
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
)
app.include_router(api_router, prefix="/api/v1")
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

from pydantic import BaseModel, Field, EmailStr, field_validator
from pydantic.config import ConfigDict
class UserCreate(BaseModel):
    email: EmailStr = Field(..., description="User email")
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_-]+$")
    password: str = Field(..., min_length=8, max_length=100)
    full_name: str = Field(..., min_length=1, max_length=100)
    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        if not any(c in '!@#$%^&*()_+-=' for c in v):
            raise ValueError('Password must contain special character')
        return v
class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr
    username: str
    full_name: str
    is_active: bool
    # ❌ NEVER include: password_hash, tokens, secrets

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_recycle=3600,
)
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
# app/db/models.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Boolean, DateTime
from datetime import datetime
class Base(DeclarativeBase):
    pass