FastAPI 섹션 2. 데이터베이스

---

1. SQLAlchemy란?

SQLAlchemy는 Python에서 관계형 데이터베이스를 편리하게 사용할 수 있도록 도와주는 라이브러리입니다.

주요 기능:

• ORM (Object-Relational Mapping)
• Query (쿼리 생성)
• Transaction (트랜잭션 처리)
• Connection Pooling (DB 연결 관리)

---

2. ORM이란?

ORM(Object-Relational Mapping) 은 관계형 데이터베이스의 테이블을 Python 클래스(객체)처럼 다룰 수 있게 해주는 기술입니다.

데이터베이스 개념	Python 개념
테이블 (Table)	클래스 (Class)
행/레코드 (Row)	객체 (Object)
컬럼 (Column)	속성 (Attribute)

예시:

데이터베이스 테이블:

id	username
1	qu3vipon

Python ORM 코드:

user = User(id=1, username='qu3vipon')

즉, SQL을 직접 쓰지 않고 Python 코드로 DB를 조작할 수 있게 됩니다.

---

3. Docker로 MySQL 실행하기

로컬에 MySQL을 직접 설치하지 않고, Docker를 이용해 간편하게 실행합니다.

# MySQL 컨테이너 실행
docker run -p 3306:3306 \
  -e MYSQL_ROOT_PASSWORD=todos \
  -e MYSQL_DATABASE=todos \
  -d \
  -v todos:/db \
  --name todos \
  mysql:8.0

# 실행 중인 컨테이너 확인
docker ps

# 컨테이너 로그 확인
docker logs todos

# 볼륨 목록 확인
docker volume ls

명령어 옵션 설명:

• -p 3306:3306 : 로컬 포트 3306 ↔ 컨테이너 포트 3306 연결
• -e MYSQL_ROOT_PASSWORD=todos : root 비밀번호 설정
• -e MYSQL_DATABASE=todos : 기본 DB 이름 설정
• -d : 백그라운드 실행
• -v todos:/db : 데이터 영속성을 위한 볼륨 마운트
• --name todos : 컨테이너 이름 지정

---

4. MySQL 접속 & 테이블 생성 (DDL)

# MySQL 컨테이너 내부 접속
docker exec -it todos bash

# MySQL 로그인
mysql -u root -p

-- 데이터베이스 목록 확인
SHOW databases;

-- todos DB 선택
USE todos;

-- todo 테이블 생성
CREATE TABLE todo(
    id INT NOT NULL AUTO_INCREMENT,
    contents VARCHAR(256) NOT NULL,
    is_done BOOLEAN NOT NULL,
    PRIMARY KEY (id)
);

-- 샘플 데이터 삽입
INSERT INTO todo (contents, is_done) VALUES ("FastAPI Section 0", true);

-- 데이터 조회
SELECT * FROM todo;

---

5. 데이터베이스 연결 설정

패키지 설치

pip install SQLAlchemy   # ORM 라이브러리
pip install pyMySQL      # Python ↔ MySQL 연결 드라이버
pip install cryptography # MySQL 인증/암호화 처리

database/connection.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# DB 연결 주소 (형식: mysql+pymysql://유저:비밀번호@호스트:포트/DB명)
DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"

# 엔진 생성 (실제 DB와 연결하는 핵심 객체, echo=True는 실행된 SQL을 콘솔에 출력)
engine = create_engine(DATABASE_URL, echo=True)

# 세션 팩토리 생성 (DB와 대화하는 세션을 만드는 공장)
# autocommit=False : 명시적으로 commit() 해야 DB에 반영됨
# autoflush=False  : 명시적으로 flush() 하기 전까지 쿼리를 DB에 보내지 않음
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

연결 확인 (Python 콘솔)

>>> from database.connection import SessionFactory
>>> session = SessionFactory()
>>> from sqlalchemy import select
>>> session.scalar(select(1))  # 1이 반환되면 연결 성공!

---

6. ORM 모델링

database/orm.py

from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()  # 모든 ORM 모델의 부모 클래스

class ToDo(Base):
    __tablename__ = "todo"  # 연결할 실제 DB 테이블 이름

    # 컬럼 정의 (DB 테이블의 각 컬럼과 1:1 매핑)
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)

    def __repr__(self):
        # 객체를 출력할 때 보여줄 문자열 형식
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"

ORM 모델 동작 확인 (Python 콘솔)

>>> from database.connection import SessionFactory
>>> session = SessionFactory()
>>> from sqlalchemy import select
>>> from database.orm import ToDo

# 모든 ToDo 조회
>>> todos = list(session.scalars(select(ToDo)))
>>> todos

# 하나씩 출력
>>> for todo in todos:
...     print(todo)

---

7. scalar vs scalars 차이

SQLAlchemy에서 헷갈리기 쉬운 두 메서드의 차이입니다.

# scalars() - 여러 행을 ORM 객체 리스트로 반환
session.scalars(select(ToDo))
# → [ToDo(id=1, ...), ToDo(id=2, ...), ...]

# scalar() - 단일 행 하나만 반환
session.scalar(select(ToDo).where(ToDo.id == 1))
# → ToDo(id=1, ...) 객체 하나

쿼리 결과는 원래 행/열의 2차원 테이블 형태인데, 첫 번째 열만 뽑아서 단순하게(scalar) 만들어줍니다.
덕분에 Row 객체가 아니라 바로 ToDo 객체로 받을 수 있습니다.

---

8. ORM 적용 — CRUD API 구현

사전 준비: 의존성 주입 (DB 세션)

FastAPI의 Depends를 이용해 각 API 핸들러에 DB 세션을 자동으로 주입합니다.

# database/connection.py 에 추가
from sqlalchemy.orm import Session

def get_db():
    # 요청마다 새 세션을 열고, 요청이 끝나면 자동으로 닫아줌
    db = SessionFactory()
    try:
        yield db
    finally:
        db.close()

---

Response 스키마 정의

API가 반환할 데이터 형식을 Pydantic으로 정의합니다.

# schema/response.py
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class Config:
        from_attributes = True  # ORM 객체 → Pydantic 변환 허용 (구버전: orm_mode = True)

class ListToDoResponse(BaseModel):
    todos: List[ToDoSchema]

---

GET — 전체 조회

# database/repository.py
from typing import List
from sqlalchemy import select
from sqlalchemy.orm import Session
from database.orm import ToDo

def get_todos(session: Session) -> List[ToDo]:
    return list(session.scalars(select(ToDo)))

# main.py
@app.get("/todos", status_code=200)
def get_todos_handler(
    order: str | None = None,
    session: Session = Depends(get_db),  # DB 세션 자동 주입
) -> ListToDoResponse:
    todos: List[ToDo] = get_todos(session=session)

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

ToDoSchema.from_orm(todo) : ORM 객체(ToDo)를 Pydantic 스키마(ToDoSchema)로 변환합니다.

---

GET — 단일 조회

# database/repository.py 에 추가
def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
    # where 조건으로 특정 id를 가진 행 하나만 반환
    return session.scalar(select(ToDo).where(ToDo.id == todo_id))

# main.py
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(
    todo_id: int,
    session: Session = Depends(get_db),
) -> ToDoSchema:
    todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)
    if todo:
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

---

POST — 생성

# schema/request.py
from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
    # id는 DB가 AUTO_INCREMENT로 자동 생성하므로 요청에 포함하지 않음

# database/repository.py 에 추가
def create_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)   # 세션에 객체 추가 (아직 DB에 저장 안 됨)
    session.commit()              # DB에 실제로 저장 (INSERT 실행)
    session.refresh(instance=todo)  # DB에서 최신 상태 다시 읽어오기 (자동 생성된 id 반영)
    return todo

# main.py
@app.post("/todos", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    session: Session = Depends(get_db),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)  # ORM 객체 생성 (id=None 상태)
    todo: ToDo = create_todo(session=session, todo=todo)  # DB 저장 후 id 할당됨
    return ToDoSchema.from_orm(todo)

session.refresh() 가 중요한 이유: DB에 저장 후 AUTO_INCREMENT로 생성된 id 값을 Python 객체에 반영하기 위해 필요합니다.

---

PATCH — 수정

# database/repository.py 에 추가
def update_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)   # 변경된 객체를 세션에 반영
    session.commit()              # DB에 UPDATE 실행
    session.refresh(instance=todo)  # 최신 상태 다시 읽기
    return todo

# main.py
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    session: Session = Depends(get_db),
):
    todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)
    if todo:
        # is_done 값에 따라 done() 또는 undone() 메서드 호출
        todo.done() if is_done else todo.undone()
        todo: ToDo = update_todo(session=session, todo=todo)
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

---

DELETE — 삭제

# database/repository.py 에 추가
from sqlalchemy import delete

def delete_todo(session: Session, todo_id: int) -> None:
    # DELETE FROM todo WHERE id = todo_id 실행
    session.execute(delete(ToDo).where(ToDo.id == todo_id))
    session.commit()

# main.py
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(
    todo_id: int,
    session: Session = Depends(get_db),
):
    todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")
    delete_todo(session=session, todo_id=todo_id)
    # 204 No Content: 성공했지만 반환할 데이터 없음 → return 생략

---

9. 리팩토링 팁 (PyCharm / IDE)

코드가 길어지면 파일을 분리해서 관리하는 게 중요합니다.

• Move : 클래스/함수를 드래그 후 우클릭 → 다른 파일로 이동
• Rename : 클래스/변수 이름 변경 시 사용된 모든 곳에 자동 반영

---

10. 전체 흐름 정리

클라이언트 요청
     ↓
FastAPI 라우터 (main.py)
     ↓
Repository 함수 (database/repository.py) ← DB 쿼리 로직 분리
     ↓
ORM 모델 (database/orm.py) ← 테이블 구조 정의
     ↓
SQLAlchemy Engine (database/connection.py) ← 실제 DB 연결
     ↓
MySQL (Docker)