FastAPI CRUD 애플리케이션 완벽 가이드: SQLAlchemy와 Tortoise ORM으로 구축하는 실전 프로젝트

FastAPI를 사용하여 CRUD(Create, Read, Update, Delete) 애플리케이션을 만드는 방법을 알아보겠습니다.
이번 포스팅에서는 SQLAlchemy를 사용하여 데이터베이스와 연동하는 방법을 설명합니다. Tortoise ORM을 사용하는 방법도 함께 설명하려 합니다.

---

필요 라이브러리 설치

먼저 FastAPI, SQLAlchemy, 데이터베이스 드라이버를 설치합니다. 여기서는 SQLite를 사용할 것입니다.

pip install fastapi uvicorn sqlalchemy databases[sqlite] pydantic

추가 권장 라이브러리:

# 개발 환경을 위한 추가 패키지
pip install pytest pytest-asyncio httpx

---

SQLAlchemy로 CRUD 애플리케이션 만들기

프로젝트 구조

fastapi-crud/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   └── database.py
├── tests/
│   ├── __init__.py
│   └── test_crud.py
└── requirements.txt

데이터베이스 설정

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

# 환경 변수에서 데이터베이스 URL 가져오기
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")

# SQLite의 경우 check_same_thread=False 추가
if DATABASE_URL.startswith("sqlite"):
    engine = create_engine(
        DATABASE_URL, 
        connect_args={"check_same_thread": False}
    )
else:
    engine = create_engine(DATABASE_URL)

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

# 의존성 주입을 위한 함수
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

데이터베이스 모델 정의

# app/models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text
from sqlalchemy.sql import func
from .database import Base

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), index=True, nullable=False)
    description = Column(Text)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(100), unique=True, index=True, nullable=False)
    username = Column(String(50), unique=True, index=True, nullable=False)
    full_name = Column(String(100))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

Pydantic 스키마 정의

# app/schemas.py
from pydantic import BaseModel, EmailStr
from typing import Optional
from datetime import datetime

# Item 스키마
class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    is_active: bool = True

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    is_active: Optional[bool] = None

class Item(ItemBase):
    id: int
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True  # Pydantic v2에서 orm_mode 대신 사용

# User 스키마
class UserBase(BaseModel):
    email: EmailStr
    username: str
    full_name: Optional[str] = None
    is_active: bool = True

class UserCreate(UserBase):
    pass

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    username: Optional[str] = None
    full_name: Optional[str] = None
    is_active: Optional[bool] = None

class User(UserBase):
    id: int
    created_at: datetime

    class Config:
        from_attributes = True

CRUD 함수 정의

# app/crud.py
from sqlalchemy.orm import Session
from sqlalchemy import and_
from typing import List, Optional
from . import models, schemas

# Item CRUD 함수들
def get_item(db: Session, item_id: int) -> Optional[models.Item]:
    return db.query(models.Item).filter(models.Item.id == item_id).first()

def get_items(
    db: Session, 
    skip: int = 0, 
    limit: int = 100,
    search: Optional[str] = None
) -> List[models.Item]:
    query = db.query(models.Item)

    if search:
        query = query.filter(
            models.Item.name.contains(search) | 
            models.Item.description.contains(search)
        )

    return query.offset(skip).limit(limit).all()

def create_item(db: Session, item: schemas.ItemCreate) -> models.Item:
    db_item = models.Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_item(
    db: Session, 
    item_id: int, 
    item_update: schemas.ItemUpdate
) -> Optional[models.Item]:
    db_item = get_item(db, item_id)
    if not db_item:
        return None

    update_data = item_update.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_item, field, value)

    db.commit()
    db.refresh(db_item)
    return db_item

def delete_item(db: Session, item_id: int) -> Optional[models.Item]:
    db_item = get_item(db, item_id)
    if not db_item:
        return None

    db.delete(db_item)
    db.commit()
    return db_item

# User CRUD 함수들
def get_user(db: Session, user_id: int) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.email == email).first()

def get_user_by_username(db: Session, username: str) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.username == username).first()

def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[models.User]:
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate) -> models.User:
    db_user = models.User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

FastAPI 애플리케이션 구현

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional

from . import crud, models, schemas
from .database import SessionLocal, engine, get_db

# 데이터베이스 테이블 생성
models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="FastAPI CRUD API",
    description="SQLAlchemy를 사용한 CRUD 애플리케이션",
    version="1.0.0"
)

# 헬스체크 엔드포인트
@app.get("/")
async def root():
    return {"message": "FastAPI CRUD API is running!"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# Item CRUD 엔드포인트
@app.post("/items/", response_model=schemas.Item, status_code=201)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    """새로운 아이템을 생성합니다."""
    return crud.create_item(db=db, item=item)

@app.get("/items/", response_model=List[schemas.Item])
def read_items(
    skip: int = Query(0, ge=0, description="건너뛸 레코드 수"),
    limit: int = Query(100, ge=1, le=1000, description="가져올 레코드 수"),
    search: Optional[str] = Query(None, description="검색어"),
    db: Session = Depends(get_db)
):
    """아이템 목록을 조회합니다."""
    items = crud.get_items(db, skip=skip, limit=limit, search=search)
    return items

@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    """특정 아이템을 조회합니다."""
    db_item = crud.get_item(db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

@app.put("/items/{item_id}", response_model=schemas.Item)
def update_item(
    item_id: int, 
    item: schemas.ItemUpdate, 
    db: Session = Depends(get_db)
):
    """아이템을 업데이트합니다."""
    db_item = crud.update_item(db, item_id=item_id, item_update=item)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

@app.delete("/items/{item_id}", response_model=schemas.Item)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    """아이템을 삭제합니다."""
    db_item = crud.delete_item(db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# User CRUD 엔드포인트
@app.post("/users/", response_model=schemas.User, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """새로운 사용자를 생성합니다."""
    # 이메일 중복 확인
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    # 사용자명 중복 확인
    db_user = crud.get_user_by_username(db, username=user.username)
    if db_user:
        raise HTTPException(status_code=400, detail="Username already taken")

    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=List[schemas.User])
def read_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    db: Session = Depends(get_db)
):
    """사용자 목록을 조회합니다."""
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """특정 사용자를 조회합니다."""
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

애플리케이션 실행

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

서버가 실행되면 http://127.0.0.1:8000/docs에 접속하여 Swagger UI를 통해 API를 테스트할 수 있습니다.

---

Tortoise ORM으로 CRUD 애플리케이션 만들기

필요한 라이브러리 설치

pip install fastapi tortoise-orm aiosqlite uvicorn

Tortoise ORM 모델 정의

# models.py
from tortoise import Model, fields
from tortoise.contrib.pydantic import pydantic_model_creator

class Item(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100, index=True)
    description = fields.TextField(null=True)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "items"

class User(Model):
    id = fields.IntField(pk=True)
    email = fields.CharField(max_length=100, unique=True, index=True)
    username = fields.CharField(max_length=50, unique=True, index=True)
    full_name = fields.CharField(max_length=100, null=True)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"

# Pydantic 모델 자동 생성
Item_Pydantic = pydantic_model_creator(Item, name="Item")
ItemIn_Pydantic = pydantic_model_creator(Item, name="ItemIn", exclude_readonly=True)

User_Pydantic = pydantic_model_creator(User, name="User")
UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", exclude_readonly=True)

FastAPI 애플리케이션 구현

# main.py
from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import register_tortoise
from tortoise.exceptions import DoesNotExist
from typing import List, Optional

from models import Item, User, Item_Pydantic, ItemIn_Pydantic, User_Pydantic, UserIn_Pydantic

app = FastAPI(
    title="FastAPI Tortoise ORM CRUD",
    description="Tortoise ORM을 사용한 비동기 CRUD 애플리케이션",
    version="1.0.0"
)

# 데이터베이스 설정
register_tortoise(
    app,
    db_url='sqlite://db.sqlite3',
    modules={'models': ['models']},
    generate_schemas=True,
    add_exception_handlers=True,
)

@app.get("/")
async def root():
    return {"message": "FastAPI Tortoise ORM CRUD API is running!"}

# Item CRUD 엔드포인트
@app.post("/items/", response_model=Item_Pydantic, status_code=201)
async def create_item(item: ItemIn_Pydantic):
    """새로운 아이템을 생성합니다."""
    item_obj = await Item.create(**item.model_dump(exclude_unset=True))
    return await Item_Pydantic.from_tortoise_orm(item_obj)

@app.get("/items/", response_model=List[Item_Pydantic])
async def get_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    search: Optional[str] = Query(None)
):
    """아이템 목록을 조회합니다."""
    query = Item.all()

    if search:
        query = query.filter(
            name__icontains=search
        ).union(
            Item.filter(description__icontains=search)
        )

    items = await query.offset(skip).limit(limit)
    return await Item_Pydantic.from_queryset(items)

@app.get("/items/{item_id}", response_model=Item_Pydantic)
async def get_item(item_id: int):
    """특정 아이템을 조회합니다."""
    try:
        item = await Item.get(id=item_id)
        return await Item_Pydantic.from_tortoise_orm(item)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Item not found")

@app.put("/items/{item_id}", response_model=Item_Pydantic)
async def update_item(item_id: int, item: ItemIn_Pydantic):
    """아이템을 업데이트합니다."""
    try:
        await Item.filter(id=item_id).update(**item.model_dump(exclude_unset=True))
        updated_item = await Item.get(id=item_id)
        return await Item_Pydantic.from_tortoise_orm(updated_item)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}", response_model=Item_Pydantic)
async def delete_item(item_id: int):
    """아이템을 삭제합니다."""
    try:
        item = await Item.get(id=item_id)
        item_data = await Item_Pydantic.from_tortoise_orm(item)
        await item.delete()
        return item_data
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Item not found")

# User CRUD 엔드포인트
@app.post("/users/", response_model=User_Pydantic, status_code=201)
async def create_user(user: UserIn_Pydantic):
    """새로운 사용자를 생성합니다."""
    # 이메일 중복 확인
    if await User.filter(email=user.email).exists():
        raise HTTPException(status_code=400, detail="Email already registered")

    # 사용자명 중복 확인
    if await User.filter(username=user.username).exists():
        raise HTTPException(status_code=400, detail="Username already taken")

    user_obj = await User.create(**user.model_dump(exclude_unset=True))
    return await User_Pydantic.from_tortoise_orm(user_obj)

@app.get("/users/", response_model=List[User_Pydantic])
async def get_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000)
):
    """사용자 목록을 조회합니다."""
    users = await User.all().offset(skip).limit(limit)
    return await User_Pydantic.from_queryset(users)

@app.get("/users/{user_id}", response_model=User_Pydantic)
async def get_user(user_id: int):
    """특정 사용자를 조회합니다."""
    try:
        user = await User.get(id=user_id)
        return await User_Pydantic.from_tortoise_orm(user)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="User not found")

애플리케이션 실행

uvicorn main:app --reload

---

테스트 코드 작성

SQLAlchemy 버전 테스트

# tests/test_crud.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_db, Base

# 테스트용 인메모리 데이터베이스
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "Test Description"}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test Item"
    assert data["description"] == "Test Description"
    assert "id" in data

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_read_item():
    # 먼저 아이템 생성
    create_response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "Test Description"}
    )
    item_id = create_response.json()["id"]

    # 아이템 조회
    response = client.get(f"/items/{item_id}")
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == item_id

def test_update_item():
    # 먼저 아이템 생성
    create_response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "Test Description"}
    )
    item_id = create_response.json()["id"]

    # 아이템 업데이트
    response = client.put(
        f"/items/{item_id}",
        json={"name": "Updated Item", "description": "Updated Description"}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Updated Item"

def test_delete_item():
    # 먼저 아이템 생성
    create_response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "Test Description"}
    )
    item_id = create_response.json()["id"]

    # 아이템 삭제
    response = client.delete(f"/items/{item_id}")
    assert response.status_code == 200

    # 삭제된 아이템 조회 시 404 확인
    response = client.get(f"/items/{item_id}")
    assert response.status_code == 404

---

결론

FastAPI를 사용하여 SQLAlchemy 또는 Tortoise ORM으로 CRUD 애플리케이션을 만드는 방법을 알아보았습니다. 각 ORM의 특징을 이해하고, 필요한 기능에 맞게 선택하여 사용할 수 있습니다.
제가 함께 일해온 친구들의 면접에서 항상 하던 질문이 있습니다.
게시판을 만드는 데 얼마나 걸릴 것 같아요?
이 질문은 여러 가지 요소를 고려해야 합니다. 큰 기능은 아니지만 요구사항 수집, 분석, DB 설계, 화면 설계, 구현, 테스트, 배포 등 프로젝트의 전반적인 과정을 머릿속에 그려봐야 답을 낼 수 있습니다.
모든 프로그램은 'HELLO WORLD!'로 시작하고 결국 CRUD 작업을 하게 됩니다. 이것만 잘 구현되면 할 수 있는 것들이 정말 많아집니다.
핵심 포인트:

• SQLAlchemy vs Tortoise ORM 선택 기준과 장단점 이해
• Pydantic 스키마를 통한 데이터 검증과 직렬화
• 의존성 주입을 활용한 데이터베이스 세션 관리
• 에러 핸들링과 HTTP 상태 코드 적절한 사용
• 테스트 코드 작성으로 안정성 확보