17 FastAP SQLAlchemy Pydantic 간단한 REST API 예제 프로젝트 설명 | ✅ 저자: 이유정(박사)

FastAPI + SQLAlchemy + Pydantic: 간단한 REST API 예제 프로젝트 설명

17 DB연결 정리 --

FastAPI, SQLAlchemy, Pydantic을 사용하여 간단한 사용자(User) 및 아이템(Item) 관리 REST API를 구축하는 구조와 로직을 전체적으로 정리합니다.

---

1. 프로젝트 구조

FastAPI/
├── sql_app/
│    ├── __init__.py
│    ├── crud.py         # DB 관련 로직 모음
│    ├── database.py     # DB 연결 및 세션 설정
│    ├── main.py         # FastAPI 진입점
│    ├── models.py       # SQLAlchemy ORM 모델 정의
│    ├── schemas.py      # Pydantic 데이터 검증 스키마

실행 위치는 반드시 FastAPI/ 폴더에서!

uvicorn sql_app.main:app --reload

위 명령은 반드시 프로젝트 루트(= FastAPI) 디렉토리에서 실행해야 합니다.

---

2. 전체 호출 로직 흐름 이해하기

FastAPI 프로젝트는 요청(request)을 라우터에서 받아, 비즈니스 로직(CRUD), 데이터 스키마(Pydantic), 데이터베이스 ORM(SQLAlchemy)을 순차적으로 호출하여 응답(response)을 생성합니다.

예를 들어, POST /users/ 요청이 들어올 때 호출 흐름은 다음과 같습니다:

[Client] → POST /users/ (요청)
   ↓
[main.py] → @app.post("/users/")
   ↓
[schemas.py] → UserCreate 모델로 요청 데이터 유효성 검사
   ↓
[crud.py] → create_user 함수로 DB 조작 로직 실행
   ↓
[models.py] → SQLAlchemy User 모델로 실제 INSERT 실행
   ↓
[database.py] → SessionLocal()로 DB 세션 열고 commit/close 관리
   ↓
[Response] → Pydantic User 모델 기반으로 응답 직렬화 후 반환

3. 주요 흐름 도식화

┌────────────┐
│   Client   │
└────┬───────┘
     ↓
[main.py - API 엔드포인트]
     ↓
[schemas.py - 입력 스키마 검증]
     ↓
[crud.py - 비즈니스 로직 실행]
     ↓
[models.py - SQLAlchemy ORM 모델]
     ↓
[database.py - DB 연결 및 트랜잭션 처리]
     ↓
[schemas.py - 출력 스키마 직렬화]
     ↓
┌────────────┐
│  Response  │
└────────────┘

이와 같이 FastAPI는 Python 타입 힌트를 기반으로 각 계층이 역할을 나누며 처리하므로 유지보수성과 가독성이 높습니다. 다음은 구성 요소별 실제 코드입니다.

---

4.구성 요소

✅ models.py: DB 테이블 구조 정의

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String  
from sqlalchemy.orm import relationship
from .database import Base  

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    items = relationship("Item", back_populates="owner")

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="items")

• User와 Item은 SQLAlchemy ORM으로 정의된 데이터베이스 모델입니다.
• relationship()은 SQL에서의 JOIN 관계를 코드에서 표현하는 방식입니다.
• 각 사용자(User)는 여러 개의 아이템(Item)을 소유할 수 있도록 1:N 관계 설정이 되어 있습니다.

✅ schemas.py: API 데이터 검증/직렬화

from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int
    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []
    class Config:
        orm_mode = True

• BaseModel을 기반으로 FastAPI는 JSON 요청에 대해 자동 유효성 검사를 수행합니다.
• orm_mode = True는 SQLAlchemy 모델을 Pydantic 객체로 자동 변환해주는 기능을 의미합니다.

✅ database.py: DB 연결 및 세션 관리

from sqlalchemy import create_engine  
from sqlalchemy.ext.declarative import declarative_base  
from sqlalchemy.orm import sessionmaker  


SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

• SQLite 사용 예제이며, 실제 운영 환경에서는 PostgreSQL 또는 MySQL로 변경 가능.
• SessionLocal은 FastAPI의 각 요청마다 독립적인 DB 세션을 제공합니다.

✅ crud.py: 비즈니스 로직 (DB 조작)

from sqlalchemy.orm import Session
from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_items(db: Session, skip=0, limit=100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def get_users(db: Session, skip=0, limit=100):
    return db.query(models.User).offset(skip).limit(limit).all()

• 실제 DB와 상호작용하는 핵심 함수들이 정의됩니다.
• 사용자 및 아이템 생성, 목록 조회 등의 기능을 포함합니다.
• 추후 단위 테스트 작성 시에도 이 함수들을 테스트하면 됩니다.

✅ main.py: FastAPI 앱 진입점

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from sql_app import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=schemas.User)
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")
    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

@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

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_user_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_items(db, skip=skip, limit=limit)

---

5. 샘플 API 테스트 URL 및 더미 데이터 예시

🟢 사용자 생성

• POST http://127.0.0.1:8000/users/

{
  "email": "user1@example.com",
  "password": "testpass123"
}

🟢 사용자 목록 조회

• GET http://127.0.0.1:8000/users/

🟢 특정 사용자 조회

• GET http://127.0.0.1:8000/users/1

🟢 사용자에게 아이템 추가

• POST http://127.0.0.1:8000/users/1/items/

{
  "title": "Sample Item 1",
  "description": "FastAPI 테스트용 아이템입니다."
}

🟢 전체 아이템 목록 조회

• GET http://127.0.0.1:8000/items/

{
  "email": "user2@example.com",
  "password": "pass2"
}
{
  "title": "Item A",
  "description": "설명 A"
}
{
  "email": "user3@example.com",
  "password": "pass3"
}
{
  "title": "Item B",
  "description": "설명 B"
}

---

---

6. VSCode REST Client 테스트 코드 예시

🧪 api_test.http

### 사용자 생성
POST http://127.0.0.1:8000/users/
Content-Type: application/json

{
  "email": "testuser1@example.com",
  "password": "testpass123"
}

### 사용자 목록 조회
GET http://127.0.0.1:8000/users/
Accept: application/json

### 특정 사용자 조회
GET http://127.0.0.1:8000/users/1
Accept: application/json

### 사용자에게 아이템 추가
POST http://127.0.0.1:8000/users/1/items/
Content-Type: application/json

{
  "title": "REST Client Test Item",
  "description": "This is a test item created from VSCode REST Client."
}

### 전체 아이템 목록 조회
GET http://127.0.0.1:8000/items/
Accept: application/json

---

7. Postman Collection (JSON)

{
  "info": {
    "_postman_id": "fastapi-example-collection",
    "name": "FastAPI Example",
    "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
  },
  "item": [
    {
      "name": "Create User",
      "request": {
        "method": "POST",
        "header": [
          {"key": "Content-Type", "value": "application/json"}
        ],
        "body": {
          "mode": "raw",
          "raw": "{\n  \"email\": \"user1@example.com\",\n  \"password\": \"testpass123\"\n}"
        },
        "url": {"raw": "http://127.0.0.1:8000/users/", "protocol": "http", "host": ["127.0.0.1"], "port": "8000", "path": ["users"]}
      }
    },
    {
      "name": "Get Users",
      "request": {
        "method": "GET",
        "header": [],
        "url": {"raw": "http://127.0.0.1:8000/users/", "protocol": "http", "host": ["127.0.0.1"], "port": "8000", "path": ["users"]}
      }
    },
    {
      "name": "Get User by ID",
      "request": {
        "method": "GET",
        "url": {"raw": "http://127.0.0.1:8000/users/1", "protocol": "http", "host": ["127.0.0.1"], "port": "8000", "path": ["users", "1"]}
      }
    },
    {
      "name": "Create Item for User",
      "request": {
        "method": "POST",
        "header": [
          {"key": "Content-Type", "value": "application/json"}
        ],
        "body": {
          "mode": "raw",
          "raw": "{\n  \"title\": \"Item 1\",\n  \"description\": \"FastAPI Postman Test\"\n}"
        },
        "url": {"raw": "http://127.0.0.1:8000/users/1/items/", "protocol": "http", "host": ["127.0.0.1"], "port": "8000", "path": ["users", "1", "items"]}
      }
    },
    {
      "name": "Get Items",
      "request": {
        "method": "GET",
        "url": {"raw": "http://127.0.0.1:8000/items/", "protocol": "http", "host": ["127.0.0.1"], "port": "8000", "path": ["items"]}
      }
    }
  ]
}

이 JSON은 Postman에서 Import > Raw text로 붙여넣기 하면 바로 사용할 수 있습니다.

---

🔚

• FastAPI는 빠르고 직관적인 API 프레임워크입니다.
• SQLAlchemy는 안정적인 ORM으로, SQL 없이도 데이터 조작이 가능합니다.
• Pydantic은 타입 기반의 유효성 검증과 직렬화를 자동으로 처리합니다.