project/
├── app/
│   ├── __init__.py
│   ├── main.py           # Application entry point
│   ├── config.py         # Settings with pydantic-settings
│   ├── dependencies.py   # Shared dependencies
│   ├── api/
│   │   ├── __init__.py
│   │   ├── deps.py       # API dependencies
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py # API v1 router
│   │   │   └── endpoints/
│   │   │       ├── users.py
│   │   │       └── items.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── security.py   # Auth utilities
│   │   └── exceptions.py # Custom exceptions
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py       # SQLAlchemy models
│   │   └── item.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py       # Pydantic schemas
│   │   └── item.py
│   ├── crud/
│   │   ├── __init__.py
│   │   ├── base.py       # Generic CRUD
│   │   └── user.py       # User CRUD
│   └── db/
│       ├── __init__.py
│       ├── session.py    # Database session
│       └── base.py       # Base model
└── tests/

from fastapi import APIRouter, Depends, HTTPException, status
from typing import Annotated

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

# Use Annotated for cleaner dependency injection
CurrentUser = Annotated[User, Depends(get_current_user)]
DbSession = Annotated[AsyncSession, Depends(get_db)]

@router.get("/", response_model=list[UserRead])
async def list_users(
    db: DbSession,
    skip: int = 0,
    limit: int = 100,
) -> list[User]:
    """List all users with pagination."""
    return await crud.user.get_multi(db, skip=skip, limit=limit)

@router.get("/{user_id}", response_model=UserRead)
async def get_user(
    user_id: int,
    db: DbSession,
) -> User:
    """Get user by ID."""
    user = await crud.user.get(db, id=user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )
    return user

@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: DbSession,
) -> User:
    """Create new user."""
    return await crud.user.create(db, obj_in=user_in)

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated, AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Database session dependency."""
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """Get current authenticated user."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = await crud.user.get(db, id=user_id)
    if user is None:
        raise credentials_exception
    return user

def require_admin(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    """Require admin role."""
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return current_user

from pydantic import BaseModel, ConfigDict, Field, EmailStr
from datetime import datetime

class UserBase(BaseModel):
    """Shared user properties."""
    email: EmailStr
    full_name: str | None = None

class UserCreate(UserBase):
    """User creation schema."""
    password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):
    """User update schema (all optional)."""
    email: EmailStr | None = None
    full_name: str | None = None
    password: str | None = Field(default=None, min_length=8)

class UserRead(UserBase):
    """User response schema."""
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    is_active: bool
    created_at: datetime

class UserInDB(UserRead):
    """User with hashed password."""
    hashed_password: str

import asyncio
from httpx import AsyncClient

# Concurrent requests
async def fetch_all_data(urls: list[str]) -> list[dict]:
    async with AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

# Background tasks
from fastapi import BackgroundTasks

@router.post("/send-notification")
async def send_notification(
    email: EmailStr,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(send_email, email)
    return {"message": "Notification scheduled"}

# Lifespan events
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await init_db()
    await create_redis_pool()
    yield
    # Shutdown
    await close_redis_pool()

app = FastAPI(lifespan=lifespan)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

# Validation error customization
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "errors": exc.errors(),
        },
    )

# Sync operations in async functions
@app.get("/bad")
async def bad_endpoint():
    time.sleep(1)  # Blocks the event loop!
    return {"status": "slow"}

# Missing type hints
@app.post("/users")
def create_user(user):  # No type = no validation
    return user

# Hardcoded secrets
SECRET = "my-secret-key"  # Never do this!

# Async operations
@app.get("/good")
async def good_endpoint():
    await asyncio.sleep(1)  # Non-blocking
    return {"status": "fast"}

# Full type hints with Pydantic
@app.post("/users", response_model=UserRead)
async def create_user(user: UserCreate) -> User:
    return await crud.user.create(user)

# Environment configuration
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    secret_key: str
    
    model_config = ConfigDict(env_file=".env")