FastAPI에서 SQLModel 및 Tortoise ORM을 사용한 비동기 데이터베이스 작업

소개

빠르게 발전하는 백엔드 개발 환경에서 고성능의 확장 가능한 웹 서비스를 구축하는 것은 매우 중요합니다. 비동기 프로그래밍은 애플리케이션이 기본 스레드를 차단하지 않고 수많은 동시 요청을 처리할 수 있도록 하여 이러한 목표를 달성하는 초석으로 부상했습니다. FastAPI는 본질적으로 비동기 작업을 지원하고 직관적인 디자인으로 현대 API 구축에 빠르게 선호되는 프레임워크가 되었습니다. 그러나 비동기 웹 프레임워크의 이점도 데이터베이스 상호 작용이 동기식으로 유지되면 크게 저해될 수 있습니다. 이러한 병목 현상은 비동기 ORM 및 데이터베이스 라이브러리가 해결하고자 하는 문제입니다. 이 글에서는 FastAPI에 두 가지 주요 비동기 데이터베이스 도구인 SQLModel과 Tortoise ORM을 통합하여 진정한 논블로킹(non-blocking) 데이터베이스 작업을 활용하는 방법을 탐색하여 백엔드 서비스의 전반적인 효율성과 응답성을 향상시키는 방법을 살펴봅니다.

핵심 개념 및 원칙

구현 세부 사항으로 들어가기 전에 FastAPI에서 비동기 데이터베이스 작업을 이해하는 데 중요한 몇 가지 기본 용어를 명확히 하겠습니다.

• 비동기 프로그래밍: 전체 애플리케이션을 중단시키지 않고 오래 실행되는 작업을 실행할 수 있는 프로그래밍 패러다임입니다. Python에서는 주로 async/await 키워드와 asyncio 라이브러리를 사용하여 협업적 멀티태스킹을 가능하게 합니다.
• ORM (Object-Relational Mapping): 개발자가 객체 지향 패러다임을 사용하여 데이터베이스와 상호 작용할 수 있게 해주는 기술입니다. ORM을 사용하면 원시 SQL 쿼리를 작성하는 대신 선호하는 프로그래밍 언어의 객체로 데이터베이스 레코드를 조작할 수 있습니다. 이러한 추상화는 데이터베이스 작업을 단순화하고 코드 가독성을 향상시키며 SQL 주입을 방지하여 보안을 강화합니다.
• FastAPI: Python 타입 힌트를 기반으로 하는 Python 3.7+를 사용하여 API를 구축하기 위한 현대적이고 빠른(고성능) 웹 프레임워크입니다. asyncio와 깊이 통합되어 비동기 웹 서비스에 이상적입니다.
• SQLModel: SQL 데이터베이스와 상호 작용하기 위한 Python 라이브러리로, 'SQL 및 Python 우선'으로 설계되었습니다. Pydantic과 SQLAlchemy를 기반으로 구축되어 데이터 유효성 검사를 위한 Pydantic 모델과 데이터베이스 상호 작용을 위한 SQLAlchemy 모델 역할을 동시에 수행하는 모델을 더 간단하고 직관적으로 정의할 수 있습니다. SQLAlchemy의 비동기 엔진을 통해 본질적으로 비동기 작업을 지원합니다.
• Tortoise ORM: asyncio 및 uvloop용으로 특별히 설계된 사용하기 쉬운 Python 비동기 ORM입니다. 비동기 특성을 유지하면서 모델 정의, 쿼리 수행 및 데이터베이스 마이그레이션 관리를 위한 간단한 API를 제공합니다.

FastAPI와 비동기 ORM을 사용하는 원칙은 간단합니다. FastAPI 애플리케이션이 데이터베이스와 통신해야 할 때, 동기적으로 데이터베이스 응답을 기다리는 대신 ORM을 사용하면 애플리케이션이 다른 작업으로 전환할 수 있습니다. 데이터베이스 작업이 완료되면 애플리케이션은 원래 요청 처리를 재개할 수 있습니다. 이러한 논블로킹 동작은 데이터베이스 호출과 같은 I/O 바운드 작업에서 중요하며, 과부하 상태에서 서버가 응답하지 않게 되는 것을 방지합니다.

비동기 데이터베이스 작업 구현

SQLModel과 Tortoise ORM을 FastAPI 애플리케이션에 통합하여 비동기 데이터베이스 상호 작용을 수행하는 방법을 살펴보겠습니다. 시연을 위해 간단한 "Hero" 모델을 사용합니다.

SQLModel 사용

SQLModel은 Pydantic 모델과 SQLAlchemy를 결합하여 데이터를 정의하는 강력하고 우아한 방법을 제공합니다.

설정

먼저 필요한 패키지를 설치합니다.

pip install fastapi "uvicorn[standard]" sqlmodel "psycopg2-binary" # 또는 asyncpg (async용)

PostgreSQL을 데이터베이스로 사용합니다. PostgreSQL에 대한 비동기 작업을 위해 asyncpg는 psycopg2-binary보다 선호되는 경우가 많지만, psycopg2-binary는 비동기 래퍼와 함께 사용할 수 있습니다. 여기서는 네이티브 비동기 경험을 위해 asyncpg를 사용합니다.

pip install fastapi "uvicorn[standard]" sqlmodel asyncpg

데이터베이스 구성 및 모델 정의

from typing import Optional, List
from sqlmodel import Field, SQLModel, Session, create_engine
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException, status

# 데이터베이스 URL, 필요에 따라 조정
DATABASE_URL = "postgresql+asyncpg://user:password@host:port/dbname"

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass

class HeroPublic(HeroBase):
    id: int

# 비동기 엔진
engine = create_engine(DATABASE_URL, echo=True)

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# 데이터베이스 세션을 얻기 위한 종속성
async def get_session():
    async with Session(engine) as session:
        yield session

# FastAPI 애플리케이션 설정
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

FastAPI 엔드포인트

이제 Hero 모델에 대한 몇 가지 기본 CRUD 엔드포인트를 만들어 보겠습니다.

from sqlmodel import select

@app.post("/heroes/", response_model=HeroPublic)
async def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=List[HeroPublic])
async def read_heroes(offset: int = 0, limit: int = Field(default=100, le=100), session: Session = Depends(get_session)):
    heroes = (await session.exec(select(Hero).offset(offset).limit(limit))).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
async def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.put("/heroes/{hero_id}", response_model=HeroPublic)
async def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroCreate):
    db_hero = (await session.get(Hero, hero_id))
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.model_dump(exclude_unset=True)
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.delete("/heroes/{hero_id}")
async def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    await session.delete(hero)
    await session.commit()
    return {"ok": True}

여기서 핵심 사항은 다음과 같습니다.

• create_engine은 비동기 드라이버를 위해 postgresql+asyncpg로 설정됩니다.
• 테이블을 비동기적으로 생성하기 위한 async with engine.begin() 및 await conn.run_sync()입니다.
• 비동기 데이터베이스 세션을 관리하기 위한 async with Session(engine) as session:입니다.
• 비동기 쿼리를 위한 await session.exec() 및 await session.get()입니다.
• 변경 사항을 저장하고 객체를 새로 고침하기 위한 await session.commit() 및 await session.refresh()입니다.

Tortoise ORM 사용

Tortoise ORM은 처음부터 비동기 작업을 위해 설계되었으며 고유한 쿼리 구문으로 보다 전통적인 ORM 경험을 제공합니다.

설정

Tortoise ORM과 선택한 비동기 데이터베이스 드라이버(예: PostgreSQL의 asyncpg)를 설치합니다.

pip install fastapi "uvicorn[standard]" tortoise-orm asyncpg

데이터베이스 구성 및 모델 정의

from typing import Optional, List
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from tortoise import fields, models
from tortoise.contrib.fastapi import register_tortoise
from tortoise.exceptions import DoesNotExist

# 데이터베이스 구성
TORTOISE_CONFIG = {
    "connections": {"default": "postgresql://user:password@host:port/dbname"},
    "apps": {
        "models": {
            "models": ["main"], # 모델이 이 파일에 있다고 가정
            "default_connection": "default",
        }
    }
}

# Tortoise ORM Hero 모델
class Hero(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, unique=True, index=True)
    secret_name = fields.CharField(max_length=255)
    age = fields.IntField(null=True, index=True)

    class Meta:
        table = "heroes"

    def __str__(self):
        return self.name

# 요청/응답 유효성 검사를 위한 Pydantic 모델
class HeroIn(BaseModel):
    name: str
    secret_name: str
    age: Optional[int] = None

class HeroOut(BaseModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None

async def init_db(app: FastAPI):
    register_tortoise(
        app,
        config=TORTOISE_CONFIG,
        generate_schemas=True, # 존재하지 않으면 테이블 생성
        add_exception_handlers=True,
    )

app = FastAPI()

# 앱 시작 시 Tortoise ORM 등록
@app.on_event("startup")
async def startup_event():
    await init_db(app)

FastAPI 엔드포인트

@app.post("/heroes/", response_model=HeroOut)
async def create_hero(hero_in: HeroIn):
    hero = await Hero.create(**hero_in.model_dump())
    return await HeroOut.from_tortoise_orm(hero)

@app.get("/heroes/", response_model=List[HeroOut])
async def get_heroes(offset: int = 0, limit: int = 100):
    heroes = await Hero.all().offset(offset).limit(limit)
    return [await HeroOut.from_tortoise_orm(hero) for hero in heroes]

@app.get("/heroes/{hero_id}", response_model=HeroOut)
async def get_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.put("/heroes/{hero_id}", response_model=HeroOut)
async def update_hero(hero_id: int, hero_in: HeroIn):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.update_from_dict(hero_in.model_dump(exclude_unset=True))
        await hero.save()
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.delete("/heroes/{hero_id}", status_code=204)
async def delete_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.delete()
        return {"message": "Hero deleted successfully"}
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

Tortoise ORM을 사용하면:

• register_tortoise는 데이터베이스 연결 및 스키마 생성을 처리합니다.
• 모델은 models.Model에서 상속하고 속성을 정의하기 위해 fields를 사용합니다.
• await Hero.create(), await Hero.all(), await Hero.get(), await hero.save(), await hero.delete()는 모두 비동기 작업입니다.
• HeroOut.from_tortoise_orm()은 ORM 인스턴스를 Pydantic 모델로 변환하기 위해 tortoise.contrib.pydantic에서 제공하는 편리한 메서드입니다.

애플리케이션 시나리오

SQLModel과 Tortoise ORM은 다음과 같은 시나리오에서 모두 뛰어납니다.

• 높은 동시성이 예상되는 경우: 많은 수의 사용자 또는 요청은 효율적인 I/O 처리를 필요로 합니다.
• 마이크로서비스 아키텍처: 분리된 서비스는 해당 데이터베이스와의 빠르고 논블로킹 통신에서 이점을 얻는 경우가 많습니다.
• 실시간 애플리케이션: 실시간 데이터 또는 업데이트를 제공하는 API는 낮은 지연 시간의 데이터베이스 작업을 요구합니다.
• 현대 Python 백엔드: FastAPI와 같은 프레임워크와의 통합은 자연스럽게 해당 비동기 기능을 활용합니다.

SQLModel의 강점은 Pydantic과의 긴밀한 통합에 있으며, 데이터 유효성 검사 및 직렬화가 중요한 프로젝트와 데이터의 단일 진실 공급원을 원하는 경우에 탁월합니다. SQLAlchemy의 강력한 기반 위에 구축되어 고급 쿼리 기능을 제공합니다.

Tortoise ORM의 강점은 asyncio 전용으로 설계된 간단한 API와 비동기 ORM에 대한 학습 곡선이 더 완만한 점입니다. 쿼리에 대한 자체 고유 구문을 가진 독립적인 ORM 솔정을 선호하는 경우 특히 매력적입니다.

결론

SQLModel 또는 Tortoise ORM과 같은 ORM을 사용하여 FastAPI 애플리케이션에 비동기 데이터베이스 작업을 통합하는 것은 성능이 뛰어나고 확장 가능한 웹 서비스를 구축하는 중요한 단계입니다. 두 도구 모두 I/O 병목 현상을 제거하는 강력한 비동기 기능을 제공하여 과부하 상태에서도 애플리케이션의 응답성을 보장합니다. SQLModel은 Pydantic과 SQLAlchemy의 강력한 기능을 갖춘 통합 모델 정의를 제공하며, Tortoise ORM은 간결한 asyncio 네이티브 경험을 제공합니다. 둘 사이의 선택은 종종 특정 프로젝트 요구 사항, 팀 친숙도 및 각 API 스타일에 대한 선호도에 달려 있지만, 어느 것을 선택하든 FastAPI 애플리케이션의 데이터베이스 상호 작용 효율성을 크게 향상시킬 것입니다. 비동기 ORM을 채택함으로써 FastAPI의 기능을 최대한 활용하고 진정으로 논블로킹이고 고성능인 백엔드를 제공할 수 있습니다.