FastAPI 기초 실습

실습

fastapi_project/
│── main.py              # 앱 실행 진입점
│── database.py          # DB 연결/세션 관리
│── models.py            # SQLAlchemy 모델 정의
│── schemas.py           # Pydantic 스키마 정의
│── crud.py              # DB CRUD 로직
│── routers/
│     └── items.py       # 라우트 (엔드포인트)
│── tests/
      └── test_items.py  # 테스트 코드

---

main.py (앱 시작점)

from fastapi import FastAPI
from routers import items

app = FastAPI()							# 앱 객체 생성

# 라우터 등록
app.include_router(items.router)

@app.get("/")   # GET /
async def root():
    return {"msg": "Hello FastAPI"}

• from routers import items (flask의 blueprint느낌)
  • routers → 폴더(패키지) 이름
    • routers/ 폴더 안에 init.py 파일이 있으면 파이썬 패키지로 인식
    • 그러면 from routers import ... 형식으로 가져올 수 있음.
  • items → 파일 이름 (items.py) 즉, routers/items.py 파일 전체를 불러온 것.
• app.include_router(items.router)
  • include_router() : 라우터를 등록할때 사용(Flask에서 app.register_blueprint())
  • router은 items.py 안에서 만든 APIRouter 인스턴스 변수 이름
  • items.router는 items.py 안에 정의된 router 객체

---

database.py (DB 연결 관리)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

# 의존성 주입 (Depends에서 사용)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• create_engine() → DB 연결
  • connect_args={"check_same_thread": False}
    • SQLite는 가벼운 파일 기반 DB라서, "한 스레드에서만 같은 연결(Connection)을 사용 가능" 하도록 제한
• sessionmaker() → 세션 팩토리
• declarative_base() → 모델 베이스
• yield + Depends(get_db) → DB 세션 주입

---

models.py (SQLAlchemy 모델)

from sqlalchemy import Column, Integer, String
from database import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)

• Column(), Integer, String → 컬럼 정의
• primary_key=True, index=True → 키/인덱스 지정

---

schemas.py (Pydantic 스키마)

from pydantic import BaseModel

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

class ItemCreate(ItemBase):
    pass

class ItemRead(ItemBase):
    id: int

    class Config:
        orm_mode = True

• BaseModel → 데이터 검증/직렬화
• orm_mode = True → SQLAlchemy 모델 → JSON 변환 지원

---

crud.py (DB 접근 함수)

from sqlalchemy.orm import Session
from models import Item
from schemas import ItemCreate

def get_items(db: Session):
    return db.query(Item).all()

def create_item(db: Session, item: ItemCreate):
    db_item = Item(name=item.name, description=item.description)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

• db.query().all() → SELECT *
• db.add(), db.commit(), db.refresh() → INSERT

---

routers/items.py (라우터)

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
import crud, schemas

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/", response_model=list[schemas.ItemRead])
def read_items(db: Session = Depends(get_db)):
    return crud.get_items(db)

@router.post("/", response_model=schemas.ItemRead)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_item(db, item)

• APIRouter() → 라우터 모듈화
• Depends(get_db) → DB 세션 의존성 주입
• HTTPException → 에러 처리
• response_model= → 응답 스키마 지정

---

tests/test_items.py (테스트)

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello FastAPI"}

• TestClient(app) → FastAPI 앱 테스트
• .get(), .post() → API 호출 시뮬레이션

---