fastapi-expert Checked

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from app.schemas import UserCreate
from app.security import hash_password

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

async def create_user(db: AsyncSession, payload: UserCreate) -> User:
    user = User(email=payload.email, hashed_password=hash_password(payload.password), name=payload.name)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

JWT Authentication Snippet

# security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

SECRET_KEY = "read-from-env"  # use os.environ / settings
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_access_token(subject: str, expires_delta: timedelta = timedelta(minutes=30)) -> str:
    payload = {"sub": subject, "exp": datetime.now(timezone.utc) + expires_delta}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
    try:
        data = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        subject: str | None = data.get("sub")
        if subject is None:
            raise ValueError
        return subject
    except (JWTError, ValueError):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

CurrentUser = Annotated[str, Depends(get_current_user)]

Reference Guide

Load detailed guidance based on context:

| Topic | Reference | Load When |

|-------|-----------|-----------|

| Pydantic V2 | references/pydantic-v2.md | Creating schemas, validation, model_config |

| SQLAlchemy | references/async-sqlalchemy.md | Async database, models, CRUD operations |

| Endpoints | references/endpoints-routing.md | APIRouter, dependencies, routing |

| Authentication | references/authentication.md | JWT, OAuth2, get_current_user |

| Testing | references/testing-async.md | pytest-asyncio, httpx, fixtures |

| Django Migration | references/migration-from-django.md | Migrating from Django/DRF to FastAPI |

Constraints

### MUST DO

• Use type hints everywhere (FastAPI requires them)
• Use Pydantic V2 syntax (field_validator, model_validator, model_config)
• Use Annotated pattern for dependency injection
• Use async/await for all I/O operations
• Use X | None instead of Optional[X]
• Return proper HTTP status codes
• Document endpoints (auto-generated OpenAPI)

### MUST NOT DO

• Use synchronous database operations
• Skip Pydantic validation
• Store passwords in plain text
• Expose sensitive data in responses
• Use Pydantic V1 syntax (@validator, class Config)
• Mix sync and async code improperly
• Hardcode configuration values

Output Templates

When implementing FastAPI features, provide:

1. Schema file (Pydantic models)

2. Endpoint file (router with endpoints)

3. CRUD operations if database involved

4. Brief explanation of key decisions

Knowledge Reference

FastAPI, Pydantic V2, async SQLAlchemy, Alembic migrations, JWT/OAuth2, pytest-asyncio, httpx, BackgroundTasks, WebSockets, dependency injection, OpenAPI/Swagger

Documentation