FastAPI 비동기 처리(Async) 완벽 가이드: 고성능 웹 개발의 핵심

FastAPI 다이어그램

1. 비동기 처리의 핵심 개념

비동기 프로그래밍은 I/O 작업(데이터베이스 쿼리, 외부 API 호출, 파일 입출력)이 완료될 때까지 기다리지 않고 다른 작업을 처리하는 방식입니다. FastAPI는 async/await 키워드를 통해 비동기 코드를 간결하게 작성할 수 있습니다.

• 동기(Synchronous) vs 비동기(Asynchronous):
  • 동기: 작업 순차 처리 → I/O 대기 시간이 성능 저하 유발
  • 비동기: I/O 대기 시간에 다른 작업 처리 → 높은 자원 활용률
• Event Loop:
  단일 스레드에서 여러 작업을 전환하며 처리하는 메커니즘 (Python의 asyncio 기반).

 

2. FastAPI에서 비동기 사용 시나리오

▷ 비동기가 필요한 상황

• 외부 API 호출
  • 장점: 요청 대기 시간 동안 다른 작업 처리 가능

 

import httpx  
@app.get("/posts")  
async def fetch_posts():  
    async with httpx.AsyncClient() as client:  
        response = await client.get("https://api.example.com/posts")  
        return response.json()

• 데이터베이스 연동
  • 주의: 동기 라이브러리(SQLAlchemy 동기 버전) 사용 시 오히려 성능 저하

 

from databases import Database  
database = Database("sqlite:///db.sqlite")  
@app.get("/users/{id}")  
async def get_user(id: int):  
    query = "SELECT * FROM users WHERE id = :id"  
    return await database.fetch_one(query, {"id": id})

• 파일 I/O

 

@app.post("/upload")  
async def upload_file(file: UploadFile):  
    contents = await file.read()  # 비동기 파일 읽기  
    return {"filename": file.filename}

 

 

3. async/await 작동 원리

• async: 함수를 비동기로 선언 → 해당 함수가 코루틴(Coroutine)으로 변환
• await: 비동기 작업 완료까지 대기 → 이벤트 루프가 다른 작업 전환

async def example():  
    result = await some_async_task()  # 다른 작업으로 전환 가능  
    return result

 

 

4. 성능 최적화 팁

• CPU-bound 작업 주의:
  • 이미지 처리, 머신러닝 연산 등 CPU 집약적 작업은 별도 스레드/프로세스에서 처리.

from concurrent.futures import ProcessPoolExecutor  
@app.get("/process")  
async def heavy_task():  
    with ProcessPoolExecutor() as pool:  
        result = await loop.run_in_executor(pool, cpu_intensive_func)  
    return result

• 동기 코드와의 혼용 방지:
  • def(동기)와 async def(비동기)를 섞어 사용하면 이벤트 루프 차단 가능성 ↑
• 적절한 라이브러리 선택:
  • 데이터베이스: databases, asyncpg
  • HTTP 클라이언트: httpx, aiohttp

 

5. 실전 예제: 비동기 미들웨어

@app.middleware("http")  
async def log_requests(request: Request, call_next):  
    start_time = time.time()  
    response = await call_next(request)  
    process_time = time.time() - start_time  
    logger.info(f"Request: {request.url} | Time: {process_time:.2f}s")  
    return response

 

 

• 모든 요청의 처리 시간을 비동기로 로깅합니다.

 

6. 주의사항

• Deadlock 방지: 중첩된 await 호출 시 실행 순서 주의
• 오류 처리: try-except로 비동기 함수 내 예외 처리 필수
• 테스트: pytest-asyncio로 비동기 코드 테스트

 

마무리

FastAPI의 비동기 처리는 I/O 병목 현상을 해결하고 초당 처리 가능한 요청 수를 극대화합니다. async/await를 활용해 외부 API, 데이터베이스, 파일 시스템과의 효율적인 상호작용을 구현하면, 기존 동기 방식 대비 3~5배 높은 성능을 달성할 수 있습니다. 단, CPU-bound 작업에는 적합하지 않으므로 동기-비동기 혼용 전략이 필요합니다.

 

[FastAPI] 비동기 처리(Async) 이해하기

 

FastAPI로 RESTful API 구축하기: 초보자를 위한 완벽 가이드