セラーリー作業員がタスクから結果を返す方法

Question

celery労働者は仕事の結果をフラスコに返すのですか？ 私はタスクを呼び出すフラスコのアプリを持っています。このタスクは、データベースからデータを取り出し、折れ線グラフをプロットし、htmlページにレンダリングされるhtmlコンテンツを返します。 Celeryを使用しない場合、Flaskアプリケーションはうまく動作し、クライアント側で折れ線グラフをレンダリングしますが、セロリをRabbitMQブローカ経由で実行するよう委任したいので、セロリシェルのログ出力を見ることができますが、フラスコサーバーアプリに送り返されます。どうやってするか？

#server-celery.py 

app = Flask(__name__) 

@app.route('/',methods=['GET']) 
def index(): 
    return render_template("index.html") 

@app.route('/', methods=['GET', 'POST']) 
def plotdata(): 
    form = InputForm(request.form) 
    if request.method == 'POST' and form.validate(): 
     lineChart = task.main.delay(form.startdate.data, form.enddate.data) 
     return render_template('view.html', form=form, result= lineChart) 

if __name__ == '__main__': 
    app.run(host='0.0.0.0', port=5000, debug=True) 

そしてCelery労働者として実行中のタスク：

#task.py 
from celery import Celery 

broker = 'amqp://localhost:5672' 
app = Celery(__name__, broker=broker) 

def queryDatabase(param1, param2): 
    #query database, return results 
    return data 

def plotLineChart(data): 
    #plot data as line Chart 
    return LineChartHtmlcontent 

@app.task 
def main(param1,param2): 
    data = queryDatabase(param1,param2) 
    return plotLineChart(data) 

とクライアントのhtmlページ：

<!--view.html--> 

<form method=post action=""> 
<table> 
    {% for field in form %} 
    <tr> 
    <td>{{ field.label }}</td> 
     <td>{{ field }}</td> 

    </tr> 
    {% endfor %} 
</table> 
<p><input type=submit value=Submit></form></p> 

<p> 
{% if result != None %} 
<img src="{{ result }}" width="500"> 
{% endif %} 
</p> 

Answer 1

より良い解決策は、タスクはそれがあったようにセロリを使用して非同期で実行できるようにするだけですページ上でjavascriptを使用してセロリのタスクを定期的にポーリングして状態を確認することを意図しています。

まず、セロリタスクを作成するときにbind = Trueパラメータを使用します。これにより、selfを関数に渡すことができます。

@app.task(bind=True) 
def main(self, param1, param2): 
    self.update_state(state='PENDING') 
    data = queryDatabase(param1,param2) 
    self.update_state(state='COMPLETE') 
    return plotLineChart(data) 

これで、タスクの状態を更新できるようになりました。コードを変更して、update_stateメソッドにメタデータを許可することもできます。たとえば、self.update_state(state='PROGRESS', meta={'current': 0, 'total': 12})

これでどのように処理されているのかを確認する必要があります。

@blueprint.route('/status/<task_id>', methods=['GET']) 
def taskstatus(task_id=None): 
    task = main.AsyncResult(task_id) 
    response = { 
     'state': task.state, 
    } 
    return jsonify(response) 

これは、タスクIDでURLをポーリングすると、jsonオブジェクトを返します。なぜあなたはセロリを使用している、あなたはすぐに結果が必要な場合はhttp://blog.miguelgrinberg.com/post/using-celery-with-flask