프로젝트 구조와 기본 설정 - Part 1

이 시리즈에서는 FastAPI의 모범 사례를 기반으로 확장 가능하고 유지보수가 쉬운 애플리케이션을 설계하는 방법을 알아보겠습니다.

프로젝트 구조 설계

첫 번째로 다룰 내용은 프로젝트 구조입니다. 잘 설계된 구조는 프로젝트의 확장성과 유지보수성을 크게 향상시킵니다.

├── app/
│   ├── api/
│   │   ├── v1/
│   │   │   ├── endpoints/
│   │   │   │   ├── user.py
│   │   │   │   └── item.py
│   │   │   └── router.py
│   │   └── deps.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── db/
│   │   ├── session.py
│   │   └── base.py
│   ├── models/
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/
│   │   ├── user.py
│   │   └── item.py
│   ├── services/
│   │   ├── user.py
│   │   └── item.py
│   └── main.py
├── tests/
├── alembic/
├── .env
└── requirements.txt

환경 설정 관리

환경 설정은 Pydantic을 활용하여 타입 안정성을 확보합니다.

# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import List

class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "FastAPI Best Practices"
    BACKEND_CORS_ORIGINS: List[str] = ["http://localhost:3000"]
    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str

    model_config = SettingsConfigDict(
        env_file=".env",
        case_sensitive=True
    )

settings = Settings()

데이터베이스 설정

SQLAlchemy를 사용한 데이터베이스 설정입니다.

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.core.config import settings

DATABASE_URL = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_SERVER}/{settings.POSTGRES_DB}"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    future=True,
    pool_pre_ping=True
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# app/db/base.py
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

의존성 주입 설정

재사용 가능한 의존성들을 관리합니다.

# app/api/deps.py
from typing import AsyncGenerator
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import AsyncSessionLocal
from app.services import user as user_service

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

async def get_current_user(
    session: AsyncSession = Depends(get_db),
    token: str = Depends(oauth2_scheme)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        user = await user_service.get_user_by_token(session, token)
        if user is None:
            raise credentials_exception
        return user
    except Exception:
        raise credentials_exception

모델 정의

SQLAlchemy 모델과 Pydantic 스키마를 분리합니다.

# app/models/user.py
from sqlalchemy import Column, Integer, String
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

# app/schemas/user.py
from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    email: EmailStr

class UserCreate(UserBase):
    password: str

class UserInDB(UserBase):
    id: int
    is_active: bool

    class Config:
        from_attributes = True

서비스 레이어 구현

비즈니스 로직을 서비스 레이어로 분리합니다.

# app/services/user.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.user import User
from app.schemas.user import UserCreate
from app.core.security import get_password_hash

async def create_user(
    session: AsyncSession,
    user_in: UserCreate
) -> User:
    user = User(
        email=user_in.email,
        hashed_password=get_password_hash(user_in.password)
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

async def get_user_by_email(
    session: AsyncSession,
    email: str
) -> User | None:
    result = await session.execute(
        select(User).where(User.email == email)
    )
    return result.scalar_one_or_none()

이것으로 Part 1에서는 FastAPI 프로젝트의 기본 구조와 설정에 대해 알아보았습니다. Part 2에서는 라우터 구현, 에러 처리, 미들웨어 설정 등 더 심화된 내용을 다루도록 하겠습니다.