## Project Structure ``` my-fastapi-app/ ├── src/ │ ├── api/ │ │ ├── __init__.py │ │ ├── deps.py # Dependencies │ │ ├── routes/ │ │ │ ├── __init__.py │ │ │ ├── users.py │ │ │ ├── items.py │ │ │ └── auth.py │ │ └── middleware.py │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # Settings │ │ ├── security.py # Auth utilities │ │ └── exceptions.py # Custom exceptions │ ├── db/ │ │ ├── __init__.py │ │ ├── session.py # Database session │ │ ├── base.py # SQLAlchemy base │ │ └── repositories/ # Data access layer │ ├── models/ │ │ ├── __init__.py │ │ ├── user.py # SQLAlchemy models │ │ └── item.py │ ├── schemas/ │ │ ├── __init__.py │ │ ├── user.py # Pydantic schemas │ │ └── item.py │ ├── services/ │ │ ├── __init__.py │ │ ├── user.py # Business logic │ │ └── item.py │ └── main.py # Application entry ├── tests/ ├── alembic/ # Migrations ├── pyproject.toml └── Dockerfile ``` ## Core Patterns ### Application Setup ```python from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from src.core.config import settings from src.api.routes import users, items, auth from src.db.session import engine from src.db.base import Base @asynccontextmanager async def lifespan(app: FastAPI): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield await engine.dispose() app = FastAPI( title=settings.PROJECT_NAME, description="A high-performance API", version="1.0.0", lifespan=lifespan ) app.add_middleware( CORSMiddleware, allow_origins=settings.ALLOWED_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"]) app.include_router(users.router, prefix="/api/v1/users", tags=["users"]) app.include_router(items.router, prefix="/api/v1/items", tags=["items"]) @app.get("/health") async def health_check(): return {"status": "healthy"} ``` ### Configuration ```python from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): PROJECT_NAME: str = "FastAPI App" DEBUG: bool = False DATABASE_URL: str DB_POOL_SIZE: int = 5 DB_MAX_OVERFLOW: int = 10 SECRET_KEY: str ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 ALGORITHM: str = "HS256" ALLOWED_ORIGINS: list[str] = ["http://localhost:3000"] class Config: env_file = ".env" @lru_cache def get_settings() -> Settings: return Settings() settings = get_settings() ``` ### Pydantic Schemas ```python from datetime import datetime from pydantic import BaseModel, EmailStr, Field class UserBase(BaseModel): email: EmailStr name: str = Field(..., min_length=2, max_length=50) class UserCreate(UserBase): password: str = Field(..., min_length=8) class UserUpdate(BaseModel): email: EmailStr | None = None name: str | None = Field(None, min_length=2, max_length=50) class UserResponse(UserBase): id: int is_active: bool created_at: datetime class Config: from_attributes = True # For ORM mode class UserWithToken(UserResponse): access_token: str token_type: str = "bearer" ``` ### SQLAlchemy Models ```python from datetime import datetime from sqlalchemy import String, Boolean, DateTime from sqlalchemy.orm import Mapped, mapped_column, relationship from src.db.base import Base class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True, index=True) email: Mapped[str] = mapped_column(String(255), unique=True, index=True) name: Mapped[str] = mapped_column(String(50)) hashed_password: Mapped[str] = mapped_column(String(255)) is_active: Mapped[bool] = mapped_column(Boolean, default=True) created_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow ) items: Mapped[list["Item"]] = relationship( back_populates="owner", cascade="all, delete-orphan" ) ``` ### Database Session ```python from sqlalchemy.ext.asyncio import ( create_async_engine, AsyncSession, async_sessionmaker ) from src.core.config import settings engine = create_async_engine( settings.DATABASE_URL, pool_size=settings.DB_POOL_SIZE, max_overflow=settings.DB_MAX_OVERFLOW, echo=settings.DEBUG ) AsyncSessionLocal = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) async def get_db(): async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() ``` ### Dependencies ```python from typing import Annotated from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from sqlalchemy.ext.asyncio import AsyncSession from jose import JWTError, jwt from src.db.session import get_db from src.core.config import settings from src.services.user import UserService oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token") async def get_current_user( token: Annotated[str, Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(get_db)] ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode( token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM] ) user_id: int = payload.get("sub") if user_id is None: raise credentials_exception except JWTError: raise credentials_exception user_service = UserService(db) user = await user_service.get_by_id(user_id) if user is None: raise credentials_exception return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)] ): if not current_user.is_active: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user" ) return current_user CurrentUser = Annotated[User, Depends(get_current_active_user)] DbSession = Annotated[AsyncSession, Depends(get_db)] ``` ### Routes ```python from fastapi import APIRouter, HTTPException, status from src.api.deps import CurrentUser, DbSession from src.schemas.user import UserResponse, UserUpdate from src.services.user import UserService router = APIRouter() @router.get("/me", response_model=UserResponse) async def get_current_user(current_user: CurrentUser): """Get current user profile.""" return current_user @router.patch("/me", response_model=UserResponse) async def update_current_user( user_update: UserUpdate, current_user: CurrentUser, db: DbSession ): """Update current user profile.""" user_service = UserService(db) updated_user = await user_service.update( current_user.id, user_update ) return updated_user @router.get("/{user_id}", response_model=UserResponse) async def get_user(user_id: int, db: DbSession): """Get user by ID.""" user_service = UserService(db) user = await user_service.get_by_id(user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user ``` ### Services (Business Logic) ```python from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.models.user import User from src.schemas.user import UserCreate, UserUpdate from src.core.security import get_password_hash, verify_password class UserService: def __init__(self, db: AsyncSession): self.db = db async def get_by_id(self, user_id: int) -> User | None: result = await self.db.execute( select(User).where(User.id == user_id) ) return result.scalar_one_or_none() async def get_by_email(self, email: str) -> User | None: result = await self.db.execute( select(User).where(User.email == email) ) return result.scalar_one_or_none() async def create(self, user_in: UserCreate) -> User: user = User( email=user_in.email, name=user_in.name, hashed_password=get_password_hash(user_in.password) ) self.db.add(user) await self.db.flush() await self.db.refresh(user) return user async def update(self, user_id: int, user_in: UserUpdate) -> User: user = await self.get_by_id(user_id) update_data = user_in.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(user, field, value) await self.db.flush() await self.db.refresh(user) return user async def authenticate( self, email: str, password: str ) -> User | None: user = await self.get_by_email(email) if not user or not verify_password(password, user.hashed_password): return None return user ``` ### Error Handling ```python from fastapi import HTTPException, Request from fastapi.responses import JSONResponse class AppException(Exception): def __init__( self, status_code: int, detail: str, error_code: str | None = None ): self.status_code = status_code self.detail = detail self.error_code = error_code async def app_exception_handler( request: Request, exc: AppException ) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={ "error": { "message": exc.detail, "code": exc.error_code, "path": str(request.url) } } ) ``` ## Testing ```python import pytest from httpx import AsyncClient, ASGITransport from src.main import app @pytest.fixture async def client(): async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test" ) as client: yield client @pytest.mark.asyncio async def test_create_user(client: AsyncClient): response = await client.post( "/api/v1/auth/register", json={ "email": "test@example.com", "name": "Test User", "password": "securepassword123" } ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "id" in data @pytest.mark.asyncio async def test_login(client: AsyncClient): response = await client.post( "/api/v1/auth/token", data={ "username": "test@example.com", "password": "securepassword123" } ) assert response.status_code == 200 data = response.json() assert "access_token" in data ``` ## Best Practices 1. **Use Async Everywhere**: Leverage async/await for I/O operations 2. **Validate with Pydantic**: Type-safe request/response handling 3. **Dependency Injection**: Use FastAPI's DI for clean, testable code 4. **Layer Architecture**: Separate routes, services, and data access 5. **Handle Errors Gracefully**: Custom exception handlers 6. **Document APIs**: FastAPI auto-generates OpenAPI docs 7. **Use Type Hints**: Full typing for better IDE support 8. **Test Thoroughly**: Use pytest with httpx for async testing ## Related Resources - [FastAPI Documentation](https://fastapi.tiangolo.com/) - [Pydantic V2 Docs](https://docs.pydantic.dev/) - [SQLAlchemy 2.0](https://docs.sqlalchemy.org/) - [Alembic Migrations](https://alembic.sqlalchemy.org/)