pydantic

FastAPI Pydantic SQLAlchemy 연동 완전 정복(2025년)

by

 

이 글은 FastAPI, Pydantic 2.x, 그리고 SQLAlchemy를 처음 접하는 개발자부터 실무에 활용하고자 하는 분들을 대상으로 합니다.

  • FastAPI 기반 웹 API를 만들 때 Pydantic 2.x 모델을 활용한 데이터 검증, SQLAlchemy ORM과의 자연스러운 연동 방식을 단계별로 이해할 수 있도록 돕습니다.
  • Python 기초 문법을 알고 있고, 웹 프레임워크 또는 ORM 사용 경험이 약간 있는 개발자. 초등학생도 이해할 수 있을 정도로 친절한 설명을 제공합니다.

FastAPI 개요

FastAPI는 현대적인 비동기(Asynchronous) 웹 프레임워크로, Starlette를 기반으로 하고, Pydantic을 통해 입력 검증과 직렬화를 제공하여 빠르고 안정적인 API를 구축할 수 있습니다.

  • 주요 특징
    • 비동기 지원: async def로 작성한 경로 함수(route)가 기본 비동기 처리
    • 자동 문서화: OpenAPI(Swagger)와 ReDoc을 자동 생성
    • 높은 성능: Uvicorn + ASGI 기반으로 빠른 처리 속도
  • 사용 장점
    1. 타입 힌트를 적극 활용한 유효성 검사
    2. 생산성과 가독성 모두 제고
    3. 최소한의 설정으로 바로 사용 가능

Pydantic 2.x 개념 이해

Pydantic은 Python 타입 힌트를 기반으로 런타임 시 데이터 검증(validation)과 직렬화(serialization)를 수행합니다. 2.x 버전은 더 가볍고 빠른 성능을 목표로 핵심 구조가 일부 변경되었습니다.

BaseModel 사용법

  1. 모델 정의 
    from pydantic import BaseModel, Field

class User(BaseModel):
    id: int
    name: str = Field(..., max_length=50)
    signup_ts: datetime | None = None
  2. 인스턴스 생성 
    user = User(id=1, name="Alice")

필드 검증(Validation)

  • 기본 검증: 타입이 일치하지 않으면 ValidationError 발생
  • 제약 조건: Field 파라미터로 min_length, max_length, gt, lt 등 설정 가능
  • 커스텀 검증: @field_validator 데코레이터로 복잡한 규칙 적용

직렬화와 역직렬화

  • 직렬화: model.json() 또는 model.dict()로 Python 객체 → JSON/Dict 변환
  • alias: 필드 이름을 외부와 다르게 매핑 가능
  • export 옵션: exclude_none=True로 None 필드 자동 제외

SQLAlchemy 소개

SQLAlchemy는 Python ORM(Object-Relational Mapping) 라이브러리로, 데이터베이스 테이블을 Python 클래스로 다룰 수 있게 해 줍니다.

엔진과 세션 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

모델 정의와 매핑

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

Base = declarative_base()

class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

Pydantic과 SQLAlchemy 연동

Pydantic 2.x와 호환되는 SQLAlchemy 버전(1.4+)을 사용하면, from_orm 옵션으로 SQLAlchemy ORM 인스턴스를 Pydantic 모델로 쉽게 변환할 수 있습니다.

from_orm 옵션

  1. Pydantic 모델 설정 
    class UserSchema(BaseModel):
    id: int
    name: str

    model_config = ConfigDict(from_attributes=True)
  2. ORM → Pydantic 변환 
    db_user = session.query(UserORM).first()
user_data = UserSchema.from_orm(db_user)

SQLModel 대안

  • SQLModel: FastAPI의 창시자 Sebastián Ramírez가 만든, Pydantic + SQLAlchemy를 합친 통합 라이브러리
  • 장점: 별도 연동 설정 불필요, from_orm이 기본
  • 단점: 커스터마이징 한계

CRUD 구현 예제

  1. Create 
    @app.post("/users/", response_model=UserSchema)
def create_user(user: UserSchema, db: Session = Depends(get_db)):
    db_user = UserORM(**user.model_dump())
    db.add(db_user); db.commit(); db.refresh(db_user)
    return db_user
  2. Read 
    @app.get("/users/{user_id}", response_model=UserSchema)
def read_user(user_id: int, db: Session = Depends(get_db)):
    return db.get(UserORM, user_id)
  3. Update 
    @app.put("/users/{user_id}", response_model=UserSchema)
def update_user(user_id: int, user: UserSchema, db: Session = Depends(get_db)):
    db_user = db.get(UserORM, user_id)
    for k, v in user.model_dump(exclude_unset=True).items():
        setattr(db_user, k, v)
    db.commit(); db.refresh(db_user)
    return db_user
  4. Delete 
    @app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.get(UserORM, user_id)
    db.delete(db_user); db.commit()
    return {"ok": True}

FastAPI 라우터와 의존성 주입

  • 라우터 구조: @app.get(), @app.post() 등 데코레이터로 엔드포인트 선언
  • 의존성 주입: Depends(get_db)로 DB 세션, 인증, 설정 등을 간편하게 삽입
def get_db():
    db = SessionLocal(); 
    try: yield db
    finally: db.close()

에러 처리 및 응답 모델

  • HTTPException 
    from fastapi import HTTPException
if not db_user:
    raise HTTPException(status_code=404, detail="User not found")
  • 응답 모델: response_model 파라미터로 자동 검증 및 문서화

설정 관리 및 환경 변수

  • pydantic-settings 
    from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    class Config:
        env_file = ".env"
settings = Settings()
  • .env 파일: 민감한 정보는 분리합니다.

테스트 작성 팁

  • pytest 사용 
    from fastapi.testclient import TestClient
client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"id":1,"name":"Alice"})
    assert response.status_code == 200
  • 테스트 DB: 별도 SQLite 인메모리 사용

성능 최적화 전략

  • 데이터베이스 튜닝: 인덱싱, 쿼리 최적화
  • 캐싱: Redis 활용, FastAPI의 BackgroundTask

배포 및 운영 고려사항

  • Docker 이미지 구성: Dockerfile과 docker-compose.yml
  • Alembic 마이그레이션: 
    alembic init alembic
alembic revision --autogenerate -m "init"
alembic upgrade head
  • CI/CD: GitHub Actions, Jenkins 등 연동

결론 및 요약

지금까지 FastAPI, Pydantic 2.x, 그리고 SQLAlchemy를 연동해 실무 API를 만드는 데 핵심이 되는 개념과 문법을 살펴보았습니다.

  • 핵심 요약:
    1. Pydantic으로 입력 검증 및 직렬화
    2. SQLAlchemy ORM과 from_orm 옵션으로 매끄러운 모델 변환
    3. FastAPI에서 의존성 주입과 자동 문서화 활용

FastAPI에 Pydantic, SQLAlchemy를 연동해, 견고하고 확장성 있는 Python 기반 API 서버를 구축해 보시길 바랍니다!

2025년 최신 npm 설치 가이드 – 윈도우, 맥, 리눅스 완벽 정복

 

Leave a Comment