celery + flask 를 이용한 파이썬 비동기 처리 API 구현

flask는 유저의 요청을 동기적으로 처리한다. 

 

실행 시간이 오래 걸릴 것으로 기대되는 엔드포인트의 경우, 비동기 처리 방식으로 구현하고 싶을 수 있는데

 

celery 를 이용해서 아래와 같이 구현하면 된다.

 

celery의 백엔드로 레디스를 사용할 것이므로 먼저 설치&실행 해준다.

 

아래 코드에선

- 비동기 작업 (adder) 시작

- 특정 태스크(id)의 상태 체크

- 결과 받아보기 endpoint를 구현하였다.

import os
import sys
import time

from flask import Flask
from flask import request
from flask import jsonify


from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL'],
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery


app = Flask(__name__)

app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)

celery = make_celery(app)

task_cache = dict()

@celery.task()
def add_together(a, b):
    time.sleep(5)
    return a+b

@app.route('/adder', methods=['GET'])
def adder():
    global task_cache
    a = int(request.args.get('a'))
    b = int(request.args.get('b'))
    task = add_together.delay(a, b)
    task_cache[task.id] = task
    return task.id

@app.route('/progress', methods=['GET'])
def progress():
    global task_cache
    task_id = request.args.get('task_id')
    task = task_cache[task_id]
    return jsonify({
        'status': task.ready()
    })

@app.route('/result', methods=['GET'])
def result():
    global task_cache
    task_id = request.args.get('task_id')
    task = task_cache[task_id]
    return jsonify({
        'result': task.get()
    })

 

위와 같이 app.py 를 작성 했을 때, 

 

먼저 celery를 실행 시켜주어야 한다. app은 app.py를 의미

 celery -A app.celery worker --loglevel=info

 

이 후에 flask app을 별도로 실행시킨다.

export FLASK_RUN_HOST=0.0.0.0
export FLASK_RUN_PORT=3000
export FLASK_APP=app.py
flask run