[Python] FastAPI 성능 최적화

1. FastAPI 란?

FastAPI는 Python을 기반으로 한 웹 프레임워크로, 성능, 사용성 그리고 Python 기능을 활용할 수 있다는 측면에서 장점이 있다. 특히, Python 기반의 웹 프레임워크라는 것은 Python 이 친숙한 AI 개발자가 모델 서빙 시 사용할 백엔드 프레임워크로 사용하기 매우 적합하다. 실제로 모델 개발 후 마이크로 서비스 형태로 서빙 시 FastAPI 를 매우 자주 사용하고 있으며, 아래와 같은 여러가지 장점 때문에 pydantic 과 함께 점점 더 사용 빈도가 증가하고 있다.

1.1. 장점은?

• 성능: FastAPI 는 starlette 과 pydantic 을 기반으로 구축되어 좋은 성능을 낼 수 있는데, 기본적으로 async 를 지원한다. 동시 요청을 효율적으로 처리할 수 있기 때문에 고성능 애플리케이션 구축에 유용하다. 또한 async 를 지원하기 때문에 ASGI(Asynchronous Server Gateway Interface) 서버인 uvicorn 과 결합이 매우 좋다.
• 데이터 validation 및 serialization: FastAPI 는 데이터 검증에 매우 강력한데, 이는 pydantic 을 기반으로 하고 있기 때문이다. 요청 시 자동으로 입력된 데이터를 검증할 수 있기 때문에 별도 검증 로직을 필요로 하지 않으며, API 요청 데이터의 유효성을 보장한다. 또한, pydantic 을 사용하여 데이터 스키마를 정의하며, 이를 통해 별도 처리 없이 데이터 serialization/deserialization 을 지원한다. 즉, POST 요청에 대해서 json 형태로 주고 받으면서 endpoint 내부적으로 pydantic 스키마 형태로 자동 변환이 가능하다.
• Swagger 문서: 개인적으로 FastAPI 를 사용하는 가장 큰 이유 중 하나인데, FastAPI 는 OpenAPI 표준을 기반으로 한 Swagger UI 를 자동으로 생성해준다. 즉, 내가 정의한 API 의 endpoints 에 대하여 실시간으로 테스트 할 수 있는 인터페이스를 제공하고, Swagger 문서도 자동으로 생성하여 제공한다. 생성된 Swagger 문서는 협업 시 API 스펙 및 스키마를 전달할 때 매우 유용하다.

2. FastAPI 성능 최적화

FastAPI 사용 시 몇 가지 성능을 더 끌어올릴 수 있는 방법에 대한 포스팅이 있어서 정리해보았다. 아래 문서를 참조하였으며, 중간 중간에 내가 이해한 형태로 수정 및 보완을 진행하였다.

Optimizing Performance with FastAPI

2.1. Utilize Asynchronous Endpoints

FastAPI 는 기본적으로 async 프로그래밍을 지원한다. (async 방식으로 동작하는 것을 기본으로 설계가 되었다.) async def 를 사용하면 I/O 바운드 작업에 특히 유용한데, 작업을 async 로 처리할 수 있기 때문에 CPU 가 블로킹되지 않고 다른 작업을 처리할 수 있어 동시 요청을 효율적으로 처리할 수 있다. 코드 구현 시 sync def 보다 고려해야 할 요소가 많은데, async/await 구문 및 coroutine 을 이해하고 적절하게 사용한다면 sync def 보다 성능 좋은 async def 를 구현할 수 있다.

2.2. Use Connection Pooling with Databases

데이터베이스 연결 풀링은 데이터베이스 서버와의 연결을 효율적으로 관리하는 기술로, 데이터베이스와의 연결을 미리 만들어 두고 그 연결을 필요할 때마다 재사용하는 방식이다. 기본적으로 데이터베이스의 연결을 여는 것은 비용이 큰데, 풀링을 통해 데이터베이스와의 연결을 미리 준비하고 필요할 때 빠르게 할당하며 작업이 끝나면 다시 반환하는 구조를 갖는다. 특히, 여러 개의 데이터베이스를 연결할 때 효율이 높고, SQLAlchemy 와 함께 사용하면 FastAPI 의 비동기적 처리 능력과 맞물려 성능이 높아진다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(
    DATABASE_URL, 
    pool_size=20, 
    max_overflow=0
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

2.3. Implement Caching

자주 엑세스하는 데이터를 캐싱하면 데이터베이스의 부하를 크게 줄이고 속도도 높일 수 있다. 캐싱에는 Redis 를 사용할 수 있다고 하는데, 실제 적용해 본 적은 없어서 예제 코드만 작성하였다.

예시: Redis 를 사용한 캐싱 (aioredis 사용)

pip install aioredis

import aioredis
from fastapi import FastAPI, Depends

app = FastAPI()
redis = aioredis.from_url("redis://localhost")

async def get_redis():
    return redis

@app.get("/cached")
async def get_cached_data(redis=Depends(get_redis)):
    cached_value = await redis.get("my_key")
    if cached_value:
        return {"value": cached_value}
    # Simulate data fetching
    data = "some expensive operation result"
    await redis.set("my_key", data)
    return {"value": data}

2.4. Optimize Query Performance

SQLAlchemy 와 같은 ORM(Object Relational Mapper) 을 사용하다보면 쿼리 성능 저하를 일으키는 N+1 문제가 발생할 수 있는데, 이를 eager loading 으로 해결하는 것이 좋다. N+1 문제란 데이터베이스에서 다대일 관계를 조회할 때 발생할 수 있는데, 1개의 쿼리로 인해 N개의 결과를 가져오고, 두 번째 쿼리로 N개의 쿼리가 발생하는 상황이다. (총 N+1 쿼리) 문제는 N+1 쿼리에 상당수의 중복 데이터가 발생한다는 것이다.

• eager loading: 미리 모든 관련 데이터를 가져오는 방식으로, 쿼리 수를 줄이고 N+1 문제를 방지할 수 있지만, 메모리 사용량이 증가할 수 있다.
• lazy loading: 실제로 필요할 때 쿼리를 실행하여 데이터를 가져오는 방식으로, 메모리 사용량을 줄일 수 있지만 많은 데이터를 다룰 때에는 N+1 문제가 발생할 수 있다.

from sqlalchemy.orm import joinedload

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).options(joinedload(User.items)).filter(User.id == user_id).first()
    return user

2.5. Leverage Gzip Middleware

Gzip 압축은 resopnse body 의 사이즈를 줄일 수 있고 네트워크로 전송되는 데이터의 양을 감소시킬 수 있어서 성능을 향상시킬 수 있다. 아래와 같은 형태로 사용하여 최적화가 가능하다.

from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)

2.6. Use FastAPI’s Background Tasks

오래 걸리는 tasks 의 경우, 메인 스레드가 블로킹 되는 것을 방지하기 위해 background tasks 를 고려해볼 수 있다. 아래 예시에서 log 를 write 하는 것은 background tasks 로 처리하였는데, add_task 를 사용하여 tasks 를 추가하는 형태로 구현할 수 있다.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as log_file:
        log_file.write(message + "\n")

@app.post("/log")
async def log_message(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, message)
    return {"message": "Message will be logged in the background"}

2.7. Profile and Monitor Your Application

2.8. Use a Content Delivery Network (CDN)

CDN 을 통해 static files 을 제공하면 대기 시간이 줄어들고 로드 시간이 향상될 수 있다. 즉, FastAPI 에서 자체적으로 static file 를 서빙할 수도 있지만, 그것보다 CDN 을 통해서 더 빠르게 제공하는 것이 일반적이다. static files 은 CDN 에 배포하고, FastAPI 에서는 static files 관련 URL 을 제공하는 형태로 구현한다. FastAPI 서버는 API 요청 처리에만 집중할 수 있고, static files 에 대한 요청은 CDN 이 처리하게 된다.

from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

# Configure your CDN to point to /static endpoint of your FastAPI app

2.9. Optimize Data Serialization

FastAPI 는 데이터 검증 및 직렬화에 pydantic 을 사용하는데, 이것은 이미 빠르지만 추가적으로 ujson 라이브러리를 사용하여 최적화할 수 있다.

UltraJSON is an ultra fast JSON encoder and decoder written in pure C with bindings for Python 3.8+.

from fastapi import FastAPI
import ujson

app = FastAPI(default_response_class=UJSONResponse)

@app.get("/")
async def read_root():
    return {"message": "Hello, World"}

3. 마치며

FastAPI는 웹 개발에서 Python 기반 빠르고 효율적인 API 구축을 원하는 사람들에게 매우 유용하다. 고성능 비동기 처리를 기본으로 하고 있으며, 최근에는 Python 에서도 필수가 되어버린 타입 힌팅과 Pydantic 을 활용한 데이터 검증을 통해 개발의 생산성을 높일 수 있다. 또한, 자동화된 API 문서화 기능은 개발 과정에서 반복적인 작업을 줄여주고, 협업 시 커뮤니케이션을 원활하게 하는 데 도움을 준다. 또한, SQLAlchemy 와 같은 ORM 과의 자연스러운 통합을 통해 복잡한 데이터베이스 작업에도 유용하다. 상황에 맞게 위에 작성한 추가적인 최적화 방법을 적용한다면 더 좋은 시스템을 구축할 수 있을 것으로 생각한다.