[파이썬] FastAPI - 비동기 처리 (asynchronous processing)

이전 내용

[FastAPI] 고급 인증: 세션

---

비동기 처리

 

◆ 비동기 처리

하나의 작업이 끝나기를 기다리지 않고 다음 작업을 실행할 수 있는 것.

예. 사용자가 데이터를 요청했을 때 이 데이터를 데이터베이스에서 가져오는 작업을 진행하는 동시에 다른 사용자의 요청을 처리

 

[동기 vs 비동기]

항목	동기(synchronous)	비동기(asynchronous)
태스크 수행 방식	직렬	병렬
처리 흐름	요청을 받은 후 응답을 받아야 다음 동작 수행 ▶ 태스크를 수행할 동안 나머지 태스크는 대기	요청을 보낸 후 응답의 수락 여부와는 무관하게 다음 태스크가 동작
성능	시스템의 전체적인 효율 저하	태스크가 실행되는 동안 다른 태스크를 실행할 수 있으므로 자원을 효율적으로 사용 가능
사용 예시	데이터베이스 쿼리	파일 다운로드, API 호출
복잡성	비교적 단순하여 이해하기 쉬움	이벤트 기반으로 복잡할 수 있음
에러 처리	직관적이며 간단하게 처리	콜백이나 프로미스를 통해 처리

 

 

FastAPI에서는 async와 await 키워드를 사용하여 비동기 처리를 구현한다.

async def 함수는 이전에 작성했던 글을 참고하면 된다.

[파이썬] FastAPI - 응답 모델

 

await 키워드는 비동기 함수 내에서만 사용할 수 있으며, await 다음에 오는 비동기 연산이 완료될 때까지 현재의 비동기 함수의 실행을 일시적으로 중단한다. 중단은 프로그램 전체가 멈추는 것을 의미하는 것이 아니며, 그동안 다른 비동기 함수나 작업이 실행될 수 있다.

▶ async def로 선언된 함수 내에서 await를 사용하면 해당 await 라인에서는 지정된 비동기 작업이 끝날 때까지 기다리고, 그동안 다른 비동기 함수나 코드는 계속 실행될 수 있다.

 

[비동기 처리 간단 예제 코드]

import asyncio
async def func1():
    print("func1: start")
    await asyncio.sleep(2) # 2초 대기
    print("func1: End")

async def func2():
    print("func2: start")
    await asyncio.sleep(1) # 1초 대기
    print("func2: End")

async def main():
    await asyncio.gather(func1(), func2()) # func1과 func2를 동시 실행

# main 함수
if __name__ == "__main__":
    asyncio.run(main())

※ asyncio:
async/await 구문을 사용하여 동시성 코드를 작성하는 라이브러리로, 고성능 네트워크 및 웹 서버 데이터베이스 연결 라이브러리, 분산 작업 큐 등을 제공하는 여러 파이썬 비동기 프레임워크 기반으로 사용된다.

 

이를 터미널에서 실행하면

python main.py

func1()과 func2()는 동시에 시작되나, func2()의 await asyncio.sleep(1)은 1초만 대기하므로 func2()가 먼저 끝난다. 그리고 func1()은 대기 중이지만, func2()의 실행을 방해하지 않는다.

---

FastAPI에서의 비동기 처리

 

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def fetch_data():
    await asyncio.sleep(2) # 2초 대기
    return {"data": "some_data"} # DB에서 데이터 가져오기 시뮬레이션

@app.get("/")
async def read_root():
    data = await fetch_data() # fetch_data 함수가 완료될 때까지 대기
    return {"message": "Hello, FastAPI", "fetch_data": data}

async def(비동기 함수) 내부에서 await 키워드를 사용할 수 있다. await 키워드는 비동기 함수의 실행을 일시적으로 멈추고, 그 함수가 끝날 때까지 대기한다. 여기서는 fetch_data() 함수가 끝날 때까지 기다린다.

 

[테스트]

터미널에서 run.bat 파일을 실행하고

※ run.bat 파일 내에는 uvicorn main:app --reload 로 uvicorn 실행 코드가 있다.

포스트맨에서 HTTP 메서드를 GET으로 두고 localhost:8000/ URL에서 [Send] 버튼을 누르면,

2초(이 테스트에선 2.04s) 후에 JSON 응답을 받게 된다.

---

비동기 처리의 성능 향상

 

서버가 여러 요청을 동시에 처리해야 할 때 각 요청이 I/O 작업(데이터베이스 쿼리, 외부 API 호출 등)을 필요로 하면, 이 작업이 끝날 때까지 다른 요청을 처리할 수 없게 된다. 비동기 처리를 사용하면 이러한 I/O 작업을 기다리는 동안 다른 요청을 처리할 수 있다.

 

[순차적 코드 vs 비동기 코드 성능 비교]

우선, 동기 비동기 코드를 실행할 수 있는 코드를 main.py에 작성한다.

from fastapi import FastAPI
from time import sleep, time
import asyncio

app = FastAPI()

# 동기 처리
@app.get("/sync")
def read_sync_item():
    start_time = time()
    sleep(1)
    elapsed_time = time() - start_time
    return {"type": "sync", "elapsed_time": elapsed_time}

# 비동기 처리
@app.get("/async")
async def read_async_time():
    start_time = time()
    await asyncio.sleep(1)
    elapsed_time = time() - start_time
    return {"type": "async", "elapsed_time": elapsed_time}

 

이 코드를 실행한 뒤, 시간을 측정하기 위한 코드를 test_app.py 파일에 작성한다.

import requests # 동기 HTTP 요청 위함
import aiohttp # 비동기 HTTP 요청 위함
import asyncio
import time

# 동기 API 호출 함수
def sync_call(url, times):
    start_time = time.time() # 시작 시간
    for _ in range(times): # 지정된 횟수만큼 반복
        requests.get(url) # 동기적으로 API 호출
    elapsed_time = time.time() - start_time # 경과 시간 계산
    print(f"Sync elapsed time: {elapsed_time} seconds") # 경과 시간 출력

# 비동기 API 호출 함수
async def async_call(url, times):
    start_time = time.time()
    async with aiohttp.ClientSession() as session: # 비동기 세션 시작
        tasks = [] # task 저장 리스트
        for _ in range(times):
            task = session.get(url) # 비동기적으로 API 호출하고 task 객체 생성
            tasks.append(task) # task 객체를 리스트에 추가
        await asyncio.gather(*tasks) # 모든 비동기 작업이 완료될 때까지 대기
    elapsed_time = time.time() - start_time
    print(f"Async elapsed time: {elapsed_time} seconds")

# 메인 실행 부분
if __name__ == "__main__":
    url_sync = "http://127.0.0.1:8000/sync" # 동기 API URL
    url_async = "http://127.0.0.1:8000/async" # 비동기 API URL
    times = 10 # 호출 횟수

    # 동기 호출 실행
    sync_call(url_sync, times)

    # 비동기 호출 실행
    asyncio.run(async_call(url_async, times)) # 비동기 함수 실행을 위한 메인 이벤트 루프

▶ 이 코드는 동기와 비동기 API를 각각 10번씩 호출하며, 얼마나 시간이 걸리는지 호출한다.

 

★ aiohttp

aiohttp는 비동기 HTTP 클라이언트/서버 프레임워크로, 파이썬의 asyncio 라이브러리를 기반으로 한다. 비동기 프로그래밍 모델을 사용함으로써, aiohttp는 고성능 네트워크 서버 및 클라이언트 개발을 가능하게 한다.

 

- async with aiohttp.ClientSession() as session 구문: 

• async with: 비동기 컨텍스트 매니저를 사용하기 위한 구문. async with를 사용함으로써 리소스를 자동으로 관리하고, 코드가 더 읽기 쉬워진다. 비동기 함수 내에서 사용된다.
• aiohttp.ClientSession(): aiohttp 라이브러리의 ClientSession 클래스는 HTTP 요청을 보내고 응답을 받을 때 사용하는 세션을 나타내며, 이를 통해 여러 요청을 효율적으로 수행할 수 있다.
• as session: session이라는 변수를 통해 만들어진 클라이언트 세션을 참조할 수 있게 된다. 이후 이 변수를 사용하여 HTTP 요청을 보낼 수 있다.

성능 테스트를 위해 main.py는 실행한 상태로 별도의 터미널 창에 python test_app.py 로 실행해 본다.

 

FastAPI 애플리케이션을 통해 /sync 엔드포인트와 /async 엔드포인트에서 각각 10번의 API 호출이 이뤄진다.

동기(sync) API 호출에는 10.1초, 비동기(async) API 호출에는 1.04초가 걸리는 것을 확인할 수 있다. 이를 통해 비동기 처리가 얼마나 효율적인지를 볼 수 있다.

---

[참고]

https://velog.io/@khy226/%EB%8F%99%EA%B8%B0-%EB%B9%84%EB%8F%99%EA%B8%B0%EB%9E%80-Promise-asyncawait-%EA%B0%9C%EB%85%90

가장 빠른 풀스택을 위한 플라스크 & FastAPI

https://docs.python.org/ko/3.13/library/asyncio.html

https://wikidocs.net/229758

---

다음 내용

[FastAPI] 파일 업로드