--- name: fastapi-codegen description: | 트리거: "fastapi 만들어줘", "fastapi codegen", "fastapi 코드 생성", "fastapi 라우터 생성", "fastapi api 만들어줘" 수행: Python FastAPI 프로젝트의 Router·Pydantic Schema·Service·Dependency 계층을 자동 생성한다. async/await 패턴, 전역 에러 핸들러, 표준 응답 모델(JSONResponse), 타입 힌트를 완전히 적용한다. 출력: 디렉토리 구조 안내 + 각 파일 코드 블록 + 실행 방법 안내. --- # FastAPI 계층 코드 생성기 ## 목적 리소스 이름과 필드 정의만 입력하면 FastAPI 표준 계층(Schema → Service → Router → Dependencies) 전체를 즉시 생성한다. Pydantic v2 기반 검증, async 패턴, 의존성 주입을 일관되게 적용한다. ## 실행 절차 1. **입력 파악**: 리소스 이름(예: `user`, `product`), 필드 목록(이름·타입·제약), DB 연동 여부 확인 2. **프로젝트 구조 결정**: `app/api/v1/`, `app/schemas/`, `app/services/`, `app/models/`, `app/dependencies/` 매핑 3. **Pydantic Schema 생성**: Base → Create → Update → Response 스키마 계층 구조 적용 4. **SQLAlchemy 모델 생성**: DB 연동 요청 시 async SQLAlchemy 모델 생성 5. **Service 레이어 생성**: 비즈니스 로직 분리, async 메서드, 예외 처리 포함 6. **Dependencies 생성**: DB 세션, 인증 토큰, 공통 쿼리 파라미터 의존성 함수 7. **Router 생성**: APIRouter, 경로 파라미터, 응답 모델, 상태 코드 명시 8. **에러 핸들러 생성**: HTTPException 커스텀 핸들러, 전역 예외 처리 9. **main.py 통합 예시**: 앱 초기화, 라우터 등록, CORS 설정 포함 ## 출력 형식 ### 프로젝트 구조 ``` app/ ├── main.py ├── api/ │ └── v1/ │ ├── __init__.py │ └── {resource}.py # Router ├── schemas/ │ └── {resource}.py # Pydantic schemas ├── services/ │ └── {resource}_service.py # Business logic ├── models/ │ └── {resource}.py # SQLAlchemy model ├── dependencies/ │ ├── database.py # DB session │ └── auth.py # Auth dependency └── core/ ├── config.py └── exceptions.py ``` ### 파일별 코드 블록 출력 후 실행 안내 ```bash # 의존성 설치 pip install fastapi uvicorn sqlalchemy asyncpg pydantic-settings # 실행 uvicorn app.main:app --reload ``` ## 사용 예시 ### 입력 ``` User 리소스 FastAPI 코드 만들어줘. 필드: id(int, PK), username(str, unique), email(EmailStr), hashed_password(str), is_active(bool, default=True), created_at(datetime) DB: PostgreSQL (async SQLAlchemy) ``` ### 출력 (Schema 예시) ```python # app/schemas/user.py from pydantic import BaseModel, EmailStr, ConfigDict from datetime import datetime from typing import Optional class UserBase(BaseModel): username: str email: EmailStr class UserCreate(UserBase): password: str # 평문 비밀번호 (서비스에서 해싱) class UserUpdate(BaseModel): username: Optional[str] = None email: Optional[EmailStr] = None is_active: Optional[bool] = None class UserResponse(UserBase): id: int is_active: bool created_at: datetime model_config = ConfigDict(from_attributes=True) class UserListResponse(BaseModel): total: int items: list[UserResponse] ``` ### 출력 (Model 예시) ```python # app/models/user.py from sqlalchemy import Column, Integer, String, Boolean, DateTime, func from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, nullable=False, index=True) email = Column(String(100), unique=True, nullable=False, index=True) hashed_password = Column(String(255), nullable=False) is_active = Column(Boolean, default=True, nullable=False) created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False) ``` ### 출력 (Service 예시) ```python # app/services/user_service.py from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from fastapi import HTTPException, status from passlib.context import CryptContext from app.models.user import User from app.schemas.user import UserCreate, UserUpdate, UserListResponse, UserResponse pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") class UserService: @staticmethod async def get_user(db: AsyncSession, user_id: int) -> User: result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"사용자를 찾을 수 없습니다: {user_id}" ) return user @staticmethod async def get_users(db: AsyncSession, skip: int = 0, limit: int = 20) -> UserListResponse: count_result = await db.execute(select(func.count(User.id))) total = count_result.scalar() result = await db.execute(select(User).offset(skip).limit(limit)) users = result.scalars().all() return UserListResponse(total=total, items=[UserResponse.model_validate(u) for u in users]) @staticmethod async def create_user(db: AsyncSession, user_create: UserCreate) -> User: # 중복 이메일 확인 result = await db.execute(select(User).where(User.email == user_create.email)) if result.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="이미 사용 중인 이메일입니다" ) hashed_password = pwd_context.hash(user_create.password) db_user = User( username=user_create.username, email=user_create.email, hashed_password=hashed_password, ) db.add(db_user) await db.commit() await db.refresh(db_user) return db_user @staticmethod async def update_user(db: AsyncSession, user_id: int, user_update: UserUpdate) -> User: user = await UserService.get_user(db, user_id) update_data = user_update.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(user, field, value) await db.commit() await db.refresh(user) return user @staticmethod async def delete_user(db: AsyncSession, user_id: int) -> None: user = await UserService.get_user(db, user_id) await db.delete(user) await db.commit() ``` ### 출력 (Dependencies 예시) ```python # app/dependencies/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from typing import AsyncGenerator DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname" engine = create_async_engine(DATABASE_URL, echo=True) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def get_db() -> AsyncGenerator[AsyncSession, None]: async with AsyncSessionLocal() as session: try: yield session except Exception: await session.rollback() raise ``` ### 출력 (Router 예시) ```python # app/api/v1/user.py from fastapi import APIRouter, Depends, status from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies.database import get_db from app.schemas.user import UserCreate, UserUpdate, UserResponse, UserListResponse from app.services.user_service import UserService router = APIRouter(prefix="/users", tags=["users"]) @router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user(user_create: UserCreate, db: AsyncSession = Depends(get_db)): return await UserService.create_user(db, user_create) @router.get("/", response_model=UserListResponse) async def list_users( skip: int = 0, limit: int = 20, db: AsyncSession = Depends(get_db) ): return await UserService.get_users(db, skip, limit) @router.get("/{user_id}", response_model=UserResponse) async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): return await UserService.get_user(db, user_id) @router.put("/{user_id}", response_model=UserResponse) async def update_user( user_id: int, user_update: UserUpdate, db: AsyncSession = Depends(get_db) ): return await UserService.update_user(db, user_id, user_update) @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)): await UserService.delete_user(db, user_id) ``` ## 주의사항 - **Pydantic v1 vs v2**: v2는 `model_config = ConfigDict(from_attributes=True)`, v1은 `class Config: orm_mode = True`. 버전 명시 없으면 v2 기준 생성. - **async 일관성**: async 함수 내부에서 동기 DB 호출 혼용 금지. `AsyncSession`과 `await` 일관 적용. - **비밀번호 처리**: `password` 필드는 스키마에서 `write_only` 처리, Response 스키마에 절대 포함 금지. - **페이지네이션**: 대량 데이터 조회는 반드시 `skip`/`limit` 또는 cursor 기반 페이지네이션 적용. - **N+1 문제**: 연관 관계 조회 시 `selectinload` 또는 `joinedload` 사용 권장. - **환경변수**: DB URL 등 민감 정보는 `pydantic-settings`의 `BaseSettings`로 관리.