FastAPI + SQLAlchemy: DB Connection이 지속적으로 생기는 문제

FastAPI는 일반적으로 SQLAlchemy와 함께 사용하며,

다수의 커넥션풀을 유지하기 위해서 QueuePool을 사용한다.

(비동기 데이터베이스 라이브러리 사용시에는 QueuePool이 아닌 async에 맞는 풀 사용)

# database.py

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
from sqlalchemy.orm import sessionmaker

sync_engine = create_engine(
    SYNC_DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=5,
    pool_timeout=30,
    pool_recycle=3600
)

SyncSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=sync_engine,
    expire_on_commit=True
)

poolclass는 QueuePool, NullPool, StaticPool 등이 있으며,

 

1. 아래와 같은 방식으로 db 세션을 가져오는 generator를 생성해서 의존성 주입을 통해 사용하거나

2. with 구문을 쓰는 context manager를 통해서 사용한다.

 

1.

# database.py

from typing import Annotated
from fastapi import Depends
from sqlalchemy.orm import Session


async def get_db_sync():
    with SyncSessionLocal() as session:
        yield session

db_session_sync = Annotated[Session, Depends(get_db_sync)]


# main.py

from sqlalchemy.future import select
from database import db_session_sync
from models import User


async def async_func(db: db_session_sync):
    result = db.execute(select(User)).all()
    global count
    print(len(result))

 

 

2.

async def read_users():
    with SyncSessionLocal() as db:
        users = db.execute(select(User)).all()
    return users

 

위와 같이 DB 연결을 위해 session을 만든 후, db에 query 등을 날리면,

DB와의 connection이 생기게 된다. DB에서 connection을 끊지 않고 있으면,  다음 query를 날릴때 connection이 없었을 때보다 더욱 빠르게 응답을 받을 수 있다.

 

그러나 db에서 유지할 수 있는 connection은 한계가 있다.

 

현재 프로젝트에서 문법에 맞게 db.close()를 해주거나, with구문을 쓰는 context manager를 통해서, 또는 의존성 함수로 db를 받아서 했음에도 커넥션이 계속 늘어나는 경우가 있었다.

물론, db 세션을 잘 닫고, 다른 곳에 문제가 없으면 기존에 이용중이던 connection 을 놔두고 새로운 connection을 여는 일은 없어야 할 것인데, 프로젝트에서 지속적인 connection이 생기고, 사용중인 MySQL의 max_connections 한계치에 도달하면서 Too Many connections 에러가 발생하고 MySQL로 부터 응답을 받지 못하는 상황이 되었다.

MySQL에서 connection을 끊는 시간이 지나기 전까지는 계속 커넥션이 남아있는데, 한계치 connection에 도달하면 그 이상 커넥션을 주지 못하면서, DB로부터 아무런 응답을 얻지 못했다.

 

DB쪽 코드의 문제라고 생각했었는데, 정작 문제는 async, sync의 혼합된 사용에 있었다.

FastAPI는 비동기처리에 특화된 프레임워크이고, 일반적으로도 router함수에 async를 붙여서 사용하기를 권장한다.

 

그런데 문제는, QueuePool을 사용하면서, sync 함수를 async router 함수에 혼합해서 사용하면서 생겼다.

async router 함수에는 async, await 처리를 한 함수를 사용하여야 하고, 특히 DB 커넥션을 유발하는 함수에 이와같이 async, sync가 섞여 있는 경우, 기존 커넥션이 있음에도 새로운 커넥션을 다시 여는 문제가 발생했다.

 

** async router함수에는 async한 함수만 사용한다. 반드시 (특히 DB 쿼리와 관련된 코드는 유념)

async router 함수에 다른 함수를 사용시 사용되는 함수는 async가 붙어있어야 하며, async가 붙어서 선언된 함수이므로, 사용시에는 await를 붙여서 호출한다.

이는, 의존성함수를 주입할 때도 마찬가지이다. 의존성함수로 주입되는 함수도 async로 선언되어야 하고, 그 안의 로직들도 await 등을 꼼꼼히 챙겨야 한다.

이와 같은 현상이 생기는 이유를 FastAPI의 공식문서의 아랫 부분에서 찾을 수 있지 않을까 싶다.

-> async router 함수에 async만 써야하는 것이 아니라,

   async router 함수에 의존성으로 들어가는 함수는 async이어야 한다.

    -> 반드시라기 보다는 DB connection을 의도치 않게 늘리지 않기 위해서는 async - async 조합이 맞아야 한다.

  (적어도 동기 database 라이브러리에서는 테스트를 통해 확인했음)

  아래에서 클래스를 의존성으로 주입하는 경우도 문제가 있음을 기술할 것인데,

  결국은 async 함수에 sync한 로직이 들어가지 않도록 하는 것이 핵심이다

  -> async 함수만 의존성 가능, 클래스는 인스턴스 선언을 비동기적인 메소드를 통해 할 것

 

Sub-dependencies¶

You can have multiple dependencies and sub-dependencies requiring each other (as parameters of the function definitions), some of them might be created with async def and some with normal def. It would still work, and the ones created with normal def would be called on an external thread (from the threadpool) instead of being "awaited".

(출처: https://fastapi.tiangolo.com/async/)

 

그런데 한가지 더 큰 문제는 의존성으로 class를 주입하는 경우였다.

class를 의존성으로 주입하면, FastAPI에서 __init__에 필요한 인자를 추출해와서 편하고 깔끔하게 인스턴스를 선언할 수 있다. 그런데, 이 __init__은 기본적으로 synchronous이고, async를 붙일 수도 없다.

결국, __init__에는 db와 관련된 코드가 없어도, router 함수등에서 사용되는 의존성 함수/의존성 클래스 생성자에 db 커넥션을 유발하는 코드가 있으면, 기존 커넥션을 무시하고 새로운 커넥션이 생기는 현상이 나타난다.

 

 

이때는 한가지 팁이 있다.

일반 메소드에 async를 붙여서 선언하고, 이 async 메소드를 통해서 인스턴스를 생성하는 것이다.

예제 코드는 다음과 같다.

# classes.py

from typing import Annotated
from fastapi import Path
from database import db_session_sync
from sqlalchemy import select

from models import User


class DBTestSync:

    def __init__(
        self,
        db: db_session_sync,
        path1: str
    ):
        self.db = db
        self.path1 = path1
        
    @classmethod
    async def async_init(
        cls,
        db: db_session_sync,
        path1: Annotated[str, Path(..., title="게시판 테이블명", description="게시판 테이블명")],
    ):
        instance = cls(db, path1)
        return instance

    async def get_users(self):
        result = self.db.execute(select(User))
        users = result.scalars().all()
        return users

 

# main.py

from typing import Annotated
from fastapi import FastAPI, Depends
from sqlalchemy.future import select
from database import db_session_sync
from models import User
from classes import DBTestSync


app = FastAPI()

@app.get("/users_wrapper/{path1}/")
async def read_users(
    db_service: Annotated[DBTestSync, Depends(DBTestSync.async_init)]
):
    users = await db_service.get_users()
    return users

 

 

FastAPI에서 더욱 빠른 속도 향상을 위해서 비동기 database library를 사용할 수 있다.

테스트 과정에서 발견한 것은, 커넥션 풀이 지속적으로 쌓여나간다는 것이다.

여러가지 요청들이 비동기적으로 지속적으로 db에 전해지기에 connection이 계속 쌓이는 것은 당연하다.

그러나 문제는 비동기 요청을 남발하면, 금방 max_connections에 도달하게 되고, 커넥션을 더 얻을 수 없어서 응답을 받지 못하면, create_engine에서 pool_timeout으로 설정한 시간이 지난 후에, 자동으로 쿼리를 중단하고 에러를 반환 받는다.

비동기 db 라이브러리는 항상 커넥션 풀을 초과하지 않도록 잘 관리하면서 써야할 것으로 보인다.

 

요약

1. FastAPI에서 async 함수에는 의존성이든 내부 함수든 async로 철저히 사용하자

  -> 1. FastAPI에서 async router 함수에는 의존성 함수를 async로 철저히 사용하자 (특히 DB 쿼리가 담긴 함수일 경우)

2. 클래스를 의존성으로 주입해서 인스턴스로 얻는 경우는

    sync로 처리가 돼서 async-sync 혼합 문제가 발생하므로, async 메소드를 통해 인스턴스를 별도로 생성하자.

3. 비동기 database 라이브러리는 connection 관리를 철저히 하면서 사용하도록