PaperclipOrg
Claude Agent Skill · by Wshobson

Fastapi Templates

The fastapi-templates skill provides production-ready project structures and patterns for building FastAPI applications with async operations, dependency inject

Install
Terminal · npx
$npx skills add https://github.com/wshobson/agents --skill fastapi-templates
Works with Paperclip

How Fastapi Templates fits into a Paperclip company.

Fastapi Templates drops into any Paperclip agent that handles this kind of work. Assign it to a specialist inside a pre-configured PaperclipOrg company and the skill becomes available on every heartbeat — no prompt engineering, no tool wiring.

S
SaaS FactoryPaired

Pre-configured AI company — 18 agents, 18 skills, one-time purchase.

$27$59
Explore pack
Source file
SKILL.md540 lines
Expand
---name: fastapi-templatesdescription: 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.--- # FastAPI Project Templates Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs. ## When to Use This Skill - Starting new FastAPI projects from scratch- Implementing async REST APIs with Python- Building high-performance web services and microservices- Creating async applications with PostgreSQL, MongoDB- Setting up API projects with proper structure and testing ## Core Concepts ### 1. Project Structure **Recommended Layout:** ```app/├── api/                    # API routes│   ├── v1/│   │   ├── endpoints/│   │   │   ├── users.py│   │   │   ├── auth.py│   │   │   └── items.py│   │   └── router.py│   └── dependencies.py     # Shared dependencies├── core/                   # Core configuration│   ├── config.py│   ├── security.py│   └── database.py├── models/                 # Database models│   ├── user.py│   └── item.py├── schemas/                # Pydantic schemas│   ├── user.py│   └── item.py├── services/               # Business logic│   ├── user_service.py│   └── auth_service.py├── repositories/           # Data access│   ├── user_repository.py│   └── item_repository.py└── main.py                 # Application entry``` ### 2. Dependency Injection FastAPI's built-in DI system using `Depends`: - Database session management- Authentication/authorization- Shared business logic- Configuration injection ### 3. Async Patterns Proper async/await usage: - Async route handlers- Async database operations- Async background tasks- Async middleware ## Implementation Patterns ### Pattern 1: Complete FastAPI Application ```python# main.pyfrom fastapi import FastAPI, Dependsfrom fastapi.middleware.cors import CORSMiddlewarefrom contextlib import asynccontextmanager @asynccontextmanagerasync def lifespan(app: FastAPI):    """Application lifespan events."""    # Startup    await database.connect()    yield    # Shutdown    await database.disconnect() app = FastAPI(    title="API Template",    version="1.0.0",    lifespan=lifespan) # CORS middlewareapp.add_middleware(    CORSMiddleware,    allow_origins=["*"],    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],) # Include routersfrom app.api.v1.router import api_routerapp.include_router(api_router, prefix="/api/v1") # core/config.pyfrom pydantic_settings import BaseSettingsfrom functools import lru_cache class Settings(BaseSettings):    """Application settings."""    DATABASE_URL: str    SECRET_KEY: str    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30    API_V1_STR: str = "/api/v1"     class Config:        env_file = ".env" @lru_cache()def get_settings() -> Settings:    return Settings() # core/database.pyfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom app.core.config import get_settings settings = get_settings() engine = create_async_engine(    settings.DATABASE_URL,    echo=True,    future=True) AsyncSessionLocal = sessionmaker(    engine,    class_=AsyncSession,    expire_on_commit=False) Base = declarative_base() async def get_db() -> AsyncSession:    """Dependency for database session."""    async with AsyncSessionLocal() as session:        try:            yield session            await session.commit()        except Exception:            await session.rollback()            raise        finally:            await session.close()``` ### Pattern 2: CRUD Repository Pattern ```python# repositories/base_repository.pyfrom typing import Generic, TypeVar, Type, Optional, Listfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy import selectfrom pydantic import BaseModel ModelType = TypeVar("ModelType")CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):    """Base repository for CRUD operations."""     def __init__(self, model: Type[ModelType]):        self.model = model     async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:        """Get by ID."""        result = await db.execute(            select(self.model).where(self.model.id == id)        )        return result.scalars().first()     async def get_multi(        self,        db: AsyncSession,        skip: int = 0,        limit: int = 100    ) -> List[ModelType]:        """Get multiple records."""        result = await db.execute(            select(self.model).offset(skip).limit(limit)        )        return result.scalars().all()     async def create(        self,        db: AsyncSession,        obj_in: CreateSchemaType    ) -> ModelType:        """Create new record."""        db_obj = self.model(**obj_in.dict())        db.add(db_obj)        await db.flush()        await db.refresh(db_obj)        return db_obj     async def update(        self,        db: AsyncSession,        db_obj: ModelType,        obj_in: UpdateSchemaType    ) -> ModelType:        """Update record."""        update_data = obj_in.dict(exclude_unset=True)        for field, value in update_data.items():            setattr(db_obj, field, value)        await db.flush()        await db.refresh(db_obj)        return db_obj     async def delete(self, db: AsyncSession, id: int) -> bool:        """Delete record."""        obj = await self.get(db, id)        if obj:            await db.delete(obj)            return True        return False # repositories/user_repository.pyfrom app.repositories.base_repository import BaseRepositoryfrom app.models.user import Userfrom app.schemas.user import UserCreate, UserUpdate class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):    """User-specific repository."""     async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:        """Get user by email."""        result = await db.execute(            select(User).where(User.email == email)        )        return result.scalars().first()     async def is_active(self, db: AsyncSession, user_id: int) -> bool:        """Check if user is active."""        user = await self.get(db, user_id)        return user.is_active if user else False user_repository = UserRepository(User)``` ### Pattern 3: Service Layer ```python# services/user_service.pyfrom typing import Optionalfrom sqlalchemy.ext.asyncio import AsyncSessionfrom app.repositories.user_repository import user_repositoryfrom app.schemas.user import UserCreate, UserUpdate, Userfrom app.core.security import get_password_hash, verify_password class UserService:    """Business logic for users."""     def __init__(self):        self.repository = user_repository     async def create_user(        self,        db: AsyncSession,        user_in: UserCreate    ) -> User:        """Create new user with hashed password."""        # Check if email exists        existing = await self.repository.get_by_email(db, user_in.email)        if existing:            raise ValueError("Email already registered")         # Hash password        user_in_dict = user_in.dict()        user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))         # Create user        user = await self.repository.create(db, UserCreate(**user_in_dict))        return user     async def authenticate(        self,        db: AsyncSession,        email: str,        password: str    ) -> Optional[User]:        """Authenticate user."""        user = await self.repository.get_by_email(db, email)        if not user:            return None        if not verify_password(password, user.hashed_password):            return None        return user     async def update_user(        self,        db: AsyncSession,        user_id: int,        user_in: UserUpdate    ) -> Optional[User]:        """Update user."""        user = await self.repository.get(db, user_id)        if not user:            return None         if user_in.password:            user_in_dict = user_in.dict(exclude_unset=True)            user_in_dict["hashed_password"] = get_password_hash(                user_in_dict.pop("password")            )            user_in = UserUpdate(**user_in_dict)         return await self.repository.update(db, user, user_in) user_service = UserService()``` ### Pattern 4: API Endpoints with Dependencies ```python# api/v1/endpoints/users.pyfrom fastapi import APIRouter, Depends, HTTPException, statusfrom sqlalchemy.ext.asyncio import AsyncSessionfrom typing import List from app.core.database import get_dbfrom app.schemas.user import User, UserCreate, UserUpdatefrom app.services.user_service import user_servicefrom app.api.dependencies import get_current_user router = APIRouter() @router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)async def create_user(    user_in: UserCreate,    db: AsyncSession = Depends(get_db)):    """Create new user."""    try:        user = await user_service.create_user(db, user_in)        return user    except ValueError as e:        raise HTTPException(status_code=400, detail=str(e)) @router.get("/me", response_model=User)async def read_current_user(    current_user: User = Depends(get_current_user)):    """Get current user."""    return current_user @router.get("/{user_id}", response_model=User)async def read_user(    user_id: int,    db: AsyncSession = Depends(get_db),    current_user: User = Depends(get_current_user)):    """Get user by ID."""    user = await user_service.repository.get(db, user_id)    if not user:        raise HTTPException(status_code=404, detail="User not found")    return user @router.patch("/{user_id}", response_model=User)async def update_user(    user_id: int,    user_in: UserUpdate,    db: AsyncSession = Depends(get_db),    current_user: User = Depends(get_current_user)):    """Update user."""    if current_user.id != user_id:        raise HTTPException(status_code=403, detail="Not authorized")     user = await user_service.update_user(db, user_id, user_in)    if not user:        raise HTTPException(status_code=404, detail="User not found")    return user @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)async def delete_user(    user_id: int,    db: AsyncSession = Depends(get_db),    current_user: User = Depends(get_current_user)):    """Delete user."""    if current_user.id != user_id:        raise HTTPException(status_code=403, detail="Not authorized")     deleted = await user_service.repository.delete(db, user_id)    if not deleted:        raise HTTPException(status_code=404, detail="User not found")``` ### Pattern 5: Authentication & Authorization ```python# core/security.pyfrom datetime import datetime, timedeltafrom typing import Optionalfrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom app.core.config import get_settings settings = get_settings()pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    """Create JWT access token."""    to_encode = data.copy()    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwt def verify_password(plain_password: str, hashed_password: str) -> bool:    """Verify password against hash."""    return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str:    """Hash password."""    return pwd_context.hash(password) # api/dependencies.pyfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom jose import JWTError, jwtfrom sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_dbfrom app.core.security import ALGORITHMfrom app.core.config import get_settingsfrom app.repositories.user_repository import user_repository oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login") async def get_current_user(    db: AsyncSession = Depends(get_db),    token: str = Depends(oauth2_scheme)):    """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, settings.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 user_repository.get(db, user_id)    if user is None:        raise credentials_exception     return user``` ## Testing ```python# tests/conftest.pyimport pytestimport asynciofrom httpx import AsyncClientfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.orm import sessionmaker from app.main import appfrom app.core.database import get_db, Base TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture(scope="session")def event_loop():    loop = asyncio.get_event_loop_policy().new_event_loop()    yield loop    loop.close() @pytest.fixtureasync def db_session():    engine = create_async_engine(TEST_DATABASE_URL, echo=True)    async with engine.begin() as conn:        await conn.run_sync(Base.metadata.create_all)     AsyncSessionLocal = sessionmaker(        engine, class_=AsyncSession, expire_on_commit=False    )     async with AsyncSessionLocal() as session:        yield session @pytest.fixtureasync def client(db_session):    async def override_get_db():        yield db_session     app.dependency_overrides[get_db] = override_get_db     async with AsyncClient(app=app, base_url="http://test") as client:        yield client # tests/test_users.pyimport pytest @pytest.mark.asyncioasync def test_create_user(client):    response = await client.post(        "/api/v1/users/",        json={            "email": "test@example.com",            "password": "testpass123",            "name": "Test User"        }    )    assert response.status_code == 201    data = response.json()    assert data["email"] == "test@example.com"    assert "id" in data```