1---
2name: fastapi-templates
3description: Create production-ready FastAPI projects with async patterns, dependency injection, and comprehensive error handling. Use when building new FastAPI applications or setting up backend API projects.
4---
5 
6# FastAPI Project Templates
7 
8Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs.
9 
10## When to Use This Skill
11 
12- Starting new FastAPI projects from scratch
13- Implementing async REST APIs with Python
14- Building high-performance web services and microservices
15- Creating async applications with PostgreSQL, MongoDB
16- Setting up API projects with proper structure and testing
17 
18## Core Concepts
19 
20### 1. Project Structure
21 
22**Recommended Layout:**
23 
24```
25app/
26├── api/                    # API routes
27│   ├── v1/
28│   │   ├── endpoints/
29│   │   │   ├── users.py
30│   │   │   ├── auth.py
31│   │   │   └── items.py
32│   │   └── router.py
33│   └── dependencies.py     # Shared dependencies
34├── core/                   # Core configuration
35│   ├── config.py
36│   ├── security.py
37│   └── database.py
38├── models/                 # Database models
39│   ├── user.py
40│   └── item.py
41├── schemas/                # Pydantic schemas
42│   ├── user.py
43│   └── item.py
44├── services/               # Business logic
45│   ├── user_service.py
46│   └── auth_service.py
47├── repositories/           # Data access
48│   ├── user_repository.py
49│   └── item_repository.py
50└── main.py                 # Application entry
51```
52 
53### 2. Dependency Injection
54 
55FastAPI's built-in DI system using `Depends`:
56 
57- Database session management
58- Authentication/authorization
59- Shared business logic
60- Configuration injection
61 
62### 3. Async Patterns
63 
64Proper async/await usage:
65 
66- Async route handlers
67- Async database operations
68- Async background tasks
69- Async middleware
70 
71## Implementation Patterns
72 
73### Pattern 1: Complete FastAPI Application
74 
75```python
76# main.py
77from fastapi import FastAPI, Depends
78from fastapi.middleware.cors import CORSMiddleware
79from contextlib import asynccontextmanager
80 
81@asynccontextmanager
82async def lifespan(app: FastAPI):
83    """Application lifespan events."""
84    # Startup
85    await database.connect()
86    yield
87    # Shutdown
88    await database.disconnect()
89 
90app = FastAPI(
91    title="API Template",
92    version="1.0.0",
93    lifespan=lifespan
94)
95 
96# CORS middleware
97app.add_middleware(
98    CORSMiddleware,
99    allow_origins=["*"],
100    allow_credentials=True,
101    allow_methods=["*"],
102    allow_headers=["*"],
103)
104 
105# Include routers
106from app.api.v1.router import api_router
107app.include_router(api_router, prefix="/api/v1")
108 
109# core/config.py
110from pydantic_settings import BaseSettings
111from functools import lru_cache
112 
113class Settings(BaseSettings):
114    """Application settings."""
115    DATABASE_URL: str
116    SECRET_KEY: str
117    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
118    API_V1_STR: str = "/api/v1"
119 
120    class Config:
121        env_file = ".env"
122 
123@lru_cache()
124def get_settings() -> Settings:
125    return Settings()
126 
127# core/database.py
128from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
129from sqlalchemy.ext.declarative import declarative_base
130from sqlalchemy.orm import sessionmaker
131from app.core.config import get_settings
132 
133settings = get_settings()
134 
135engine = create_async_engine(
136    settings.DATABASE_URL,
137    echo=True,
138    future=True
139)
140 
141AsyncSessionLocal = sessionmaker(
142    engine,
143    class_=AsyncSession,
144    expire_on_commit=False
145)
146 
147Base = declarative_base()
148 
149async def get_db() -> AsyncSession:
150    """Dependency for database session."""
151    async with AsyncSessionLocal() as session:
152        try:
153            yield session
154            await session.commit()
155        except Exception:
156            await session.rollback()
157            raise
158        finally:
159            await session.close()
160```
161 
162### Pattern 2: CRUD Repository Pattern
163 
164```python
165# repositories/base_repository.py
166from typing import Generic, TypeVar, Type, Optional, List
167from sqlalchemy.ext.asyncio import AsyncSession
168from sqlalchemy import select
169from pydantic import BaseModel
170 
171ModelType = TypeVar("ModelType")
172CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
173UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
174 
175class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
176    """Base repository for CRUD operations."""
177 
178    def __init__(self, model: Type[ModelType]):
179        self.model = model
180 
181    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
182        """Get by ID."""
183        result = await db.execute(
184            select(self.model).where(self.model.id == id)
185        )
186        return result.scalars().first()
187 
188    async def get_multi(
189        self,
190        db: AsyncSession,
191        skip: int = 0,
192        limit: int = 100
193    ) -> List[ModelType]:
194        """Get multiple records."""
195        result = await db.execute(
196            select(self.model).offset(skip).limit(limit)
197        )
198        return result.scalars().all()
199 
200    async def create(
201        self,
202        db: AsyncSession,
203        obj_in: CreateSchemaType
204    ) -> ModelType:
205        """Create new record."""
206        db_obj = self.model(**obj_in.dict())
207        db.add(db_obj)
208        await db.flush()
209        await db.refresh(db_obj)
210        return db_obj
211 
212    async def update(
213        self,
214        db: AsyncSession,
215        db_obj: ModelType,
216        obj_in: UpdateSchemaType
217    ) -> ModelType:
218        """Update record."""
219        update_data = obj_in.dict(exclude_unset=True)
220        for field, value in update_data.items():
221            setattr(db_obj, field, value)
222        await db.flush()
223        await db.refresh(db_obj)
224        return db_obj
225 
226    async def delete(self, db: AsyncSession, id: int) -> bool:
227        """Delete record."""
228        obj = await self.get(db, id)
229        if obj:
230            await db.delete(obj)
231            return True
232        return False
233 
234# repositories/user_repository.py
235from app.repositories.base_repository import BaseRepository
236from app.models.user import User
237from app.schemas.user import UserCreate, UserUpdate
238 
239class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):
240    """User-specific repository."""
241 
242    async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:
243        """Get user by email."""
244        result = await db.execute(
245            select(User).where(User.email == email)
246        )
247        return result.scalars().first()
248 
249    async def is_active(self, db: AsyncSession, user_id: int) -> bool:
250        """Check if user is active."""
251        user = await self.get(db, user_id)
252        return user.is_active if user else False
253 
254user_repository = UserRepository(User)
255```
256 
257### Pattern 3: Service Layer
258 
259```python
260# services/user_service.py
261from typing import Optional
262from sqlalchemy.ext.asyncio import AsyncSession
263from app.repositories.user_repository import user_repository
264from app.schemas.user import UserCreate, UserUpdate, User
265from app.core.security import get_password_hash, verify_password
266 
267class UserService:
268    """Business logic for users."""
269 
270    def __init__(self):
271        self.repository = user_repository
272 
273    async def create_user(
274        self,
275        db: AsyncSession,
276        user_in: UserCreate
277    ) -> User:
278        """Create new user with hashed password."""
279        # Check if email exists
280        existing = await self.repository.get_by_email(db, user_in.email)
281        if existing:
282            raise ValueError("Email already registered")
283 
284        # Hash password
285        user_in_dict = user_in.dict()
286        user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))
287 
288        # Create user
289        user = await self.repository.create(db, UserCreate(**user_in_dict))
290        return user
291 
292    async def authenticate(
293        self,
294        db: AsyncSession,
295        email: str,
296        password: str
297    ) -> Optional[User]:
298        """Authenticate user."""
299        user = await self.repository.get_by_email(db, email)
300        if not user:
301            return None
302        if not verify_password(password, user.hashed_password):
303            return None
304        return user
305 
306    async def update_user(
307        self,
308        db: AsyncSession,
309        user_id: int,
310        user_in: UserUpdate
311    ) -> Optional[User]:
312        """Update user."""
313        user = await self.repository.get(db, user_id)
314        if not user:
315            return None
316 
317        if user_in.password:
318            user_in_dict = user_in.dict(exclude_unset=True)
319            user_in_dict["hashed_password"] = get_password_hash(
320                user_in_dict.pop("password")
321            )
322            user_in = UserUpdate(**user_in_dict)
323 
324        return await self.repository.update(db, user, user_in)
325 
326user_service = UserService()
327```
328 
329### Pattern 4: API Endpoints with Dependencies
330 
331```python
332# api/v1/endpoints/users.py
333from fastapi import APIRouter, Depends, HTTPException, status
334from sqlalchemy.ext.asyncio import AsyncSession
335from typing import List
336 
337from app.core.database import get_db
338from app.schemas.user import User, UserCreate, UserUpdate
339from app.services.user_service import user_service
340from app.api.dependencies import get_current_user
341 
342router = APIRouter()
343 
344@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
345async def create_user(
346    user_in: UserCreate,
347    db: AsyncSession = Depends(get_db)
348):
349    """Create new user."""
350    try:
351        user = await user_service.create_user(db, user_in)
352        return user
353    except ValueError as e:
354        raise HTTPException(status_code=400, detail=str(e))
355 
356@router.get("/me", response_model=User)
357async def read_current_user(
358    current_user: User = Depends(get_current_user)
359):
360    """Get current user."""
361    return current_user
362 
363@router.get("/{user_id}", response_model=User)
364async def read_user(
365    user_id: int,
366    db: AsyncSession = Depends(get_db),
367    current_user: User = Depends(get_current_user)
368):
369    """Get user by ID."""
370    user = await user_service.repository.get(db, user_id)
371    if not user:
372        raise HTTPException(status_code=404, detail="User not found")
373    return user
374 
375@router.patch("/{user_id}", response_model=User)
376async def update_user(
377    user_id: int,
378    user_in: UserUpdate,
379    db: AsyncSession = Depends(get_db),
380    current_user: User = Depends(get_current_user)
381):
382    """Update user."""
383    if current_user.id != user_id:
384        raise HTTPException(status_code=403, detail="Not authorized")
385 
386    user = await user_service.update_user(db, user_id, user_in)
387    if not user:
388        raise HTTPException(status_code=404, detail="User not found")
389    return user
390 
391@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
392async def delete_user(
393    user_id: int,
394    db: AsyncSession = Depends(get_db),
395    current_user: User = Depends(get_current_user)
396):
397    """Delete user."""
398    if current_user.id != user_id:
399        raise HTTPException(status_code=403, detail="Not authorized")
400 
401    deleted = await user_service.repository.delete(db, user_id)
402    if not deleted:
403        raise HTTPException(status_code=404, detail="User not found")
404```
405 
406### Pattern 5: Authentication & Authorization
407 
408```python
409# core/security.py
410from datetime import datetime, timedelta
411from typing import Optional
412from jose import JWTError, jwt
413from passlib.context import CryptContext
414from app.core.config import get_settings
415 
416settings = get_settings()
417pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
418 
419ALGORITHM = "HS256"
420 
421def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
422    """Create JWT access token."""
423    to_encode = data.copy()
424    if expires_delta:
425        expire = datetime.utcnow() + expires_delta
426    else:
427        expire = datetime.utcnow() + timedelta(minutes=15)
428    to_encode.update({"exp": expire})
429    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
430    return encoded_jwt
431 
432def verify_password(plain_password: str, hashed_password: str) -> bool:
433    """Verify password against hash."""
434    return pwd_context.verify(plain_password, hashed_password)
435 
436def get_password_hash(password: str) -> str:
437    """Hash password."""
438    return pwd_context.hash(password)
439 
440# api/dependencies.py
441from fastapi import Depends, HTTPException, status
442from fastapi.security import OAuth2PasswordBearer
443from jose import JWTError, jwt
444from sqlalchemy.ext.asyncio import AsyncSession
445 
446from app.core.database import get_db
447from app.core.security import ALGORITHM
448from app.core.config import get_settings
449from app.repositories.user_repository import user_repository
450 
451oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
452 
453async def get_current_user(
454    db: AsyncSession = Depends(get_db),
455    token: str = Depends(oauth2_scheme)
456):
457    """Get current authenticated user."""
458    credentials_exception = HTTPException(
459        status_code=status.HTTP_401_UNAUTHORIZED,
460        detail="Could not validate credentials",
461        headers={"WWW-Authenticate": "Bearer"},
462    )
463 
464    try:
465        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
466        user_id: int = payload.get("sub")
467        if user_id is None:
468            raise credentials_exception
469    except JWTError:
470        raise credentials_exception
471 
472    user = await user_repository.get(db, user_id)
473    if user is None:
474        raise credentials_exception
475 
476    return user
477```
478 
479## Testing
480 
481```python
482# tests/conftest.py
483import pytest
484import asyncio
485from httpx import AsyncClient
486from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
487from sqlalchemy.orm import sessionmaker
488 
489from app.main import app
490from app.core.database import get_db, Base
491 
492TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
493 
494@pytest.fixture(scope="session")
495def event_loop():
496    loop = asyncio.get_event_loop_policy().new_event_loop()
497    yield loop
498    loop.close()
499 
500@pytest.fixture
501async def db_session():
502    engine = create_async_engine(TEST_DATABASE_URL, echo=True)
503    async with engine.begin() as conn:
504        await conn.run_sync(Base.metadata.create_all)
505 
506    AsyncSessionLocal = sessionmaker(
507        engine, class_=AsyncSession, expire_on_commit=False
508    )
509 
510    async with AsyncSessionLocal() as session:
511        yield session
512 
513@pytest.fixture
514async def client(db_session):
515    async def override_get_db():
516        yield db_session
517 
518    app.dependency_overrides[get_db] = override_get_db
519 
520    async with AsyncClient(app=app, base_url="http://test") as client:
521        yield client
522 
523# tests/test_users.py
524import pytest
525 
526@pytest.mark.asyncio
527async def test_create_user(client):
528    response = await client.post(
529        "/api/v1/users/",
530        json={
531            "email": "[email protected]",
532            "password": "testpass123",
533            "name": "Test User"
534        }
535    )
536    assert response.status_code == 201
537    data = response.json()
538    assert data["email"] == "[email protected]"
539    assert "id" in data
540```
541