FastAPI 기본부터 극한까지 날먹하기

서론

일반적인 FastAPI 강의는 지양한다. 그럴거면 ChatGPT에게 물어보는게 낫다.

아키텍처 구조부터 모든 한 줄마다 연구하고 최적의 방향을 서술할 것이다.

환경 구축

Python 버전 관리

파이썬 설치 방법까지는 언급하지 않겠다. 그러나 경우에 따라 Python의 버전 관리가 필요할 수 있다.

Java에서 sdkman으로 JDK 17, 21을 전환하듯, Python에서는 pyenv를 사용한다.

# 설치
curl https://pyenv.run | bash

# ~/.bashrc 또는 ~/.zshrc에 추가
export PYENV_ROOT="$HOME/.pyenv"
export PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"

# Python 버전 설치 및 전환
pyenv install 3.12.4
pyenv install 3.11.9
pyenv global 3.12.4          # 시스템 전체 기본
pyenv local 3.11.9           # 현재 디렉토리에서만 (.python-version 파일 생성)

# 설치된 버전 확인
pyenv versions

프로젝트 초기화

라이브러리 천국 Python 답게 패키지매니저도 다양하다.

• venv + pip (비추)

문제는 requirements.txt에 의존한다는 것.

lock 파일이 아니므로 재현성 보장 안 된다는 의견도 있으나, 보통 pip freeze 쓰니깐 상관 없을 듯

• 개발/프로덕션 의존성 분리 불가
• 의존성 해결이 단순해 충돌 가능

# 디렉토리 생성
mkdir practice-api && cd practice-api

# 가상환경 생성 (= 이 프로젝트 전용 Python 사본 생성)
python3 -m venv .venv

# 활성화
source .venv/bin/activate          # Linux/Mac
# .venv\Scripts\activate           # Windows

# 프롬프트가 바뀜:
# (.venv) user@host:~/practice-api$

# pip 업그레이드 후 패키지 설치
pip install --upgrade pip
pip install fastapi "uvicorn[standard]" sqlalchemy alembic pydantic-settings

# 설치 기록 (= 의존성 스냅샷)
pip freeze > requirements.txt

# 다른 환경에서 복원
pip install -r requirements.txt

• poetry

gradle과 유사한 느낌이 든다.

UV와 성능차이가 궁금해서 조사하다가 아래 블로그를 발견했다. 궁금하면 들어가보시길

https://devocean.sk.com/blog/techBoardDetail.do?ID=167425&boardType=techBlog

Python Poetry 대신 UV를 써보면서 느낀 점들

# Poetry 설치 (공식 방법)
curl -sSL https://install.python-poetry.org | python3 -

# 프로젝트 초기화
mkdir practice-api && cd practice-api
poetry init --no-interaction

# 핵심 의존성 추가 (= implementation 'group:artifact:version')
poetry add fastapi
poetry add "uvicorn[standard]"
poetry add sqlalchemy[asyncio]
poetry add asyncpg                # PostgreSQL async 드라이버
poetry add alembic
poetry add pydantic-settings
poetry add structlog              # 구조화 로깅
poetry add httpx                  # async HTTP 클라이언트
poetry add "redis[hiredis]"       # Redis 클라이언트 (성능 최적화)
poetry add "python-jose[cryptography]"  # JWT

# 개발 의존성 (= testImplementation, developmentOnly)
poetry add --group dev pytest
poetry add --group dev pytest-asyncio
poetry add --group dev pytest-cov
poetry add --group dev httpx         # 테스트용 async HTTP 클라이언트
poetry add --group dev ruff          # 린터 + 포매터 (통합)
poetry add --group dev mypy          # 타입 체커
poetry add --group dev factory-boy   # 테스트 픽스처 팩토리
poetry add --group dev faker         # 테스트 데이터 생성

# 의존성 설치 (= gradle build)
poetry install

# 실행
poetry run uvicorn app.main:app --reload

pyproject.toml

build.gradle 과 유사하다. Node 사용자라면 package-lock 생각해도 될 듯.

[tool.poetry]
name = "practice-api"
version = "0.1.0"
description = "API 연습용 서버"
authors = ["Team Lead <lead@practice.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.12"
fastapi = "^0.115.0"
uvicorn = {extras = ["standard"], version = "^0.30.0"}
sqlalchemy = {extras = ["asyncio"], version = "^2.0"}
asyncpg = "^0.30.0"
alembic = "^1.14"
pydantic-settings = "^2.5"
structlog = "^24.4"
httpx = "^0.27"
redis = {extras = ["hiredis"], version = "^5.0"}
python-jose = {extras = ["cryptography"], version = "^3.3"}

[tool.poetry.group.dev.dependencies]
pytest = "^8.3"
pytest-asyncio = "^0.24"
pytest-cov = "^5.0"
ruff = "^0.6"
mypy = "^1.11"
factory-boy = "^3.3"
faker = "^30.0"

# === 린터/포매터 설정 (= Checkstyle 설정) ===
[tool.ruff]
target-version = "py312"
line-length = 100

[tool.ruff.lint]
select = [
    "E",    # pycodestyle errors
    "W",    # pycodestyle warnings
    "F",    # pyflakes
    "I",    # isort (import 정렬)
    "N",    # pep8-naming
    "UP",   # pyupgrade
    "B",    # flake8-bugbear
    "SIM",  # flake8-simplify
    "ASYNC",# flake8-async
]

# === 타입 체커 설정 (= Java 컴파일러의 타입 검사에 해당) ===
[tool.mypy]
python_version = "3.12"
strict = true                    # 가장 엄격한 모드
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true     # 타입 힌트 없는 함수 금지

# === 테스트 설정 ===
[tool.pytest.ini_options]
asyncio_mode = "auto"            # async 테스트 자동 감지
testpaths = ["tests"]
addopts = "-v --tb=short"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

• uv (매우 추천 & 최신 표준)

uv (차세대 표준)

uv는 Rust로 작성된 초고속 패키지 매니저이다.

pip 대비 10~100배 빠르다고 알려져있어서 Poetry 대안으로 요즘 뜨고있다.

 

세상 깔끔하다.

주의!
현재 app 폴더를 만들지 않았으므로, 실행 시 발생하는 에러는 정상입니다.
  uv run uvicorn app.main:app --reload     →   에러

# 설치
curl -LsSf https://astral.sh/uv/install.sh | sh

# 프로젝트 초기화
uv init practice-api
cd practice-api

# 의존성 추가 (lock 파일 자동 생성)
uv add fastapi "uvicorn[standard]" sqlalchemy asyncpg
uv add --dev pytest ruff mypy

# 실행 (가상환경 자동 생성/관리)
uv run uvicorn app.main:app --reload

# 동기화 (= poetry install)
uv sync

개발 환경 표준화 (생략해도 좋음)

그냥 통상적인 컨벤션이다. 필요 시 더보기를 통해 확인하면 된다.

.editorconfig (= 팀 코딩 스타일 통일)

root = true

[*]
indent_style = space
indent_size = 4
end_of_line = lf
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true

[*.{yml,yaml,json}]
indent_size = 2

pre-commit (= Git 컨벤션)

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.6.0
    hooks:
      - id: ruff          # 린트
        args: [--fix]
      - id: ruff-format   # 포맷

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.11.0
    hooks:
      - id: mypy
        additional_dependencies: [pydantic, sqlalchemy[mypy]]

pip install pre-commit
pre-commit install
# 이후 모든 git commit 시 자동으로 린트/포맷/타입체크 실행

VS Code 설정

// .vscode/settings.json
{
    "python.defaultInterpreterPath": ".venv/bin/python",
    "python.analysis.typeCheckingMode": "strict",
    "[python]": {
        "editor.defaultFormatter": "charliermarsh.ruff",
        "editor.formatOnSave": true,
        "editor.codeActionsOnSave": {
            "source.fixAll.ruff": "explicit",
            "source.organizeImports.ruff": "explicit"
        }
    }
}

Hello World

이제 진짜 시작해보자. 먼저 다음 두 파일을 생성하자.

• app/__init__.py : 빈 파일로 생성한다. (없으면 에러. Python 패키지 선언)
• app/main.py

# app/main.py
from fastapi import FastAPI

app = FastAPI(
    title="FastAPI 연습 API",
    version="0.1.0",
    description="FastAPI 연습용 API 스펙",
)

@app.get("/")
async def root():
    return {"message": "Hello, World"}

@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "practive-api"}

# 실행
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

app.main   → app/ 내의 main.py 파일
:app           → 그 안의 app 변수 (FastAPI 인스턴스)
--reload    → 파일 변경 감지 자동 재시작

브라우저에서 확인해보자. FastAPI는 Swagger가 내장이다.

• API: http://localhost:8000/
• Swagger UI: http://localhost:8000/docs ← 킬러 기능. 별도 라이브러리 불필요
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

FastAPI 핵심 개념

깔끔하게 다음에 대해 설명한다. 이 정도만 알아도 FastAPI의 코드를 이해하는데 문제가 없다.

이 글로 3분만에 개념을 날먹해보자.

• API(라우팅) 작성 방법
• Pydantic
• Depends
• Annotated

라우팅

기본 CRUD 매핑

전혀 어려울게 없다. 기본적인 구조는 다음과 같다.

@router.{post/get/put/patch/delete}(
	스웨거 명세
)
async def 함수명 (인자) -> 결과타입:
	함수 코드

 

초보자용 상세설명은 더보기 클릭. 

가령 다음을 보자.

# GET — 단건 조회 (Path 파라미터)
@router.get(
    "/{payment_id}", 
    response_model=PaymentResponse,
    status_code=status.HTTP_200_OK,
    responses={
        404: {"description": "없는 결제 내역"},
        422: {"description": "요청 데이터 검증 실패"},
    },
)
async def get_payment(
    payment_id: UUID = Path(..., description="결제 고유 ID"),
    db: DBSession,
    user: AuthUser,
) -> PaymentResponse:
    ...(구현할 코드)...

@router.get() 분석

• 첫 번째 인자는 api경로이다. /{payment_id} 
  → 즉, https://서버주소/{payment_id}로 GET 요청하면 접근 가능하다는 것이다.
  → 결제 ID가 1234-5678-9012-3456이라면? https://서버주소/1234-5678-9012-3456 로 접근하면 된다.
  
  
• 두 번째 인자부터는 의미 없다. 그냥 Swagger 전용 Spec 문서이다.
  → 기본 응답값 200, 그 외에는 상황에 따른 응답값

async def get_payment() -> PaymentResponse 분석

• 인자는 payment_id, db, user를 받는다.
  → payment_id 는 사용자로부터 받는다.
  → DBSession과 AuthUser는 사용자에게 받지 않고, FastAPI가 자동으로 주입(DI)한다.
  
  DBSession과 AuthUser가 자동 주입되는 것은 현재 이해되지 않는 것이 당연하다.
  Depends와 Annotated의 조합형태이다. 본문 하단에 자세히 후술하겠다.
  
  
• 결과 형태는 PaymentResponse이다.
  async def get_payment(인자) -> PaymentResponse 형태로 결과 형태를 정의했다.

실전 코드를 보면 이해가 쉬울 것이다.

더보기 없이 아래 코드만 읽어봐도 사용법이 바로 감 잡힐 것이다.

from fastapi import APIRouter, Path, Query, Body, Header, status
from uuid import UUID

router = APIRouter(
    prefix="/api/v1/payments",
    tags=["Payments"],          # Swagger UI에서 그룹핑
    responses={
        401: {"description": "인증 실패"},
        403: {"description": "권한 없음"},
    },
)

# GET — 단건 조회 (Path 파라미터)
@router.get("/{payment_id}", response_model=PaymentResponse)
async def get_payment(
    payment_id: UUID = Path(..., description="결제 고유 ID"),
    db: DBSession,
    user: AuthUser,
) -> PaymentResponse:
    ...

# POST — 생성
@router.post(
    "/",
    status_code=status.HTTP_201_CREATED,
    response_model=PaymentResponse,
    responses={
        409: {"description": "중복 결제 요청"},
        422: {"description": "요청 데이터 검증 실패"},
    },
)
async def create_payment(
    request: CreatePaymentRequest = Body(...),
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
    db: DBSession,
    user: AuthUser,
) -> PaymentResponse:
    ...

와! 당신은 벌써 api를 찍어낼 수 있다!

DB에 데이터 저장하는 방법만 안다면 당신도 지금 해커톤 가능!

(심화 내용 → 지금 이해 안가면 무시하고 넘어가시면 됩니다.)
잠깐! DB 연결을 router(api)에서 인자로 받는다고요?
이러면 api 구현할 때마다 인자에 번거롭게 DBSession을 써야하는데,
실제 로직을 구현할 파일(Service)에서 주입하면 인자도 줄고 좋지 않나요?

충분히 의문이 들 수 있다. 그러나 이는 FastAPI에서 명시성을 위해 의도한 것이다.

 

여러 관점에서 볼 수 있는데, 필자가 느낀 대표적인 이유는 다음과 같다.

• 하위 레이어 독립성
• 생명주기 관점

무슨 뜻인지 상세히 설명하면,

하위 레이어 독립성

(필자가 만든 말이니 검색해도 안나올 수 있다.)

Service에서 아예 FastAPI 의존성을 없앤다는 의미이다. 즉, fastapi를 일체 import하지 않는다.

 서비스가 FastAPI를 모르므로 다른 곳(배치, 스크립트)에서 재사용 가능하다.

 

가령, 스케줄러와 api, 워커 등 다양한 환경을 한 프로젝트에서 구성할 수 있다.

api가 아닌 다른 컨테이너는 fastapi가 필요없다.

 

개인적으로 이렇게 코드베이스 공유하면서 간단하게 쓸 수 있는게 Python의 치트키같다.

같은 이미지(또는 같은 코드베이스)
- api 컨테이너        → uvicorn main:app
- worker 컨테이너     → celery -A tasks worker
- scheduler 컨테이너  → python jobs/daily_cleanup.py
- migration 컨테이너  → alembic upgrade

 

생명주기 관점

Depends를 router에서 받으면 요청마다 세션을 만들고, 응답 후 자동으로 닫아줄 수 있다.

즉, 트랜잭션과 요청의 경계가 동일하다.

잠깐, 그럼 응답 전까지 Connection Pool을 그대로 점유한다고요?

필자가 가장 우려한 의문이다.

영속성 컨텍스트가 너무 오래 살아있는거 아닌가? 세션의 수명을 그대로 끌고가면 트래픽이 몰리는 경우 어떻게 대응하는가?

 

실제로 그렇다. 다만, 커넥션을 View 영역까지 끌고가는 문제부터 말해보면,

Fastapi는 Eventloop 환경 때문에 비동기로 db 세션을 연결한다.

이 때, SQLAlchemy는 기본이 lazy load이므로 비동기에서 문제가 자주 발생한다.

따라서 명시적으로 eager load를 강제하는 경우가 많으므로 뷰에서 N+1 문제가 발생하는 경우는 줄어들긴 한다.

더불어 DTO로 즉시 반환하도록 하면 lazy load가 불가해지므로 대응 가능하다.

 

문제는 앞서 말한 Connection Pool 문제이다.

여러 방법이 있는데 SessionFactory을 주입해서 아예 세션을 Service에서 열고 닫는 방법도 있다.

이러면 응답 직렬화 시점엔 커넥션이 풀로 돌아간다.

# 세션 팩토리만 주입
@router.get("/users/{id}")
async def get_user(id: int, session_factory = Depends(get_session_factory)):
    return await user_service.get(session_factory, id)

# service
async def get(session_factory, id):
    async with session_factory() as session:
        async with session.begin():
            user = await repo.find(session, id)
    # 여기서 이미 세션/커넥션 반납됨
    return UserDTO.from_orm(user)  # DTO 변환은 세션 밖

 

근데 그 정도로 코어 시스템이면 그냥 SpringBoot 쓰는 것도 답인 것 같다.

그럼 이렇게 불편하게 살아요?

필자는 아예 다음과 같이 서비스도 Depends로 묶는다. 꽤 흔한 방식이다.

router의 인자에 DBSession 대신 UserServiceDep이 들어가긴 하나,

Service의 각 메서드에서 불필요한 db session을 입력할 필요가 없다.

def get_user_service(session: AsyncSessionDep) -> UserService:
    """Get UserService instance."""
    return UserService(session)

UserServiceDep = Annotated[UserService, Depends(get_user_service)]

@router.post(
    "/",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
    summary="사용자 생성",
)
async def create_user(
    data: UserCreate,
    service: UserServiceDep,
) -> UserResponse:
    """새로운 사용자 생성"""
    user = await service.create_user(data)
    return UserResponse.model_validate(user)

UserService의 __init__에서 session만 입력받아주면 된다.

class UserService:
    def __init__(self, session: AsyncSession):
        self.repository = UserRepository(session)

    async def create_user(self, data: UserCreate) -> User:
        ...생략...
        return await self.repository.create(user)

 

라우터 등록 패턴

# app/main.py
from app.domain.payment.router import router as payment_router
from app.domain.account.router import router as account_router
from app.domain.transfer.router import router as transfer_router
from app.domain.auth.router import router as auth_router
from app.domain.admin.router import router as admin_router

# 버전별 그룹핑
app.include_router(payment_router, prefix="/api/v1")
app.include_router(account_router, prefix="/api/v1")
app.include_router(transfer_router, prefix="/api/v1")
app.include_router(auth_router, prefix="/api/v1")

# 관리자 API는 별도 prefix
app.include_router(admin_router, prefix="/api/admin")

Pydantic

더 들어가기 전에 기초 정도는 짚고 넘어가자.

Fastapi를 사용하면 Pydantic 라이브러리가 자주 언급된다.

Pydantic은 Python에서 데이터 검증 + 타입 기반 모델링을 자동으로 처리해주는 라이브러리다.

 

Python은 기본적으로 타입이 느슨하다.

def func(age):
    return age + 1

func("10")  # 에러 발생

이를 자동으로 처리해주는 라이브러리라고 보면 된다. 기본적으로 BaseModel을 활용한다.

아래 예시를 보면 age를 string 형태로 넣었으나, 자동으로 정수형으로 인식한 것을 알 수 있다.

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int
    
user = User(name="kim", age="20")
print(user.age)  # 20 (자동 변환됨)

검증도 매우 간편한데 다음과 같은 검증도 지원한다.

gt(Greater Than, 초과), ge(Greater Than or Equal, 이상), lt(Less Than, 미만), le(Less Than or Eqaul, 이하)

from pydantic import BaseModel, Field

class User(BaseModel):
    age: int = Field(gt=0)
    
User(age=-1) # 에러

객체 변환도 간편하다. 딕셔너리 형태를 Unpacking해서 인자로 전달할 수 있다.

JS로 치면 destructing, SpringBoot로 치면 Jackson의 Mapping 느낌?

data = {"name": "kim", "age": 20}

user = User(**data)
print(user.model_dump())  # dict 변환

눈치 챘겠지만 Fastapi의 api 호출에 활용된다.

from fastapi import FastAPI

app = FastAPI()

@app.post("/user")
def create_user(user: User):
    return user

Depends

Depends또한 중요한 핵심 개념으로, 모르면 이해하기 힘들 것이다.

Depends는 DI(의존성 주입)을 진행한다.

 

다음의 경우, get_user에 대한 반환값을 user에 대입해준다.

def get_user():
    return {"name": "Lee"}

@app.get("/")
def read_root(user = Depends(get_user)):
    return user

결과를 넣어준다면 그냥 함수를 호출하면 될텐데? 왜 쓰는걸까?

가장 큰 이유는 생명주기 관리이다. Depends 없이 직접 DB세션을 호출하는 경우를 보자.

참고: Fastapi에서 DB 연결은 비동기로 처리된다.

@router.get("/users/{id}")
async def get_user(id: int):
    session = await get_session()  # 근데 이거 언제 닫지?
    ...
    await session.close()  # 매번 try/finally로 감싸야 함

Depends는 단순히 함수 결과를 주입하는 게 아니라, 컨텍스트 매니저를 관리한다.

async def get_session():
    async with SessionLocal() as session:
        yield session
    # yield 이후: 요청 끝나면 FastAPI가 자동으로 여기로 돌아와 정리

@router.get("/users/{id}")
async def get_user(id: int, session = Depends(get_session)):
    ...  # 예외 터져도 세션 정리됨

또한 의존성을 자동으로 처리하는데,

아래의 경와 같이 요청 내에서 get_session이 여러 번 필요해도 한 번만 호출하고 캐시한다.

async def get_session():
    async with SessionLocal() as session:
        yield session

async def get_user_repo(session = Depends(get_session)):
    return UserRepo(session)

async def get_post_repo(session = Depends(get_session)):
    return PostRepo(session)

async def get_current_user(
    token: str = Header(...),
    session = Depends(get_session),
) -> User:
    return await session.get(User, decode(token))

@router.post("/posts")
async def create_post(
    body: PostCreate,
    user: User = Depends(get_current_user),
    user_repo: UserRepo = Depends(get_user_repo),
    post_repo: PostRepo = Depends(get_post_repo),
):
    ...

 

더불어 get_current_user가 없다면 매 첫 줄마다 다음을 입력했어야 한다. 심지어 OpenAPI 스펙에도 반영되지 않는다.

user = await current_user(request.headers["authorization"])

Annotated

그래서 맨날 귀찮게 이렇게 써야할까?

session: AsyncSession = Depends(get_session)

다음과 같이 사전에 정의해두고 재활용할 수 있다.

SessionDep = Annotated[AsyncSession, Depends(get_session)]

세상 깔끔하다. 이제 이렇게 표현할 수 있다.

SessionDep = Annotated[AsyncSession, Depends(get_session)]
CurrentUser = Annotated[User, Depends(get_current_user)]

# routers/users.py
@router.get("/users/{id}")
async def get_user(id: int, session: SessionDep, user: CurrentUser):
    ...

@router.delete("/users/{id}")
async def delete_user(id: int, session: SessionDep, user: CurrentUser):
    ...

다음을 진행하기 전에

Pydantic, Depends 심화

아래 게시글을 읽지 않아도 된다.

그러나, 실전에서 필히 마주칠 문제이므로 익히고 가면 이후 개발이 훨씬 용이할 것이다.

https://leestana01.tistory.com/25

FastAPI 기본부터 극한까지 날먹하기 (Pydantic, Depends 심화)

SpringBoot 개발자용

SpringBoot 에서 넘어왔다면 다음을 먼저 읽어보자.

Java와 Python 그리고 SpringBoot와 Fastapi를 비교하여 개념을 정리했다.

이것만 읽어도 FastAPI가 훨씬 쉬워질 것이다.

https://leestana01.tistory.com/24

FastAPI 기본부터 극한까지 파보자 (SpringBoot에서 넘어가기)

아키텍처 구조

Spring에서는 프로젝트가 무거워질수록, Facade 패턴이 유용하다.

FastAPI에서는 어떤 아키텍처 구조가 좋을까?

Spring에서 Facade가 자연스러운 이유

Spring Boot에서 Facade 패턴이 자연스럽게 작동하는 데에는 프레임워크의 근본적인 구조가 뒷받침된다.

 

우선, Spring IoC Container가 싱글톤 빈을 관리한다. 무거운 Service 객체들이 애플리케이션 생명주기 동안 한 번만 생성되므로, 여러 Service를 주입받아 조합하는 Facade 클래스가 비용 없이 동작한다.

 

또한 @Transactional도 AOP 프록시로 동작한다. Facade 메서드에 @Transactional을 선언하면 내부에서 호출하는 모든 Repository 작업이 하나의 트랜잭션으로 묶인다. 이 선언적 트랜잭션 관리가 Facade의 오케스트레이션과 궁합이 맞는다.

 

FastAPI에서 Facade가 어색한 이유

Python은 근본적으로 다른 언어이다. 우선, 함수가 일급 객체이다. 

비즈니스 로직 오케스트레이션을 위해 클래스를 만들 필요가 없다.

 

async def execute_transfer(...)라는 함수 하나가 오케스트레이터 역할을 완벽히 수행한다.

또한 Spring의 싱글톤 빈과 달리, FastAPI의 Depends()는 매 요청마다 새 인스턴스를 생성한다.

 

더 중요한건, AOP가 없다. Python에는 @Transactional 같은 것이 없다.

데코레이터로 유사하게 구현할 수 있지만, Spring AOP처럼 프록시 기반이 아니라 함수 래핑이다.

 

모듈 관점에서 바라봐도 Python에서는 파일 하나에 관련 함수를 모아두는 것이 자연스럽다.

Java: 패키지 → 클래스 → 메서드

Python: 패키지 → 모듈(파일) → 함수/클래스

 

Facade가 복잡한 비즈니스 로직을 해결하는 본질적 문제는 FastAPI에서도 당연히 존재한다.

다만 해결 형태가 다르다.

Use Case패턴

Use Case의 원칙

• 하나의 비즈니스 시나리오만 담당
• 도메인 로직은 Domain Service에 위임
• 인프라 호출(DB, 외부 API)은 Repository/Client에 위임
• Use Case 자체는 "흐름 조율"만 담당

취소나 정산은 별도 Use Case.

도메인 별 디렉토리 구조

app/domain/transfer/
├── __init__.py
├── router.py                    # 라우터 (= Controller)
├── models.py                    # ORM 엔티티 (= JPA Entity)
├── schemas.py                   # 요청/응답 DTO (= Request/Response DTO)
├── enums.py                     # 상태 enum
├── repository.py                # 데이터 접근 (= Repository)
├── exceptions.py                # 도메인 예외
├── use_cases/                   # ← Facade 대신 이것
│   ├── __init__.py
│   ├── execute_transfer.py      # 이체 실행
│   ├── cancel_transfer.py       # 이체 취소
│   └── get_transfer_status.py   # 이체 상태 조회 (단순 조회는 함수로도 충분)
└── services/                    # 도메인 서비스 (순수 비즈니스 로직)
    ├── __init__.py
    ├── fee_calculator.py
    └── limit_checker.py

 

UseCase 예시

다음 예제를 통해 execute_transfer.py에 대한 Use Case 를 살펴보자.

통상 Use Case는 Class로 선언한 뒤, Annotated 타입으로 깔끔하게 호출한다.

# app/domain/transfer/use_cases/execute_transfer.py

from dataclasses import dataclass
from sqlalchemy.ext.asyncio import AsyncSession

from app.domain.transfer.schemas import TransferCommand, TransferResult
from app.domain.transfer.services.fee_calculator import FeeCalculator
from app.domain.account.repository import AccountRepository
from app.infrastructure.notification.service import NotificationService

@dataclass(frozen=True)
class ExecuteTransferUseCase:
    """
    이체 실행 UseCase.
    """
    db: AsyncSession
    transfer_repo: TransferRepository
    account_repo: AccountRepository
    fee_calculator: FeeCalculator
    notification: NotificationService

    async def execute(self, command: TransferCommand) -> TransferResult:
        ...이체 시도 로직...
        return TransferResult.from_entity(transfer)


# DI 팩토리
from fastapi import Depends
from app.config.database import get_db

def get_execute_transfer_use_case(
    db: AsyncSession = Depends(get_db),
    ...레포 및 서비스 선언...
) -> ExecuteTransferUseCase:
    return ExecuteTransferUseCase(
        db=db,
        ...레포 및 서비스 초기화...
    )

# Annotated 타입으로 깔끔하게
ExecuteTransfer = Annotated[ExecuteTransferUseCase, Depends(get_execute_transfer_use_case)]

Router에서 사용

# app/domain/transfer/router.py
from fastapi import APIRouter, Depends, Header, status
from app.domain.transfer.schemas import TransferCommand, TransferResult
from app.domain.transfer.use_cases.execute_transfer import ExecuteTransfer
from app.domain.transfer.use_cases.cancel_transfer import CancelTransfer
from app.core.security import AuthUser

router = APIRouter(prefix="/transfers", tags=["Transfers"])

@router.post("/", status_code=status.HTTP_201_CREATED)
async def execute_transfer(
    command: TransferCommand,
    use_case: ExecuteTransfer,
    user: AuthUser,
) -> TransferResult:
    """이체 실행"""
    command.idempotency_key = idempotency_key
    command.sender_user_id = user.user_id
    return await use_case.execute(command)

@router.post("/{transfer_id}/cancel")
async def cancel_transfer(
    transfer_id: UUID,
    use_case: CancelTransfer,
    user: AuthUser,
) -> TransferResult:
    """이체 취소"""
    return await use_case.execute(transfer_id, reason, user.user_id)

단순 CRUD는 함수형으로

모든 것을 Use Case 클래스로 만들 필요는 없다. 단순 조회는 함수로 충분하다.

오히려 모든 것을 Use Case로 만드는건 오버엔지니어링일 수 있다.

# app/domain/transfer/use_cases/get_transfer_status.py

from uuid import UUID
from fastapi import Depends
from app.domain.transfer.repository import TransferRepository
from app.domain.transfer.schemas import TransferResponse
from app.domain.transfer.exceptions import TransferNotFoundError

async def get_transfer_status(
    transfer_id: UUID,
    repo: TransferRepository = Depends(get_transfer_repo),
) -> TransferResponse:
    transfer = await repo.find_by_id(transfer_id)
    if not transfer:
        raise TransferNotFoundError(transfer_id)
    return TransferResponse.model_validate(transfer)

# 라우터에서 함수를 직접 호출하거나, 더 간단하게:
@router.get("/{transfer_id}")
async def get_transfer(
    transfer_id: UUID,
    repo: TransferRepository = Depends(get_transfer_repo),
    user: AuthUser,
) -> TransferResponse:
    transfer = await repo.find_by_id(transfer_id)
    if not transfer:
        raise TransferNotFoundError(transfer_id)
    return TransferResponse.model_validate(transfer)

전체 프로젝트 구조

필자는 다음과 같이 프로젝트 구조를 잡아 확장 가능하도록 설계를 마련했다.

프로젝트/
├── pyproject.toml                 # = build.gradle
├── alembic/                       # = DB 마이그레이션
│   ├── alembic.ini
│   └── versions/
├── app/
│   ├── __init__.py
│   ├── main.py                    # = 진입점
│   ├── config/                    # = 설정 폴더
│   │   ├── __init__.py
│   │   ├── settings.py            # = 기본 세팅
│   │   ├── database.py            # = DataSource 설정
│   │   ├── redis.py               # = RedisConfig
│   │   └── logging.py             # = 로깅 설정
│   │
│   ├── core/                      # = 공통 인프라
│   │   ├── __init__.py
│   │   ├── exceptions.py          # = @ControllerAdvice + 커스텀 예외
│   │   ├── security.py            # = SecurityConfig
│   │   ├── middleware.py          # = Filter/Interceptor
│   │   └── dependencies.py        # = 공통 DI 팩토리
│   │
│   ├── domain/                    # = 도메인 레이어
│   │   ├── 특정도메인/               # = 특정 도메인 패키지
│   │   │   ├── __init__.py
│   │   │   ├── router.py          # = API 스펙
│   │   │   ├── service.py         # = 기본적인 스펙
│   │   │   ├── repository.py      # = DB 연동 스펙
│   │   │   ├── models.py          # = DB 모델(Entity) 구현
│   │   │   ├── schemas.py         # = DTO 구현
│   │   │   ├── enums.py           # = 상태 ENUM
│   │   │   └── exceptions.py      # = DomainNotFoundException 등
│   │   │
│   │   └── user/
│   │       └── ...
│   │
│   └── infrastructure/            # = 외부 시스템 연동
│       ├── __init__.py
│       ├── notification/          # = 알림 서비스
│       │   ├── sms.py
│       │   ├── push.py
│       │   └── email.py
│       ├── pg_gateway/            # = PG사 연동
│       │   └── ...
│       └── cache/
│           └── redis_client.py
│
├── tests/
│   ├── conftest.py                # = 공통 테스트
│   ├── unit/
│   │   ├── ...
│   ├── integration/
│   │   ├── ...
│   └── e2e/
│       └── ...
│
├── docker/                        # 귀찮으면 루트에 둠
│   ├── Dockerfile
│   └── docker-compose.yml
│
├── scripts/
│   └── ...
│
└── docs/
    └── api-spec.md

아키텍처 구조

한 페이지에 다 담으려고 했는데, 게시글이 너무 길다..

아키텍처 각각에 대한 구현 방식은 차후 게시글을 따로 작성하여 여기에 링크하겠다.