FastAPI와 AsyncPG를 활용한 고성능 비동기 데이터베이스 처리 실전 적용기

FastAPI를 사용하는 가장 큰 이유는 Python의 비동기(Asynchronous) 처리를 통해 높은 동시성을 확보하기 위함입니다. 하지만 아무리 웹 프레임워크가 빨라도, 데이터베이스 드라이버가 동기(Blocking) 방식으로 동작한다면 그 이점은 반감됩니다.

이번 Swifty Backend API 개발 과정에서 기존의 psycopg2 대신 asyncpg를 도입하여 완전한 비동기 스택을 구축했습니다. 단순한 설치 가이드보다는 실제 프로젝트에 적용하며 고민했던 설계 패턴과 성능 최적화 팁 위주로 정리했습니다.

왜 AsyncPG인가?

가장 결정적인 이유는 성능과 구조적 일관성입니다. 사용자가 식단 사진을 업로드하면 Gemini API로 분석을 요청하고, 동시에 DB에 로그를 남겨야 하는 시나리오가 있었습니다. I/O 바운드 작업이 많은 이 상황에서 psycopg2 같은 동기 드라이버는 스레드를 점유해버려 병목을 유발합니다.

AsyncPG는 asyncio에 최적화되어 있고, 벤치마크상으로도 압도적인 속도를 보여줍니다. 또한 PostgreSQL의 파이프라이닝과 Prepared Statement 기능을 적극적으로 활용합니다.

1. Connection Pool의 생명주기 관리

매 요청마다 DB 연결을 맺고 끊는 오버헤드를 줄이기 위해 Connection Pool을 사용합니다. FastAPI에서는 lifespan 이벤트 핸들러를 통해 애플리케이션의 시작과 종료 시점에 맞춰 Pool을 안전하게 관리할 수 있습니다.

database.py에서 전역 Pool을 관리하는 패턴입니다.

# database.py
import asyncpg
from contextlib import asynccontextmanager
from fastapi import FastAPI
 
pool: asyncpg.Pool = None
 
async def get_pool():
    global pool
    return pool
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 앱 시작 시 Pool 생성
    global pool
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=5,
        max_size=20, # 트래픽에 맞춰 조정
        server_settings={'search_path': 'swifty_app'}
    )
    yield
    # 앱 종료 시 Pool 정리
    if pool:
        await pool.close()

이 방식의 장점은 전역 변수 pool을 통해 어디서든 연결을 가져올 수 있으면서도, 연결의 생성과 소멸이 앱의 생명주기와 정확히 일치한다는 점입니다.

2. 트랜잭션과 Bulk Insert 최적화

식단 기록 기능은 '식단 정보(Food Log)'와 '재료 목록(Ingredients)'을 저장해야 합니다. 이 두 작업은 원자성(Atomicity)이 보장되어야 합니다. AsyncPG는 async with connection.transaction(): 컨텍스트 매니저를 통해 이를 직관적으로 처리합니다.

특히 재료 목록처럼 여러 행을 한 번에 넣을 때는 루프를 돌며 execute를 호출하는 대신 executemany를 사용하여 네트워크 왕복 비용을 획기적으로 줄였습니다.

# crud.py
async def log_food(request: FoodLogRequest):
    pool = await get_pool()
    async with pool.acquire() as connection:
        # 트랜잭션 시작: 하나라도 실패하면 전체 롤백
        async with connection.transaction():
            # 1. 메인 식단 로그 저장
            food_log_id = await connection.fetchval(
                """
                INSERT INTO food_logs (user_key, meal_type, date)
                VALUES ($1, $2, $3) RETURNING id
                """,
                request.userKey, request.mealType, request.date
            )
 
            # 2. 재료 목록 Bulk Insert
            # executemany를 사용하여 다량의 데이터를 한 번의 통신으로 처리
            if request.ingredients:
                await connection.executemany(
                    """
                    INSERT INTO food_ingredients (food_log_id, name, color) 
                    VALUES ($1, $2, $3)
                    """,
                    [(food_log_id, ing.name, ing.color) for ing in request.ingredients]
                )
                
            return {"id": str(food_log_id)}

3. N+1 문제 해결과 JSON Aggregation

ORM을 사용하지 않고 Raw SQL을 사용할 때의 장점은 쿼리 최적화가 자유롭다는 점입니다. "오늘의 식단 목록"과 "각 식단의 재료"를 조회할 때, 일반적인 방식이라면 식단 목록을 조회하고 각 식단마다 재료를 조회하는 쿼리를 날려 N+1 문제가 발생합니다.

PostgreSQL의 json_agg와 json_build_object를 활용하면 조인된 데이터를 JSON 형태로 말아서 단 한 번의 쿼리로 가져올 수 있습니다. AsyncPG는 JSONB 타입을 Python의 dict/list로 자동 변환해주므로 추가적인 파싱 로직도 필요 없습니다.

# crud.py (최적화된 조회)
async def get_today_food_logs(userKey: str):
    query = """
        SELECT
            fl.id,
            fl.meal_type,
            fl.date,
            -- 연관된 재료들을 JSON 배열로 집계
            COALESCE(
                (SELECT json_agg(json_build_object('name', fi.name, 'color', fi.color))
                 FROM food_ingredients fi
                 WHERE fi.food_log_id = fl.id),
                '[]'::json
            ) AS ingredients
        FROM food_logs fl
        WHERE fl.user_key = $1 AND fl.date = CURRENT_DATE
    """
    
    pool = await get_pool()
    async with pool.acquire() as connection:
        rows = await connection.fetch(query, userKey)
        
        # rows['ingredients']는 이미 Python list로 변환되어 있음
        return [
            FoodLogResponse(
                id=str(row['id']),
                mealType=row['meal_type'],
                ingredients=row['ingredients'] # 별도 파싱 불필요
            ) for row in rows
        ]

4. 도입 시 주의했던 점 (Gotchas)

실전 적용 과정에서 psycopg2나 다른 ORM에 익숙했다면 헷갈릴 수 있는 포인트들입니다.

파라미터 플레이스홀더: Python 표준인 %s나 ?를 사용하지 않고, PostgreSQL Native 방식인 $1, $2, $3... 을 사용합니다. 인덱스가 1부터 시작한다는 점을 주의해야 합니다.
엄격한 타입 캐스팅: AsyncPG는 타입에 민감합니다. 예를 들어, DB 컬럼이 DATE 타입이라면 문자열 '2023-01-01'을 그냥 넘기면 에러가 날 수 있습니다. Python의 datetime.date 객체로 명확하게 변환해서 넘기는 것이 안전합니다.
커넥션 관리: pool.acquire()로 얻은 커넥션은 반드시 반환해야 합니다. async with 구문을 사용하면 예외가 발생해도 안전하게 반환되므로, 수동으로 release를 호출하는 것보다 컨텍스트 매니저 패턴을 지향해야 합니다.

FastAPI와 AsyncPG의 조합은 Python으로 고성능 백엔드를 구축할 때 가장 강력한 선택지 중 하나입니다. ORM의 편리함을 일부 포기해야 하지만, lifespan을 통한 깔끔한 리소스 관리, executemany를 통한 배치 처리, 그리고 PostgreSQL의 기능을 100% 활용하는 쿼리 작성을 통해 서비스의 안정성과 응답 속도를 크게 개선할 수 있었습니다.

FastAPI와 AsyncPG로 PostgreSQL 연동하기