DokumentePreiseBlog
Kontaktieren Sie uns
AnmeldenKOSTENLOS anmelden
Back to Tech Articles

FastAPI, SQLAlchemy 2.0, 및 Asyncpg를 활용한 고성능 비동기 API 구축

Aug. 10, 2025
# Misc

James Reed

Infrastructure Engineer · Leapcell

Dieser Beitrag ist auch in English, 日本語 und Deutsch verfügbar
Cover of "FastAPI, SQLAlchemy 2.0, 및 Asyncpg를 활용한 고성능 비동기 API 구축"

소개

웹 개발의 빠르게 진화하는 환경에서 성능이 뛰어나고 확장 가능한 애플리케이션을 구축하는 것은 매우 중요합니다. 특히 데이터베이스 상호 작용을 다룰 때 기존의 동기 I/O 작업은 종종 병목 현상을 일으켜 사용자 경험을 저하시키고 서버 처리량을 감소시킬 수 있습니다. 여기서 비동기 프로그래밍이 빛을 발하며, 애플리케이션이 논블로킹(non-blocking) 작업을 수행할 수 있게 하여 더 많은 동시 요청을 효율적으로 처리할 수 있는 패러다임 전환을 제공합니다. FastAPI는 내재된 비동기 기능을 통해 고성능 API 구축을 위한 프레임워크로 부상했습니다. 그러나 비동기 프레임워크를 사용하더라도 데이터베이스 계층은 올바르게 처리하지 않으면 여전히 동기 블로킹의 원인이 될 수 있습니다. 본 문서는 FastAPI, SQLAlchemy 2.0의 비동기 엔진, 그리고 asyncpg 드라이버의 강력한 시너지를 탐구하며, 비동기 데이터베이스 상호 작용의 잠재력을 최대한 활용하는 견고하고 고도로 동시적인 API를 구축하는 방법을 시연합니다.

비동기 데이터베이스 액세스를 위한 핵심 구성 요소

구현 세부 사항을 살펴보기 전에 FastAPI 애플리케이션 내에서 비동기 데이터베이스 액세스를 달성하는 데 관련된 주요 기술에 대한 명확한 이해를 확립해 보겠습니다.

  • FastAPI: Python 3.7+를 기반으로 표준 Python 타입 힌트를 사용하여 API를 구축하는 현대적이고 빠른(고성능) 웹 프레임워크입니다. 이 프레임워크의 네이티브 비동기 지원은 I/O 바운드 애플리케이션에 이상적인 선택입니다.
  • SQLAlchemy 2.0: Python을 위한 강력하고 유연한 ORM(객체 관계 매퍼)입니다. 2.0 버전은 특히 비동기 데이터베이스 작업에 대한 1급 지원을 도입하여 최신 비동기 프레임워크와 더 잘 맞습니다.
  • 비동기 엔진 (SQLA 2.0): SQLAlchemy의 새로운 create_async_engine 및 관련 비동기 구성을 나타냅니다. 이러한 비동기 구성 요소는 데이터베이스 응답을 기다리는 동안 이벤트 루프를 차단하는 대신 애플리케이션이 다른 작업으로 전환할 수 있게 하여 동시성을 향상시킵니다.
  • asyncpg: Python을 위한 빠른 PostgreSQL 데이터베이스 드라이버입니다. asyncio를 사용하여 처음부터 비동기 I/O를 위해 설계되었으며, 비동기 컨텍스트에서 사용할 때 기존 드라이버에 비해 우수한 성능을 제공합니다. SQLAlchemy의 비동기 PostgreSQL 백엔드에 권장되는 드라이버입니다.
  • async/await: 코루틴을 정의하고 실행하기 위한 Python의 키워드로, 실행을 일시 중지했다가 나중에 다시 시작할 수 있는 함수로, 논블로킹 I/O를 가능하게 합니다.

비동기 데이터베이스 상호 작용 원칙

SQLAlchemy 2.0 및 asyncpg를 사용한 비동기 데이터베이스 상호 작용의 핵심 원칙은 모든 데이터베이스 작업이 논블로킹되도록 보장하는 것입니다. async 함수가 데이터베이스 쿼리를 발급할 때, 데이터베이스 응답을 기다리는 대신 이벤트 루프에 제어권을 양보합니다. 그런 다음 이벤트 루프는 데이터베이스 응답이 준비될 때까지 다른 보류 중인 작업(예: 다른 들어오는 요청 처리)을 실행할 수 있습니다. 응답이 도착하면 원래 async 함수가 다시 시작됩니다.

비동기 엔진 및 세션 설정

시작하기 위해 SQLAlchemy를 asyncpg를 사용하여 비동기 기능을 사용하도록 구성해야 합니다.

먼저 필요한 라이브러리가 설치되었는지 확인하십시오:

pip install fastapi "uvicorn[standard]" sqlalchemy "asyncpg<0.29.0"

참고: SQLAlchemy 2.0.x 시리즈의 경우, asyncpg 버전 0.29.0 이상은 create_async_engine과 함께 문제를 일으킬 수 있습니다. 현재는 <0.29.0으로 고정하는 것이 더 안전합니다.

다음으로 비동기 엔진 및 세션 팩토리를 설정해 보겠습니다:

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base

# PostgreSQL 연결 문자열로 바꾸십시오
# 'postgresql+asyncpg' 접두사는 SQLAlchemy가 asyncpg를 사용하도록 지시합니다.
DATABASE_URL = "postgresql+asyncpg://user:password@host:port/dbname"

# 비동기 엔진 생성
# pool_pre_ping=True는 장기 실행 애플리케이션의 연결을 유지하는 데 도움이 됩니다.
engine = create_async_engine(DATABASE_URL, echo=True, pool_pre_ping=True)

# 비동기 세션메이커 구성
# expire_on_commit=False는 커밋 후 객체가 만료(분리)되는 것을 방지합니다.
# 이는 후속 작업을 위해 객체를 세션에 연결하여 유지하는 데 유용할 수 있습니다.
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

# 비동기 데이터베이스 세션을 얻기 위한 종속성
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

이 설정에서는:

  • create_async_engine은 asyncpg를 사용하여 PostgreSQL 데이터베이스에 연결된 비동기 엔진을 초기화합니다.
  • async_sessionmaker는 AsyncSession 객체를 생성하는 팩토리를 제공합니다.
  • get_db는 AsyncSession을 생성하는 FastAPI 종속성으로, 적절한 세션 라이프사이클 관리(생성 및 닫기)를 보장합니다.

모델 정의 및 마이그레이션 수행

동기 SQLAlchemy와 마찬가지로 declarative_base를 사용하여 모델을 정의합니다.

# models.py
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)

    def __repr__(self):
        return f"<Item(id={self.id}, name='{self.name}')>"

# 테이블 생성을 위한 예시 (일반적으로 Alembic을 사용하여 마이그레이션 수행)
async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 일반적으로 데이터베이스를 설정하기 위해 이 코드를 한 번 실행합니다.
# import asyncio
# async def main():
#     await create_db_and_tables()
# 
# if __name__ == "__main__":
#     asyncio.run(main())

FastAPI 엔드포인트와 통합

이제 이러한 구성 요소를 FastAPI 애플리케이션에 통합하여 비동기 CRUD 작업을 수행해 보겠습니다.

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

from database import get_db, create_db_and_tables, engine, AsyncSessionLocal
from models import Item
from schemas import ItemCreate, ItemResponse # 이러한 Pydantic 모델이 정의되었다고 가정

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # 예시: 시작 시 테이블 생성 (간단한 앱의 경우 프로덕션에서는 Alembic 사용)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("Database tables created (if not exist).")


@app.post("/items/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate, db: AsyncSession = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    await db.commit()  # 비동기 커밋
    await db.refresh(db_item) # 생성된 ID/기본값 등을 가져오기 위해 새로고침
    return db_item

@app.get("/items/", response_model=list[ItemResponse])
async def read_items(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
    # 비동기 쿼리 실행
    result = await db.execute(select(Item).offset(skip).limit(limit))
    items = result.scalars().all()
    return items

@app.get("/items/{item_id}", response_model=ItemResponse)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item).filter(Item.id == item_id))
    item = result.scalars().first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

# Pydantic 모델 예시 (schemas.py)
# from pydantic import BaseModel
# class ItemBase(BaseModel):
#     name: str
#     description: str | None = None
#
# class ItemCreate(ItemBase):
#     pass
#
# class ItemResponse(ItemBase):
#     id: int
#
#     class Config:
#         orm_mode = True # SQLAlchemy ORM 호환성을 위해

main.py 예시에서는:

  • startup_event는 conn.run_sync(Base.metadata.create_all)를 사용합니다. create_all은 Base.metadata의 동기 메서드이기 때문입니다. conn.run_sync는 이러한 동기 호출을 비동기 컨텍스트 내에서 다리 역할을 하며 스레드 풀에서 실행합니다. 이는 동기 작업을 비동기 흐름에 통합하는 일반적인 패턴입니다.
  • 모든 데이터베이스 작업(db.add, db.commit, db.execute)은 await로 접두사가 붙어 비동기 특성을 나타냅니다.
  • select(Item)은 SQLAlchemy 2.0에서 쿼리를 구성하는 최신 방식이며, 더 명시적인 제어와 더 나은 타입 추론을 제공합니다.
  • result.scalars().all()은 쿼리에서 모든 스칼라 결과를 효율적으로 검색합니다.

실제 시나리오에서의 응용

이 패턴은 다음 예에 매우 적합합니다:

  • 마이크로서비스: 데이터베이스와 상호 작용하는 독립적이고 고성능인 서비스 구축.
  • 실시간 API: 채팅 애플리케이션, 게임 백엔드 또는 금융 거래 플랫폼과 같이 낮은 지연 시간과 높은 동시성이 중요한 경우.
  • 데이터 수집 API: 대량의 데이터를 효율적으로 수신하고 저장.
  • 웹훅: 블로킹 없이 외부 서비스에서 들어오는 데이터를 처리.

SQLAlchemy의 비동기 엔진 내에서 asyncpg를 활용함으로써 FastAPI 애플리케이션은 기존의 동기 설정에 비해 훨씬 높은 동시성과 처리량을 달성할 수 있습니다. asyncpg의 PostgreSQL 상호 작용에 대한 고도로 최적화된 C 구현과 asyncio의 이벤트 루프 관리 기능은 유휴 시간을 최소화하여 API가 더 많은 동시 클라이언트 연결을 효율적으로 처리할 수 있도록 합니다.

결론

FastAPI, SQLAlchemy 2.0의 비동기 엔진, 그리고 asyncpg 드라이버의 조합은 Python에서 고성능의 확장 가능한 웹 API를 구축하기 위한 강력하고 현대적인 스택을 나타냅니다. 프레임워크 수준부터 데이터베이스 드라이버까지 비동기 프로그래밍을 채택함으로써 개발자는 부하가 많이 걸릴 때 신속하게 응답하고 시스템 리소스를 효율적으로 사용하는 애플리케이션을 만들 수 있습니다. 이 접근 방식은 사용자 경험을 향상시킬 뿐만 아니라 향후 확장성을 위한 견고한 기반을 제공합니다. 이러한 도구를 사용하여 비동기 네이티브 애플리케이션을 구축하는 것은 현대적이고 효율적인 웹 개발을 향한 중요한 단계입니다.

Share this article

HackerNews

More Posts from Leapcell

Popular Posts