오늘은 Celery와 Flask를 사용하여 비동기 API 처리를 하는 방법에 대해 알아보겠습니다. 비동기 처리는 대용량 요청을 처리하고 응답 시간을 최적화하는 데에 매우 유용합니다. Celery는 Python 기반의 분산 작업 큐로, 비동기 작업을 쉽게 구현할 수 있도록 도와줍니다. Flask는 경량화된 웹 애플리케이션 프레임워크로, API를 구현하기에 적합합니다.
Celery 설정
먼저, Celery를 설정해야 합니다. celery 명령을 사용하여 Celery 애플리케이션을 생성하고, 사용할 broker 및 backend를 설정해야 합니다. 이 예제에서는 Redis를 사용하여 메시지 브로커와 결과 백엔드를 설정하겠습니다. 다음과 같이 celery.py 파일을 생성합니다.
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
@app.task
def process_request(request):
# 요청 처리 로직을 작성합니다.
return response
Flask 앱과의 통합
Celery 앱을 Flask 앱과 통합시켜 비동기 작업을 수행하려면 다음 단계를 따릅니다.
- Flask 앱과 Celery 앱을 동일한 프로젝트 내에서 실행할 수 있도록 합니다.
- Flask 앱에서 Celery 앱을 import하여 사용할 수 있도록 합니다.
- 비동기 작업을 호출하는 API 엔드포인트를 구현합니다.
from flask import Flask, request
from celery import group
app = Flask(__name__)
app.config.update(
CELERY_BROKER_URL='redis://localhost:6379/0',
CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)
celery_app = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
@celery_app.task
def process_request(request):
# 요청 처리 로직을 작성합니다.
return response
@app.route('/api/process', methods=['POST'])
def process_api_request():
request_data = request.json
tasks = []
for data in request_data:
tasks.append(process_request.delay(data))
result = group(tasks)().get()
return result
if __name__ == '__main__':
app.run()
위 예제에서는 Flask 앱과 Celery 앱을 동일한 파일 내에서 실행하기 위해 celery_app을 Flask 앱과 동일한 컨텍스트에서 선언합니다. process_api_request() 함수는 비동기 작업을 실행하고 결과를 반환하는 API 엔드포인트를 구현한 예시입니다.
실행 및 테스트
위의 설정과 코드를 모두 준비한 후, 다음 단계를 따라서 실행하고 테스트할 수 있습니다.
- Redis 서버를 실행합니다.
- Celery worker를 실행합니다.
celery -A celery_app worker --loglevel=info명령을 사용합니다. - Flask 앱을 실행합니다.
python app.py명령을 사용합니다. - API 클라이언트를 사용하여
/api/process엔드포인트에 POST 요청을 보내고 결과를 확인합니다.
비동기 API 처리를 구현하는데 있어서 Celery와 Flask는 강력한 조합입니다. Celery를 이용하면 비동기 작업을 쉽게 구현할 수 있으며, Flask를 이용하여 API를 구현하면 더욱 편리하게 사용할 수 있습니다. 이 두 가지 기술을 조합하여 대용량 요청을 효율적으로 처리하는 서비스를 만들어보세요.