Python 비동기 DB 쿼리 최적화 - asyncpg로 마이그레이션

문제 상황

서비스 트래픽이 증가하면서 DB 연결이 병목이 되기 시작했다. FastAPI를 사용하고 있지만 psycopg2로 동기 방식으로 쿼리를 처리하고 있어서, 비동기의 이점을 제대로 살리지 못하고 있었다.

기존 코드는 이런 형태였다.

@app.get("/users/{user_id}")
def get_user(user_id: int):
    conn = psycopg2.connect(DATABASE_URL)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    result = cursor.fetchone()
    conn.close()
    return result

동시 요청이 들어오면 각 요청이 DB 응답을 기다리는 동안 블로킹되어 전체 처리량이 떨어졌다.

asyncpg 도입

asyncpg는 PostgreSQL을 위한 고성능 비동기 드라이버다. psycopg3도 비동기를 지원하지만, asyncpg가 성능 면에서 더 우수하다는 벤치마크를 확인했다.

먼저 커넥션 풀을 설정했다.

import asyncpg
from contextlib import asynccontextmanager

db_pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global db_pool
    db_pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=10,
        max_size=50
    )
    yield
    await db_pool.close()

app = FastAPI(lifespan=lifespan)

엔드포인트는 이렇게 변경했다.

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return dict(row) if row else None

성능 개선 결과

locust로 부하 테스트를 돌려본 결과:

• 동시 100 사용자 기준 RPS: 230 → 680
• P95 응답 시간: 420ms → 145ms

특히 여러 쿼리를 순차적으로 실행하는 경우 asyncio.gather로 병렬화할 수 있어서 효과가 더 컸다.

주의사항

placeholder가 %s에서 $1, $2로 바뀌는 점을 놓쳐서 초반에 오류가 좀 나왔다. 또한 asyncpg는 결과를 Record 타입으로 반환하므로 dict 변환이 필요한 경우가 있다.

마이그레이션 자체는 하루 정도 걸렸고, 성능 개선 효과가 확실해서 만족스러웠다.