FastAPI + PostgreSQL DB Connection Pool로 안정성 개선하기

들어가며

이 글에서 나타나는 모든 코드, 네이밍 등은 사내 보안상 핵심 개념을 바탕으로 임의로 재작성한 것이다.

사내 AI 코드 어시스턴트를 개발하면서 DB 연결 방식에 문제가 있다는 걸 알게 됐다. 요청이 들어올 때마다 DB 연결을 새로 만들고, 응답 후 끊는 방식이었다. 동시 사용자가 늘어나면 언젠가 문제가 생길 게 뻔했다. 이번 글은 DB Connection Pool을 적용하면서 배운 것들을 정리한 글이다.

---

프로젝트 구조

서버는 FastAPI 기반 마이크로서비스 구조로 되어있다.

gateway        → JWT 인증, 라우팅
member_svr     → 사용자 관리, 로그인
domain_svr     → 코드 어시스턴트 tool
model_svr      → vLLM 모델 호출

DB 연결은 공통 유틸 패키지에서 관리하고 있었고, 아래처럼 요청마다 연결을 생성/종료하는 방식이었다.

@contextmanager
def get_pg_db():
    conn = psycopg2.connect(DATABASE_URL)  # 요청마다 새 연결
    try:
        yield conn.cursor(cursor_factory=RealDictCursor), conn
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()  # 요청마다 연결 종료

---

문제점

DB 연결 하나를 만드는 데 보통 50~100ms가 걸린다. 동시 사용자가 10명이면 이 비용이 10번 반복된다.

[ 기존 방식 ]
요청 A → connect(50~100ms) → 쿼리(20ms) → close()
요청 B → connect(50~100ms) → 쿼리(20ms) → close()
요청 C → connect(50~100ms) → 쿼리(20ms) → close()

또한 JWT 인증을 담당하는 jwt_handler.py도 모든 요청마다 asyncpg로 직접 연결하고 있었다. 인증은 모든 API 요청에서 실행되기 때문에 사실상 이쪽이 더 큰 병목이었다.

async def verify_token(token: str) -> dict:
    conn = await asyncpg.connect(DATABASE_URL)  # 모든 요청마다 새 연결
    try:
        row = await conn.fetchrow(...)
    finally:
        await conn.close()

---

해결 방법 — Connection Pool

Connection Pool은 미리 DB 연결을 여러 개 만들어두고, 요청이 올 때 빌려주고, 끝나면 반납받아 재사용하는 방식이다.

[ Pool 적용 후 ]
앱 시작 → DB 연결 2개 미리 생성

요청 A → getconn(1ms 미만) → 쿼리(20ms) → putconn() (반납)
요청 B → getconn(1ms 미만) → 쿼리(20ms) → putconn() (반납)
요청 C → getconn(1ms 미만) → 쿼리(20ms) → putconn() (반납)

연결 생성 비용이 앱 시작 시 한 번으로 줄어든다.

---

구현

1. pg_connection.py — psycopg2 Pool

from psycopg2.pool import ThreadedConnectionPool

_pool: ThreadedConnectionPool | None = None

def init_pool(minconn: int = 2, maxconn: int = 8, app_name: str = "app"):
    global _pool
    _pool = ThreadedConnectionPool(
        minconn=minconn,
        maxconn=maxconn,
        dsn=DATABASE_URL + f"?application_name={app_name}"
    )

def close_pool():
    global _pool
    if _pool:
        _pool.closeall()

@contextmanager
def get_pg_db():
    global _pool
    if _pool is None:
        # Pool 미초기화 시 fallback
        conn = psycopg2.connect(DATABASE_URL)
        try:
            yield conn.cursor(cursor_factory=RealDictCursor), conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            raise
        finally:
            conn.close()
        return

    conn = _pool.getconn()       # Pool에서 연결 빌려옴
    try:
        yield conn.cursor(cursor_factory=RealDictCursor), conn
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        _pool.putconn(conn)      # Pool에 반납 (close 하지 않음)

2. jwt_handler.py — asyncpg Pool

import asyncpg

_async_pool: asyncpg.Pool | None = None

async def init_async_pool(minsize: int = 2, maxsize: int = 8, app_name: str = "app"):
    global _async_pool
    _async_pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=minsize,
        max_size=maxsize,
        server_settings={"application_name": app_name}
    )

async def close_async_pool():
    global _async_pool
    if _async_pool:
        await _async_pool.close()

async def verify_token(token: str) -> dict:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    user_id = payload.get("sub")

    async with _async_pool.acquire() as conn:   # Pool에서 빌려서 사용
        row = await conn.fetchrow(
            "SELECT user_id, current_token FROM users WHERE user_id = $1",
            user_id
        )
    ...

3. FastAPI lifespan — 앱 시작/종료 시 Pool 관리

Pool은 정의만 해두면 안 되고, 앱 시작 시 실제로 초기화해줘야 한다. FastAPI의 lifespan을 사용하면 앱 시작/종료 시점에 코드를 실행할 수 있다.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_pool(minconn=2, maxconn=8, app_name="member_svr")
    await init_async_pool(minsize=2, maxsize=8, app_name="member_svr_jwt")
    yield   # ← 앱이 여기서 실제로 동작
    close_pool()
    await close_async_pool()

app = FastAPI(title="LLM Proxy", lifespan=lifespan)

yield 이전 코드는 앱 시작 시, yield 이후 코드는 앱 종료 시 실행된다.

---

psycopg2(동기) vs asyncpg(비동기)를 함께 쓰는 이유

처음에는 왜 두 가지를 같이 쓰냐는 의문이 들었다.

• jwt_handler.py는 원래부터 asyncpg로 작성되어 있었다.
• 서비스 파일들(user_service.py, dictionary_service.py 등)은 전부 동기(def) 방식이었다.

asyncpg로 통일하려면 서비스 파일 전체를 async def로 바꿔야 한다. 하지만 초기 테스터가 10명 규모인 지금 당장 전부 바꾸는 건 리스크가 크다고 판단했다. 한 번에 갈아엎기보다는 현재 시스템의 부하와 병목을 모니터링하면서 점진적으로 리팩토링하는 방향을 선택했고, 그 전까지는 동기 방식인 부분은 그대로 유지하기로 했다.

---

workers 수와 Pool 크기 계산

gunicorn으로 운영 중이고, 서비스별 workers 설정은 아래와 같다.

서비스 workers

member_svr	2
domain_svr	2
model_svr	64

Pool은 프로세스(worker)마다 독립적으로 생성된다. 따라서 총 연결 수 = workers × maxconn이 된다.

member_svr:  workers 2 × maxconn 8  = 최대 16개
domain_svr:  workers 2 × maxconn 8  = 최대 16개
model_svr:   workers 64 × maxconn 2 = 최대 128개 (JWT만 사용)

PostgreSQL 기본 max_connections=100을 초과하지 않도록 model_svr은 minsize=1, maxsize=2로 설정했다.

---

모니터링 — application_name 설정

Pool이 실제로 동작하는지 확인하기 위해 application_name을 설정했다. PostgreSQL의 pg_stat_activity 뷰에서 서비스별 연결 현황을 볼 수 있다.

SELECT application_name, count(*), state
FROM pg_stat_activity
WHERE datname = 'your_db'
GROUP BY application_name, state
ORDER BY application_name;

적용 결과:

application_name count state

domain_svr	4	idle
domain_svr_jwt	4	idle
member_svr	4	idle
member_svr_jwt	4	idle
model_svr_jwt	64	idle

idle 상태로 연결이 유지되고 있는 것을 확인할 수 있다. 부하 테스트 중에는 active 상태로 연결이 늘어났다가, 끝나면 다시 idle로 돌아오는 것도 확인했다.

---

트러블슈팅 — TooManyConnectionsError

처음 적용했을 때 model_svr에서 아래 에러가 발생했다.

asyncpg.exceptions.TooManyConnectionsError: sorry, too many clients already

원인은 model_svr의 workers가 64개인데, minsize=2로 설정하면 앱 시작 시에만 64 × 2 = 128개 연결이 생겨서 PostgreSQL의 max_connections=100을 초과했기 때문이다.

minsize=1, maxsize=2로 줄여서 해결했다.

---

마치며

이번 작업의 핵심은 극적인 속도 향상보다는 안정성이다.

항목 이전 이후

DB 연결 생성	요청마다	앱 시작 시 한 번
동시 요청 처리	불안정	Pool로 안정적 관리
연결 수 제어	불가	maxconn으로 제한
연결 모니터링	불가	application_name으로 가능
앱 종료 시 연결 정리	미보장	lifespan으로 자동 정리

사용자가 10명에서 50명, 100명으로 늘어날 때 버티는 힘을 키우는 작업이었다. Python과 FastAPI를 많이 사용해보지 않아서 contextmanager, asynccontextmanager, lifespan 등 낯선 개념들을 많이 접했는데, 직접 적용해보면서 자연스럽게 이해할 수 있었다.