FastAPI 섹션 5. 기능 고도화

---

1. SQL JOIN

두 개의 테이블을 공통 컬럼을 기준으로 연결해서 조회하는 기능입니다.

예시

User 테이블:

id	username
1	qu3vipon

ToDo 테이블:

id	user_id	contents
1	1	FastAPI Section 1
2	1	FastAPI Section 2

JOIN 쿼리:

SELECT u.username, t.contents
FROM user u
JOIN todo t ON u.id = t.user_id;

결과:

username	contents
qu3vipon	FastAPI Section 1
qu3vipon	FastAPI Section 2

user.id = todo.user_id 라는 공통 컬럼을 기준으로 두 테이블을 연결합니다.

---

2. User 모델링

ORM 클래스 → SQL 변환 확인 (Python 콘솔)

from sqlalchemy.schema import CreateTable
from database.orm import ToDo
from database.connection import engine

# ORM 클래스가 어떤 SQL CREATE TABLE로 변환되는지 확인
print(CreateTable(ToDo.__table__).compile(engine))

출력 결과:

CREATE TABLE todo (
    id INTEGER NOT NULL AUTO_INCREMENT,
    contents VARCHAR(256) NOT NULL,
    is_done BOOL NOT NULL,
    PRIMARY KEY (id)
)

database/orm.py — User 모델 추가

from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from schema.request import CreateToDoRequest

Base = declarative_base()

class ToDo(Base):
    __tablename__ = "todo"

    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    # ForeignKey: todo.user_id가 user.id를 참조함 (외래키 설정)
    user_id = Column(Integer, ForeignKey("user.id"))

    def __repr__(self):
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"

    @classmethod
    def create(cls, request: CreateToDoRequest) -> "ToDo":
        return cls(
            contents=request.contents,
            is_done=request.is_done,
        )

    def done(self) -> "ToDo":
        self.is_done = True
        return self

    def undone(self) -> "ToDo":
        self.is_done = False
        return self


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), nullable=False)
    password = Column(String(256), nullable=False)

    # relationship: User 객체에서 user.todos로 연관된 ToDo 목록에 바로 접근 가능
    # lazy="joined": Eager Loading 방식 (User 조회 시 ToDo도 함께 JOIN해서 가져옴)
    todos = relationship("ToDo", lazy="joined")

    @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username=username,
            password=hashed_password,
        )

---

3. User 테이블 생성 (MySQL DDL)

-- Docker MySQL 접속
-- docker exec -it todos bash
-- mysql -u root -p

USE todos;

-- User 테이블 생성
CREATE TABLE user (
    id INTEGER NOT NULL AUTO_INCREMENT,
    username VARCHAR(256) NOT NULL,
    password VARCHAR(256) NOT NULL,
    PRIMARY KEY (id)
);

-- todo 테이블에 user_id 컬럼 추가
ALTER TABLE todo ADD COLUMN user_id INTEGER;

-- 외래키 설정 (todo.user_id → user.id)
ALTER TABLE todo ADD FOREIGN KEY(user_id) REFERENCES user(id);

-- 샘플 유저 삽입
INSERT INTO user(username, password) VALUES ("admin", "password");

-- 기존 todo에 user_id 연결
UPDATE todo SET user_id=1 WHERE id=1;
UPDATE todo SET user_id=1 WHERE id=2;
UPDATE todo SET user_id=1 WHERE id=3;

-- JOIN 확인
SELECT u.username, t.contents, t.is_done
FROM todo t
JOIN user u ON t.user_id = u.id;

---

4. Lazy Loading vs Eager Loading

데이터를 언제 가져오는지에 대한 전략입니다.

Lazy Loading (지연 로딩)

연관된 데이터가 실제로 필요한 시점에 조회합니다.

• 장점: 첫 조회 속도가 빠름
• 단점: N+1 문제 발생 가능

N+1 문제 예시:

# todos를 조회하는 쿼리 1번 + 각 todo마다 user 조회 N번 = 총 N+1번 쿼리 실행
for todo in todos:
    print(todo.user.username)  # 이 시점마다 DB 조회 발생

Eager Loading (즉시 로딩)

데이터를 조회할 때 처음부터 연관 객체를 JOIN해서 함께 가져옵니다.

• 장점: N+1 문제 방지, 데이터를 효율적으로 가져옴
• 단점: 필요 없는 데이터까지 JOIN해서 가져올 수 있음

# lazy="joined" → Eager Loading 설정
todos = relationship("ToDo", lazy="joined")

ORM JOIN 동작 확인 (Python 콘솔)

from database.connection import SessionFactory
from database.orm import User
from sqlalchemy import select

session = SessionFactory()
user = session.scalar(select(User))

# user.todos로 연관된 ToDo 목록 바로 접근 (이미 JOIN해서 가져왔으므로 추가 쿼리 없음)
user.todos
# [ToDo(id=1, ...), ToDo(id=2, ...), ToDo(id=3, ...)]

---

5. 회원가입 API

패키지 설치

pip install bcrypt  # 비밀번호 해싱 라이브러리

bcrypt 동작 확인 (Python 콘솔)

import bcrypt

password = "password"
byte_password = password.encode("UTF-8")  # bytes로 변환

# 같은 비밀번호도 매번 다른 해시값이 생성됨 (salt 때문)
hash_1 = bcrypt.hashpw(byte_password, salt=bcrypt.gensalt())
hash_2 = bcrypt.hashpw(byte_password, salt=bcrypt.gensalt())

# 검증은 checkpw로 → 내부적으로 동일한 원문인지 확인
bcrypt.checkpw(byte_password, hash_1)  # True
bcrypt.checkpw(byte_password, hash_2)  # True

bcrypt는 같은 비밀번호도 매번 다른 해시값을 만들기 때문에 단순히 값을 비교하면 안 됩니다. checkpw()로만 검증해야 합니다.

schema/request.py

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool

class SignUpRequest(BaseModel):
    username: str
    password: str

service/user.py

import bcrypt

class UserService:
    encoding: str = "UTF-8"

    def hash_password(self, plain_password: str) -> str:
        # 문자열 → bytes 변환 후 해싱
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()  # 매번 랜덤한 salt 생성
        )
        # bytes → 문자열로 변환해서 반환 (DB에 저장하기 위해)
        return hashed_password.decode(self.encoding)

database/repository.py — UserRepository 추가

class UserRepository:
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session

    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()           # DB 저장
        self.session.refresh(instance=user)  # 자동 생성된 id 반영
        return user

schema/response.py — UserSchema 추가

class UserSchema(BaseModel):
    id: int
    username: str  # password는 보안상 응답에 포함하지 않음

    class Config:
        from_attributes = True

api/user.py — 회원가입 핸들러

from fastapi import APIRouter, Depends
from database.orm import User
from database.repository import UserRepository
from schema.response import UserSchema
from schema.request import SignUpRequest
from service.user import UserService

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
):
    # 1. 요청에서 username, password 받기
    # 2. 비밀번호 해싱
    hashed_password: str = user_service.hash_password(
        plain_password=request.password
    )
    # 3. User 객체 생성 (id=None, DB 저장 전)
    user: User = User.create(
        username=request.username,
        hashed_password=hashed_password,
    )
    # 4. DB에 저장 (id=int, DB 저장 후)
    user: User = user_repo.save_user(user=user)
    # 5. id, username만 반환 (password 제외)
    return UserSchema.from_orm(user)

회원가입 테스트

from service.user import UserService
from database.orm import User
from database.repository import UserRepository

def test_user_sign_up(client, mocker):
    # hash_password를 mock으로 교체 → "hashed" 문자열 반환
    hash_password = mocker.patch.object(
        UserService,
        "hash_password",
        return_value="hashed"
    )
    # User.create를 mock으로 교체 → id=None인 User 반환
    user_create = mocker.patch.object(
        User,
        "create",
        return_value=User(id=None, username="test", password="hashed")
    )
    # save_user를 mock으로 교체 → id=1인 User 반환 (DB 저장 후 상태)
    mocker.patch.object(
        UserRepository,
        "save_user",
        return_value=User(id=1, username="test", password="hashed")
    )

    body = {"username": "test", "password": "plain"}
    response = client.post("/users/sign-up", json=body)

    # hash_password가 plain 비밀번호로 호출됐는지 검증
    hash_password.assert_called_once_with(plain_password="plain")
    # User.create가 올바른 인자로 호출됐는지 검증
    user_create.assert_called_once_with(username="test", hashed_password="hashed")

    assert response.status_code == 201
    assert response.json() == {"id": 1, "username": "test"}

---

6. 로그인 API & JWT 인증

JWT란?

JWT(JSON Web Token) 는 사용자 인증에 사용되는 JSON 포맷의 토큰입니다.

장점:

• 토큰 자체에 유저 정보가 담겨 있어 별도의 DB 조회 없이 인증 처리 가능
• 토큰 변조를 검증할 수 있어 내장 데이터를 신뢰 가능
• 토큰에 만료 시간 설정 가능

인증 절차

클라이언트                         서버
    │                               │
    │──── id & password ──────────▶│
    │                          id&password 검증
    │                          JWT 생성
    │◀─── JWT 반환(access_token) ───│
    │                               │
    │ JWT 저장 (localStorage 등)    │
    │                               │
    │──── API 요청 (헤더: JWT) ───▶│
    │                          JWT 검증
    │◀─── 응답 ─────────────────────│

패키지 설치

pip install python-jose  # JWT 생성 및 검증 라이브러리

Secret Key 생성

# 터미널에서 랜덤한 비밀 키 생성
openssl rand -hex 32
# 출력 예: 8db9665caeab0ead6ebdf79150b7e4976a62e11e8cb15779fe06b64834b36dea

이 키는 JWT를 서명하고 검증하는 데 사용됩니다. 절대 외부에 노출하면 안 됩니다.

service/user.py — JWT 관련 메서드 추가

import bcrypt
from datetime import datetime, timedelta
from jose import jwt

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "8db9665caeab0ead6ebdf79150b7e4976a62e11e8cb15779fe06b64834b36dea"
    jwt_algorithm: str = "HS256"  # 서명 알고리즘

    def hash_password(self, plain_password: str) -> str:
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()
        )
        return hashed_password.decode(self.encoding)

    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        # 입력한 비밀번호와 DB의 해시값을 비교
        return bcrypt.checkpw(
            plain_password.encode(self.encoding),
            hashed_password.encode(self.encoding)
        )

    def create_jwt(self, username: str) -> str:
        return jwt.encode(
            {
                "sub": username,       # 토큰의 주체 (유저 식별자)
                "exp": datetime.now() + timedelta(days=1)  # 만료 시간: 1일
            },
            self.secret_key,
            algorithm=self.jwt_algorithm
        )

    def decode_jwt(self, access_token: str) -> str:
        # 토큰을 복호화해서 payload(dict)를 꺼냄
        payload: dict = jwt.decode(
            access_token,
            self.secret_key,
            algorithms=[self.jwt_algorithm]
        )
        # sub 필드에서 username 반환
        return payload["sub"]

database/repository.py — get_user_by_username 추가

class UserRepository:
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session

    def get_user_by_username(self, username: str) -> User | None:
        # username으로 유저 조회
        return self.session.scalar(
            select(User).where(User.username == username)
        )

    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()
        self.session.refresh(instance=user)
        return user

schema/response.py — JWTResponse 추가

class JWTResponse(BaseModel):
    access_token: str  # 로그인 성공 시 반환할 JWT 토큰

api/user.py — 로그인 핸들러 추가

@router.post("/login-in")
def user_login_handler(
    request: LoginRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
):
    # 1. username으로 DB에서 유저 조회
    user: User | None = user_repo.get_user_by_username(username=request.username)
    if not user:
        raise HTTPException(status_code=404, detail="User Not Found")

    # 2. 입력한 비밀번호와 DB의 해시 비밀번호 비교
    verified: bool = user_service.verify_password(
        plain_password=request.password,
        hashed_password=user.password,
    )
    if not verified:
        raise HTTPException(status_code=401, detail="Not Authorized")

    # 3. JWT 생성
    access_token: str = user_service.create_jwt(username=user.username)

    # 4. JWT 반환
    return JWTResponse(access_token=access_token)

---

7. JWT 인증 적용

로그인 이후 API 요청 시 JWT 토큰을 헤더에 담아 보내고, 서버에서 검증합니다.

security.py

from fastapi import HTTPException, Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

def get_access_token(
    # HTTPBearer: Authorization 헤더에서 Bearer 토큰을 자동으로 파싱
    # auto_error=False: 토큰이 없어도 에러를 자동으로 내지 않고 None 반환
    auth_header: HTTPAuthorizationCredentials | None = Depends(HTTPBearer(auto_error=False))
) -> str:
    if auth_header is None:
        raise HTTPException(
            status_code=401,
            detail="Not Authorized",
        )
    return auth_header.credentials  # Bearer 뒤의 실제 토큰 문자열 반환

api/todo.py — GET 전체 조회에 JWT 인증 적용

@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),  # JWT 토큰 검증
    order: str | None = None,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
) -> ToDoListSchema:
    # 토큰에서 username 추출
    username: str = user_service.decode_jwt(access_token=access_token)

    # username으로 유저 조회
    user: User | None = user_repo.get_user_by_username(username=username)
    if not user:
        raise HTTPException(status_code=404, detail="User Not Found")

    # Eager Loading으로 이미 가져온 todos 사용 (추가 DB 조회 없음)
    todos: List[ToDo] = user.todos

    if order and order == "DESC":
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

테스트 코드 변경 — JWT 헤더 추가

def test_get_todos(client, mocker):
    # 실제 JWT를 생성해서 테스트 헤더에 포함
    access_token: str = UserService().create_jwt(username="test")
    headers = {"Authorization": f"Bearer {access_token}"}

    # 유저 객체에 todos를 직접 할당 (Eager Loading 시뮬레이션)
    user = User(id=1, username="test", password="hashed")
    user.todos = [
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDo(id=2, contents="FastAPI Section 1", is_done=False),
    ]

    mocker.patch.object(
        UserRepository,
        "get_user_by_username",
        return_value=user
    )

    # 헤더에 JWT 포함해서 요청
    response = client.get("/todos", headers=headers)
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 1", "is_done": False},
        ]
    }

    response = client.get("/todos?order=DESC", headers=headers)
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 2, "contents": "FastAPI Section 1", "is_done": False},
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

---

8. 캐싱 & OTP

캐싱(Caching)이란?

데이터를 임시 저장소에 저장해서 빠르게 조회하는 기술입니다.

특징	설명
빠른 조회	DB보다 훨씬 빠름
영속성 없음	서버 재시작 시 데이터 소실
주요 활용	임시 데이터, 반복 조회 데이터, 빠른 응답이 필요한 데이터

OTP(One-Time Password)란?

일회용 비밀번호로, 캐싱의 대표적인 활용 사례입니다.

• 이메일 인증 시 4자리 랜덤 숫자 생성
• Redis에 {email: otp} 형태로 저장
• 3분 후 자동 만료

Redis란?

캐싱에 자주 사용되는 key-value 데이터 스토어 (NoSQL의 일종)

pip install redis

Redis 동작 확인 (Python 콘솔)

import redis

redis_client = redis.Redis(
    host="127.0.0.1", port=6379, db=0,
    encoding="UTF-8", decode_responses=True
)

redis_client.set("key", "value")   # 저장
redis_client.get("key")            # 조회 → 'value'
redis_client.expire("key", 10)     # 10초 후 만료 설정

cache.py

import redis

redis_client = redis.Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    encoding="utf-8",
    decode_responses=True  # bytes가 아닌 str로 반환
)

service/user.py — OTP 생성 메서드 추가

import random

@staticmethod
def create_otp() -> int:
    # 1000~9999 사이의 랜덤 4자리 숫자 반환
    return random.randint(1000, 9999)

schema/request.py — OTP 요청 모델 추가

class CreateOTPRequest(BaseModel):
    email: str

class VerifyOTPRequest(BaseModel):
    email: str
    otp: int

---

9. OTP 생성 & 검증 API

api/user.py — OTP 관련 핸들러

from schema.request import CreateOTPRequest, VerifyOTPRequest
from cache import redis_client

# OTP 생성 API
# POST /users/email/otp
@router.post("/email/otp")
def create_otp_handler(
    request: CreateOTPRequest,
    _: str = Depends(get_access_token),   # 로그인한 유저만 접근 가능 (토큰 검증만 하고 값은 사용 안 함)
    user_service: UserService = Depends(),
):
    # 1. 4자리 OTP 생성
    otp: int = user_service.create_otp()

    # 2. Redis에 저장 (key: email, value: otp, 만료: 3분)
    redis_client.set(request.email, otp)
    redis_client.expire(request.email, 3 * 60)

    # 3. 실제 서비스에서는 이메일로 전송, 현재는 응답으로 반환
    return {"otp": otp}


# OTP 검증 API
# POST /users/email/otp/verify
@router.post("/email/otp/verify")
def verify_otp_handler(
    request: VerifyOTPRequest,
    access_token: str = Depends(get_access_token),
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
):
    # 1. Redis에서 해당 이메일의 OTP 조회
    otp: str | None = redis_client.get(request.email)

    if not otp:
        # Redis에 OTP가 없음 = 만료됐거나 요청하지 않은 이메일
        raise HTTPException(status_code=400, detail="Bad Request")

    if request.otp != int(otp):
        # 입력한 OTP가 저장된 OTP와 다름
        raise HTTPException(status_code=400, detail="Bad Request")

    # 2. 토큰에서 username 추출 후 유저 조회
    username: str = user_service.decode_jwt(access_token=access_token)
    user: User | None = user_repo.get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=404, detail="User Not Found")

    # 3. 이메일 저장 (실제 구현은 생략)
    return UserSchema.from_orm(user)

_: str = Depends(get_access_token) : 토큰 검증만 하고 값은 사용하지 않겠다는 의미로 변수명을 _로 씁니다.

---

10. Background Task (백그라운드 작업)

이메일 전송처럼 시간이 오래 걸리는 작업을 응답 이후에 백그라운드에서 처리하는 기능입니다.

일반 처리 vs 백그라운드 처리

# 일반 처리 (응답이 느림)
요청 → OTP 검증 → 이메일 전송(10초) → 응답

# 백그라운드 처리 (응답이 빠름)
요청 → OTP 검증 → 응답
                 → 이메일 전송(10초, 백그라운드에서 별도 처리)

api/user.py — Background Task 적용

from fastapi import BackgroundTasks

@router.post("/email/otp/verify")
def verify_otp_handler(
    request: VerifyOTPRequest,
    background_tasks: BackgroundTasks,   # FastAPI가 자동으로 주입해주는 백그라운드 작업 큐
    access_token: str = Depends(get_access_token),
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
):
    otp: str | None = redis_client.get(request.email)
    if not otp:
        raise HTTPException(status_code=400, detail="Bad Request")
    if request.otp != int(otp):
        raise HTTPException(status_code=400, detail="Bad Request")

    username: str = user_service.decode_jwt(access_token=access_token)
    user: User | None = user_repo.get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=404, detail="User Not Found")

    # 응답을 먼저 보내고, 이메일 전송은 백그라운드에서 처리
    background_tasks.add_task(
        user_service.send_email_to_user,  # 실행할 함수
        email="admin@fastapi.com"         # 함수에 전달할 인자
    )

    return UserSchema.from_orm(user)

---

11. 전체 흐름 정리

회원가입
클라이언트 → POST /users/sign-up (username, password)
           → 비밀번호 bcrypt 해싱
           → DB 저장
           → UserSchema 반환 (id, username)

로그인
클라이언트 → POST /users/login-in (username, password)
           → DB에서 유저 조회
           → bcrypt로 비밀번호 검증
           → JWT 생성 (유효기간 1일)
           → JWTResponse(access_token) 반환

인증이 필요한 API 요청
클라이언트 → GET /todos (헤더: Authorization: Bearer {token})
           → JWT 검증 (security.py)
           → 토큰에서 username 추출
           → DB에서 유저 조회
           → 유저의 todos 반환

OTP 이메일 인증
클라이언트 → POST /users/email/otp (email)
           → 4자리 OTP 생성
           → Redis 저장 (만료: 3분)
           → OTP 반환 (실제로는 이메일 전송)

           → POST /users/email/otp/verify (email, otp)
           → Redis에서 OTP 조회 및 검증
           → 이메일 저장
           → 이메일 전송 (Background Task)

---

12. 핵심 개념 요약

개념	설명
ForeignKey	다른 테이블의 컬럼을 참조하는 외래키
relationship	ORM에서 연관 테이블 데이터를 객체로 접근 가능하게 함
lazy="joined"	Eager Loading — 처음부터 JOIN해서 함께 조회
bcrypt	단방향 해싱 라이브러리 (비밀번호 저장용)
JWT	유저 정보를 담은 인증 토큰 (만료 시간 설정 가능)
HTTPBearer	FastAPI에서 Bearer 토큰을 헤더에서 자동 추출
Redis	key-value 기반 캐시 저장소
BackgroundTasks	응답 후 비동기로 실행되는 백그라운드 작업