сокол питон пример с сельдереем

Question

import falcon 
import json 
from tasks import add 
from waitress import serve 


class tasksresource: 
    def on_get(self, req, resp): 
     """Handles GET requests""" 
     self.result = add.delay(1, 2) 
     self.context = {'ID': self.result.id, 'final result': self.result.ready()} 
     resp.body = json.dumps(self.context) 



api = falcon.API() 
api.add_route('/result', tasksresource()) 
# api.add_route('/result/task', taskresult()) 
if __name__ == '__main__': 
    serve(api, host='127.1.0.1', port=5555) 

как я получить Получить идентификатор задачи от JSon полезной нагрузки (почтовые данные) и добавить маршрут к нему

Answer 1

Вот небольшой пример. Структура файлов:

/project 
     __init__.py 
     app.py # routes, falcon etc. 
     tasks.py # celery 
example.py # script for demonstration how it works 

app.py:

import json 

import falcon 
from tasks import add 
from celery.result import AsyncResult 


class StartTask(object): 

    def on_get(self, req, resp): 
     # start task 
     task = add.delay(4, 4) 
     resp.status = falcon.HTTP_200 
     # return task_id to client 
     result = {'task_id': task.id} 
     resp.body = json.dumps(result) 


class TaskStatus(object): 

    def on_get(self, req, resp, task_id): 
     # get result of task by task_id and generate content to client 
     task_result = AsyncResult(task_id) 
     result = {'status': task_result.status, 'result': task_result.result} 
     resp.status = falcon.HTTP_200 
     resp.body = json.dumps(result) 


app = falcon.API() 

# registration of routes 
app.add_route('/start_task', StartTask()) 
app.add_route('/task_status/{task_id}', TaskStatus()) 

tasks.py:

from time import sleep 

import celery 


app = celery.Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') 


@app.task 
def add(x, y): 
    """ 
    :param int x: 
    :param int y: 
    :return: int 
    """ 
    # sleep just for demonstration 
    sleep(5) 

    return x + y 

Теперь мы должны начать celery приложения. Перейти к project папку и запустите:

celery -A tasks worker --loglevel=info 

После этого мы должны начать Falcon приложения. Перейдите на страницу project папку и запустить:

gunicorn app:app 

Ok. Все готово.

example.py небольшая сторона клиента, который может помочь понять:

from time import sleep 

import requests 
# start new task 
task_info = requests.get('http://127.0.0.1:8000/start_task') 
task_info = task_info.json() 

while True: 
    # check status of task by task_id while task is working 
    result = requests.get('http://127.0.0.1:8000/task_status/' + task_info['task_id']) 
    task_status = result.json() 

    print task_status 

    if task_status['status'] == 'SUCCESS' and task_status['result']: 
     print 'Task with id = %s is finished' % task_info['task_id'] 
     print 'Result: %s' % task_status['result'] 
     break 
    # sleep and check status one more time 
    sleep(1) 

python ./example.py Просто позвоните, и вы должны увидеть что-то вроде этого:

{u'status': u'PENDING', u'result': None} 
{u'status': u'PENDING', u'result': None} 
{u'status': u'PENDING', u'result': None} 
{u'status': u'PENDING', u'result': None} 
{u'status': u'PENDING', u'result': None} 
{u'status': u'SUCCESS', u'result': 8} 
Task with id = 76542904-6c22-4536-99d9-87efd66d9fe7 is finished 
Result: 8 

Надеется, что это помогает.