FastAPI 서버개발, 비동기 데이터베이스 연동 (async SQLAlchemy, Tortoise ORM)

FastAPI는 비동기 지원을 기본으로 제공하여 높은 성능과 효율성을 자랑하는 웹 프레임워크입니다. 이러한 FastAPI의 장점은 비동기 데이터베이스를 통해 더욱 극대화할 수 있습니다. 본 강좌에서는 FastAPI와 함께 사용할 수 있는 두 가지 비동기 데이터베이스 ORM인 async SQLAlchemy와 Tortoise ORM을 이용한 서버 개발 방법을 알아보겠습니다.

1. FastAPI 소개

FastAPI는 Python으로 작성된 웹 프레임워크로, 빠르고 간편하게 RESTful API를 개발할 수 있도록 도와줍니다. FastAPI는 Starlette을 기반으로 하여 구현되었으며, 타입 힌트를 활용하여 자동으로 API 문서를 생성합니다. 이러한 특성 덕분에 FastAPI는 높은 생산성과 성능을 제공하며, 대규모 서비스와 실시간 API에 적합한 선택이 됩니다.

2. 비동기 프로그래밍 개념

비동기 프로그래밍은 태스크를 동시에 처리할 수 있도록 해주는 프로그래밍 모델입니다. 일반적인 동기 프로그래밍에서는 하나의 작업이 끝난 후 다음 작업을 시작하는 방식이지만, 비동기 프로그래밍에서는 작업을 기다리는 동안 다른 작업을 수행할 수 있습니다. 이로 인해 I/O 작업에 소요되는 시간을 줄일 수 있습니다.

3. async SQLAlchemy

3.1 소개

SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(Object Relational Mapping) 라이브러리입니다. async SQLAlchemy는 비동기 I/O를 지원하는 SQLAlchemy의 확장으로, FastAPI와 함께 사용할 수 있는 강력한 도구입니다.

3.2 설치

pip install sqlalchemy[asyncio] asyncpg

3.3 기본 사용법

async SQLAlchemy를 사용하여 FastAPI와 연결하는 기본적인 방법은 다음과 같습니다:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)
Base = declarative_base()

app = FastAPI()

# 데이터 모델 정의
class Item(Base):
    __tablename__ = 'items'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)

# 비동기 데이터베이스 세션 관리 커럽틴 함수
async def get_db():
    async with SessionLocal() as session:
        yield session

3.4 CRUD 생성

FastAPI에서 비동기 데이터베이스를 사용하여 CRUD(Create, Read, Update, Delete) API를 만드는 방법은 다음과 같습니다:

@app.post("/items/")
async def create_item(item: Item, db: AsyncSession = Depends(get_db)):
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    return await db.get(Item, item_id)

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, db: AsyncSession = Depends(get_db)):
    db_item = await db.get(Item, item_id)
    if db_item:
        db_item.name = item.name
        db_item.description = item.description
        await db.commit()
        await db.refresh(db_item)
        return db_item
    return {"error": "Item not found"}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int, db: AsyncSession = Depends(get_db)):
    db_item = await db.get(Item, item_id)
    if db_item:
        await db.delete(db_item)
        await db.commit()
        return {"message": "Item deleted"}
    return {"error": "Item not found"}

4. Tortoise ORM

4.1 소개

Tortoise ORM은 비동기 환경을 위해 설계된 ORM으로, 간결하고 사용하기 쉬운 API를 제공합니다. 비동기 기반의 FastAPI와 잘 어울립니다.

4.2 설치

pip install tortoise-orm aiosqlite

4.3 기본 사용법

FastAPI와 Tortoise ORM을 연동하기 위한 기본적인 설정은 다음과 같습니다:

from fastapi import FastAPI
from tortoise import fields
from tortoise.models import Model
from tortoise import Tortoise

app = FastAPI()

# 데이터 모델 정의
class Item(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255)
    description = fields.TextField()

# Tortoise 초기화
@app.on_event("startup")
async def startup():
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',
        modules={"models": ["__main__"]}
    )
    await Tortoise.generate_schemas()

@app.on_event("shutdown")
async def shutdown():
    await Tortoise.close_connections()

4.4 CRUD 생성

Tortoise ORM을 사용하여 CRUD API를 구현하는 방법은 다음과 같습니다:

@app.post("/items/")
async def create_item(item: Item):
    await item.save()
    return item

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return await Item.get(id=item_id)

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    db_item = await Item.get(id=item_id)
    if db_item:
        db_item.name = item.name
        db_item.description = item.description
        await db_item.save()
        return db_item
    return {"error": "Item not found"}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    db_item = await Item.get(id=item_id)
    if db_item:
        await db_item.delete()
        return {"message": "Item deleted"}
    return {"error": "Item not found"}

5. 결론

FastAPI는 비동기 처리를 기본으로 지원하여, 높은 성능의 API 서버를 손쉽게 구축할 수 있게 해줍니다. async SQLAlchemy와 Tortoise ORM은 각각의 장점을 가지고 있으며, 상황에 맞추어 적절한 선택을 할 수 있습니다. 본 강좌를 통해 FastAPI와 비동기 데이터베이스의 통합 방법을 배우고, 실제로 API를 구현해 보셨기를 바랍니다. FastAPI의 강력한 기능을 활용하여 훌륭한 웹 애플리케이션을 개발하시기 바랍니다.