SQLAlchemy와 데이터베이스 연동

지금까지의 문제점

4편에서 만든 API는 데이터를 **메모리(딕셔너리)**에 저장했다. 서버를 재시작하면 모든 데이터가 사라진다.

# 메모리 저장 - 서버 재시작하면 사라짐
self._users: dict[int, UserResponse] = {}

실제 서비스라면 데이터베이스에 영구적으로 저장해야 한다. 이 글에서는 SQLAlchemy로 실제 DB에 연동한다.

---

ORM이란?

ORM(Object-Relational Mapping) 은 Python 클래스와 데이터베이스 테이블을 연결해주는 기술이다.

ORM 없이 SQL을 직접 쓰면:

# SQL 직접 작성
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
row = cursor.fetchone()
user = User(id=row[0], name=row[1], email=row[2])  # 수동으로 변환

ORM을 쓰면:

# ORM 사용
user = db.query(User).filter(User.id == user_id).first()
# Python 객체로 바로 받는다

SQL을 직접 쓸 필요 없이 Python 코드로 DB를 다룰 수 있다.

Spring Boot의 JPA/Hibernate와 같은 역할이다.

---

SQLAlchemy 설치

pip install sqlalchemy

이 글에서는 별도 설치 없이 바로 쓸 수 있는 SQLite를 사용한다. 실제 서비스에서는 PostgreSQL, MySQL 등을 쓴다.

---

1단계: DB 연결 설정

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLite 파일로 저장 (app.db 파일 생성됨)
DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite 전용 옵션
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

• engine : DB 연결 엔진. 실제 DB 파일/서버와 통신한다
• SessionLocal : DB 세션 팩토리. 요청마다 세션을 하나씩 만든다
• Base : ORM 모델이 상속할 베이스 클래스

PostgreSQL로 바꾸려면

# PostgreSQL
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
# connect_args 옵션 제거

연결 URL만 바꾸면 나머지 코드는 그대로 쓸 수 있다.

---

2단계: DB 모델 정의

DB 테이블과 1:1로 대응하는 클래스를 만든다.

# users/models.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"  # DB 테이블 이름

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)

• __tablename__ : DB 테이블 이름
• Column : 컬럼 정의
• primary_key=True : 기본키 (자동 증가)
• unique=True : 중복 불가
• nullable=False : NULL 불가

Spring Boot의 @Entity, @Column과 같은 역할이다.

---

3단계: 테이블 생성

앱 시작 시 DB 테이블을 생성한다.

# main.py
from fastapi import FastAPI
from database import engine, Base
import users.models  # 모델을 임포트해야 Base가 인식한다

# 테이블 생성 (이미 있으면 건너뜀)
Base.metadata.create_all(bind=engine)

app = FastAPI()

create_all()은 Base를 상속한 모든 모델을 DB 테이블로 만든다. 이미 테이블이 있으면 건너뛴다.

---

4단계: DB 세션 의존성

요청마다 DB 세션을 열고, 요청이 끝나면 닫아야 한다. yield 의존성으로 처리한다.

# database.py (추가)
from typing import Generator
from sqlalchemy.orm import Session

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db         # 이 지점에서 라우터 함수가 실행됨
    finally:
        db.close()       # 요청이 끝나면 항상 세션 닫기

yield 앞은 요청 처리 전, finally 블록은 요청 처리 후에 실행된다. 예외가 발생해도 세션이 닫힌다.

---

5단계: Repository - SQL 쿼리 작성

이제 Repository에서 실제 DB를 다룬다.

# users/repository.py
from sqlalchemy.orm import Session
from users.models import User

class UserRepository:

    def __init__(self, db: Session):
        self.db = db

    def find_all(self) -> list[User]:
        return self.db.query(User).all()

    def find_by_id(self, user_id: int) -> User | None:
        return self.db.query(User).filter(User.id == user_id).first()

    def find_by_email(self, email: str) -> User | None:
        return self.db.query(User).filter(User.email == email).first()

    def save(self, name: str, email: str) -> User:
        user = User(name=name, email=email)
        self.db.add(user)       # INSERT 준비
        self.db.commit()        # DB에 실제로 저장
        self.db.refresh(user)   # id 등 DB 생성 값 갱신
        return user

    def delete(self, user: User) -> None:
        self.db.delete(user)
        self.db.commit()

---

6단계: Service - 비즈니스 로직

Service는 이전과 거의 같다. 이메일 중복 체크가 추가됐다.

# users/service.py
from fastapi import HTTPException
from users.repository import UserRepository
from users.models import User
from users.schemas import CreateUserRequest

class UserService:

    def __init__(self, repository: UserRepository):
        self.repository = repository

    def get_all_users(self) -> list[User]:
        return self.repository.find_all()

    def get_user(self, user_id: int) -> User:
        user = self.repository.find_by_id(user_id)
        if user is None:
            raise HTTPException(status_code=404, detail=f"유저를 찾을 수 없습니다. id: {user_id}")
        return user

    def create_user(self, request: CreateUserRequest) -> User:
        # 이메일 중복 체크
        existing = self.repository.find_by_email(request.email)
        if existing:
            raise HTTPException(status_code=400, detail="이미 사용 중인 이메일입니다.")
        return self.repository.save(request.name, request.email)

---

7단계: 의존성 함수 수정

DB 세션을 받아서 Repository, Service를 만든다.

# users/dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session
from database import get_db
from users.repository import UserRepository
from users.service import UserService

def get_user_service(db: Session = Depends(get_db)) -> UserService:
    repository = UserRepository(db)
    return UserService(repository)

get_db가 DB 세션을 만들어서 db에 넣어주고, get_user_service가 그 db로 Repository와 Service를 만든다.

---

8단계: 응답 스키마에 Config 추가

ORM 객체를 Pydantic 모델로 자동 변환하려면 from_attributes = True 가 필요하다.

# users/schemas.py
from pydantic import BaseModel

class CreateUserRequest(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True  # SQLAlchemy 모델 → Pydantic 변환 허용

이 설정 없이 SQLAlchemy 모델을 반환하면 FastAPI가 직렬화할 수 없다.

---

전체 파일 구조

my_app/
├── main.py              ← 앱 생성, 테이블 생성, 라우터 등록
├── database.py          ← DB 연결, 세션, get_db 의존성
│
└── users/
    ├── models.py        ← SQLAlchemy 모델 (DB 테이블)
    ├── schemas.py       ← Pydantic 스키마 (요청/응답)
    ├── repository.py    ← DB 쿼리
    ├── service.py       ← 비즈니스 로직
    ├── router.py        ← HTTP 라우터
    └── dependencies.py  ← 의존성 함수

---

요청부터 DB까지 전체 흐름

POST /users 요청
    │
    ▼
Router.create_user()
  - Pydantic이 요청 바디 검증
  - Depends()로 UserService 주입
  - Depends(get_db)로 DB 세션 생성
    │
    ▼
UserService.create_user()
  - 이메일 중복 체크
    │
    ▼
UserRepository.save()
  - db.add() → db.commit()
  - SQLite 파일(app.db)에 영구 저장
    │
    ▼
Router
  - User 모델 → UserResponse 자동 변환
  - JSON으로 응답
    │
    ▼
get_db의 finally 블록
  - db.close() 세션 반환

서버를 재시작해도 app.db 파일에 데이터가 남아있다.

---

정리

---

시리즈: Python FastAPI 입문

1. 시작하기 전에
2. FastAPI 구조 - 라우터, 3계층 아키텍처
3. Pydantic과 의존성 주입
4. REST API 만들기
5. SQLAlchemy와 데이터베이스 연동 ← 현재 글