1# fastapi-templates — detailed worked examples
2 
3## Implementation Patterns
4 
5### Pattern 1: Complete FastAPI Application
6 
7```python
8# main.py
9from fastapi import FastAPI, Depends
10from fastapi.middleware.cors import CORSMiddleware
11from contextlib import asynccontextmanager
12 
13@asynccontextmanager
14async def lifespan(app: FastAPI):
15    """Application lifespan events."""
16    # Startup
17    await database.connect()
18    yield
19    # Shutdown
20    await database.disconnect()
21 
22app = FastAPI(
23    title="API Template",
24    version="1.0.0",
25    lifespan=lifespan
26)
27 
28# CORS middleware
29app.add_middleware(
30    CORSMiddleware,
31    allow_origins=["*"],
32    allow_credentials=True,
33    allow_methods=["*"],
34    allow_headers=["*"],
35)
36 
37# Include routers
38from app.api.v1.router import api_router
39app.include_router(api_router, prefix="/api/v1")
40 
41# core/config.py
42from pydantic_settings import BaseSettings
43from functools import lru_cache
44 
45class Settings(BaseSettings):
46    """Application settings."""
47    DATABASE_URL: str
48    SECRET_KEY: str
49    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
50    API_V1_STR: str = "/api/v1"
51 
52    class Config:
53        env_file = ".env"
54 
55@lru_cache()
56def get_settings() -> Settings:
57    return Settings()
58 
59# core/database.py
60from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
61from sqlalchemy.ext.declarative import declarative_base
62from sqlalchemy.orm import sessionmaker
63from app.core.config import get_settings
64 
65settings = get_settings()
66 
67engine = create_async_engine(
68    settings.DATABASE_URL,
69    echo=True,
70    future=True
71)
72 
73AsyncSessionLocal = sessionmaker(
74    engine,
75    class_=AsyncSession,
76    expire_on_commit=False
77)
78 
79Base = declarative_base()
80 
81async def get_db() -> AsyncSession:
82    """Dependency for database session."""
83    async with AsyncSessionLocal() as session:
84        try:
85            yield session
86            await session.commit()
87        except Exception:
88            await session.rollback()
89            raise
90        finally:
91            await session.close()
92```
93 
94### Pattern 2: CRUD Repository Pattern
95 
96```python
97# repositories/base_repository.py
98from typing import Generic, TypeVar, Type, Optional, List
99from sqlalchemy.ext.asyncio import AsyncSession
100from sqlalchemy import select
101from pydantic import BaseModel
102 
103ModelType = TypeVar("ModelType")
104CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
105UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
106 
107class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
108    """Base repository for CRUD operations."""
109 
110    def __init__(self, model: Type[ModelType]):
111        self.model = model
112 
113    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
114        """Get by ID."""
115        result = await db.execute(
116            select(self.model).where(self.model.id == id)
117        )
118        return result.scalars().first()
119 
120    async def get_multi(
121        self,
122        db: AsyncSession,
123        skip: int = 0,
124        limit: int = 100
125    ) -> List[ModelType]:
126        """Get multiple records."""
127        result = await db.execute(
128            select(self.model).offset(skip).limit(limit)
129        )
130        return result.scalars().all()
131 
132    async def create(
133        self,
134        db: AsyncSession,
135        obj_in: CreateSchemaType
136    ) -> ModelType:
137        """Create new record."""
138        db_obj = self.model(**obj_in.dict())
139        db.add(db_obj)
140        await db.flush()
141        await db.refresh(db_obj)
142        return db_obj
143 
144    async def update(
145        self,
146        db: AsyncSession,
147        db_obj: ModelType,
148        obj_in: UpdateSchemaType
149    ) -> ModelType:
150        """Update record."""
151        update_data = obj_in.dict(exclude_unset=True)
152        for field, value in update_data.items():
153            setattr(db_obj, field, value)
154        await db.flush()
155        await db.refresh(db_obj)
156        return db_obj
157 
158    async def delete(self, db: AsyncSession, id: int) -> bool:
159        """Delete record."""
160        obj = await self.get(db, id)
161        if obj:
162            await db.delete(obj)
163            return True
164        return False
165 
166# repositories/user_repository.py
167from app.repositories.base_repository import BaseRepository
168from app.models.user import User
169from app.schemas.user import UserCreate, UserUpdate
170 
171class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):
172    """User-specific repository."""
173 
174    async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:
175        """Get user by email."""
176        result = await db.execute(
177            select(User).where(User.email == email)
178        )
179        return result.scalars().first()
180 
181    async def is_active(self, db: AsyncSession, user_id: int) -> bool:
182        """Check if user is active."""
183        user = await self.get(db, user_id)
184        return user.is_active if user else False
185 
186user_repository = UserRepository(User)
187```
188 
189### Pattern 3: Service Layer
190 
191```python
192# services/user_service.py
193from typing import Optional
194from sqlalchemy.ext.asyncio import AsyncSession
195from app.repositories.user_repository import user_repository
196from app.schemas.user import UserCreate, UserUpdate, User
197from app.core.security import get_password_hash, verify_password
198 
199class UserService:
200    """Business logic for users."""
201 
202    def __init__(self):
203        self.repository = user_repository
204 
205    async def create_user(
206        self,
207        db: AsyncSession,
208        user_in: UserCreate
209    ) -> User:
210        """Create new user with hashed password."""
211        # Check if email exists
212        existing = await self.repository.get_by_email(db, user_in.email)
213        if existing:
214            raise ValueError("Email already registered")
215 
216        # Hash password
217        user_in_dict = user_in.dict()
218        user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))
219 
220        # Create user
221        user = await self.repository.create(db, UserCreate(**user_in_dict))
222        return user
223 
224    async def authenticate(
225        self,
226        db: AsyncSession,
227        email: str,
228        password: str
229    ) -> Optional[User]:
230        """Authenticate user."""
231        user = await self.repository.get_by_email(db, email)
232        if not user:
233            return None
234        if not verify_password(password, user.hashed_password):
235            return None
236        return user
237 
238    async def update_user(
239        self,
240        db: AsyncSession,
241        user_id: int,
242        user_in: UserUpdate
243    ) -> Optional[User]:
244        """Update user."""
245        user = await self.repository.get(db, user_id)
246        if not user:
247            return None
248 
249        if user_in.password:
250            user_in_dict = user_in.dict(exclude_unset=True)
251            user_in_dict["hashed_password"] = get_password_hash(
252                user_in_dict.pop("password")
253            )
254            user_in = UserUpdate(**user_in_dict)
255 
256        return await self.repository.update(db, user, user_in)
257 
258user_service = UserService()
259```
260 
261### Pattern 4: API Endpoints with Dependencies
262 
263```python
264# api/v1/endpoints/users.py
265from fastapi import APIRouter, Depends, HTTPException, status
266from sqlalchemy.ext.asyncio import AsyncSession
267from typing import List
268 
269from app.core.database import get_db
270from app.schemas.user import User, UserCreate, UserUpdate
271from app.services.user_service import user_service
272from app.api.dependencies import get_current_user
273 
274router = APIRouter()
275 
276@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
277async def create_user(
278    user_in: UserCreate,
279    db: AsyncSession = Depends(get_db)
280):
281    """Create new user."""
282    try:
283        user = await user_service.create_user(db, user_in)
284        return user
285    except ValueError as e:
286        raise HTTPException(status_code=400, detail=str(e))
287 
288@router.get("/me", response_model=User)
289async def read_current_user(
290    current_user: User = Depends(get_current_user)
291):
292    """Get current user."""
293    return current_user
294 
295@router.get("/{user_id}", response_model=User)
296async def read_user(
297    user_id: int,
298    db: AsyncSession = Depends(get_db),
299    current_user: User = Depends(get_current_user)
300):
301    """Get user by ID."""
302    user = await user_service.repository.get(db, user_id)
303    if not user:
304        raise HTTPException(status_code=404, detail="User not found")
305    return user
306 
307@router.patch("/{user_id}", response_model=User)
308async def update_user(
309    user_id: int,
310    user_in: UserUpdate,
311    db: AsyncSession = Depends(get_db),
312    current_user: User = Depends(get_current_user)
313):
314    """Update user."""
315    if current_user.id != user_id:
316        raise HTTPException(status_code=403, detail="Not authorized")
317 
318    user = await user_service.update_user(db, user_id, user_in)
319    if not user:
320        raise HTTPException(status_code=404, detail="User not found")
321    return user
322 
323@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
324async def delete_user(
325    user_id: int,
326    db: AsyncSession = Depends(get_db),
327    current_user: User = Depends(get_current_user)
328):
329    """Delete user."""
330    if current_user.id != user_id:
331        raise HTTPException(status_code=403, detail="Not authorized")
332 
333    deleted = await user_service.repository.delete(db, user_id)
334    if not deleted:
335        raise HTTPException(status_code=404, detail="User not found")
336```
337 
338### Pattern 5: Authentication & Authorization
339 
340```python
341# core/security.py
342from datetime import datetime, timedelta
343from typing import Optional
344from jose import JWTError, jwt
345from passlib.context import CryptContext
346from app.core.config import get_settings
347 
348settings = get_settings()
349pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
350 
351ALGORITHM = "HS256"
352 
353def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
354    """Create JWT access token."""
355    to_encode = data.copy()
356    if expires_delta:
357        expire = datetime.utcnow() + expires_delta
358    else:
359        expire = datetime.utcnow() + timedelta(minutes=15)
360    to_encode.update({"exp": expire})
361    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
362    return encoded_jwt
363 
364def verify_password(plain_password: str, hashed_password: str) -> bool:
365    """Verify password against hash."""
366    return pwd_context.verify(plain_password, hashed_password)
367 
368def get_password_hash(password: str) -> str:
369    """Hash password."""
370    return pwd_context.hash(password)
371 
372# api/dependencies.py
373from fastapi import Depends, HTTPException, status
374from fastapi.security import OAuth2PasswordBearer
375from jose import JWTError, jwt
376from sqlalchemy.ext.asyncio import AsyncSession
377 
378from app.core.database import get_db
379from app.core.security import ALGORITHM
380from app.core.config import get_settings
381from app.repositories.user_repository import user_repository
382 
383oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
384 
385async def get_current_user(
386    db: AsyncSession = Depends(get_db),
387    token: str = Depends(oauth2_scheme)
388):
389    """Get current authenticated user."""
390    credentials_exception = HTTPException(
391        status_code=status.HTTP_401_UNAUTHORIZED,
392        detail="Could not validate credentials",
393        headers={"WWW-Authenticate": "Bearer"},
394    )
395 
396    try:
397        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
398        user_id: int = payload.get("sub")
399        if user_id is None:
400            raise credentials_exception
401    except JWTError:
402        raise credentials_exception
403 
404    user = await user_repository.get(db, user_id)
405    if user is None:
406        raise credentials_exception
407 
408    return user
409```
410