U dE@srddlZddlZddlmZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z dd lmZdd lmZdd lmZd d lmZddlmZd dlmZd dlmZddlmZeeZ GdddeZ!Gddde!Z"Gddde!Z#Gddde!Z$Gddde!Z%Gddde!Z&Gdd d e!Z'Gd!d"d"e!Z(Gd#d$d$e!Z)Gd%d&d&e!Z*dS)'N)datetime)web)gen)IOLoop) json_decode) HTTPError)states) AsyncResult)AbortableAsyncResult)DisabledBackend)tasks)BaseApiHandler)Broker)ControlHandler) OrderedDictc@sDeZdZdZddZeddZddZdd Zd d Z d d Z dS)BaseTaskHandlerz%Y-%m-%d %H:%M:%S.%fc Csz|jj}|rt|ni}Wn0tk rL}ztdt|W5d}~XYnXt|tsbtdd|dg}|di}t|t t fstdd|||fS)Nzinvalid optionsargskwargszargs must be an array) requestbodyr ValueErrorrstr isinstancedictpoplisttuple)selfroptionserrr#4/tmp/pip-unpacked-wheel-3pokl8eb/flower/api/tasks.py get_task_argss      zBaseTaskHandler.get_task_argscCst|jt SN)rbackendr )resultr#r#r$backend_configured0sz"BaseTaskHandler.backend_configuredcKs||dSr&)Z set_status)r status_coderr#r#r$ write_error4szBaseTaskHandler.write_errorcCsB|jtjkr(|||j|jdn|d||jidS)N)r( tracebackr()staterFAILUREupdate safe_resultr(r,)r responser(r#r#r$update_response_result7s   z&BaseTaskHandler.update_response_resultcCsd|krt|d|j|d<d|kr6t|d|d<d|kr~|d}z t|}Wn"tk rtt||j}YnX||d<dS)NetaZ countdownexpires)rstrptime DATE_FORMATfloatr)r r!r4r#r#r$normalize_options>s  z!BaseTaskHandler.normalize_optionscCs4zt|Wntk r*t|YSX|SdS)zreturns json encodable resultN)jsondumps TypeErrorrepr)r r(r#r#r$r0Ls zBaseTaskHandler.safe_resultN) __name__ __module__ __qualname__r6r% staticmethodr)r+r2r8r0r#r#r#r$rs rc@s(eZdZejejddZddZdS) TaskApplyccs|\}}}td|||z|jj|}Wn"tk rPtdd|YnXz||Wntk r~tddYnX|j f||d|}d|j i}t d|j||V}||dS) a# Execute a task by name and wait results **Example request**: .. sourcecode:: http POST /api/task/apply/tasks.add HTTP/1.1 Accept: application/json Accept-Encoding: gzip, deflate, compress Content-Length: 16 Content-Type: application/json; charset=utf-8 Host: localhost:5555 { "args": [1, 2] } **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 71 Content-Type: application/json; charset=UTF-8 { "state": "SUCCESS", "task-id": "c60be250-fe52-48df-befb-ac66174076e6", "result": 3 } :query args: a list of arguments :query kwargs: a dictionary of arguments :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task 'Invoking a task '%s' with '%s' and '%s'Unknown task '%s'rInvalid optionrrtask-idN)r%loggerdebugcappr KeyErrorrr8r apply_asynctask_idrcurrentZrun_in_executor wait_resultswriter tasknamerrr!taskr(r1r#r#r$postWs.* zTaskApply.postcCs4|jdd|||||r0|j|jd|S)NF) propagater-)getr2r)r/r-)r r(r1r#r#r$rOs    zTaskApply.wait_resultsN) r=r>r?r authenticatedr coroutinerTrOr#r#r#r$rAVs =rAc@seZdZejddZdS)TaskAsyncApplycCs|\}}}td|||z|jj|}Wn"tk rPtdd|YnXz||Wntk r~tddYnX|j f||d|}d|j i}| |r|j |j d||d S) ac Execute a task **Example request**: .. sourcecode:: http POST /api/task/async-apply/tasks.add HTTP/1.1 Accept: application/json Accept-Encoding: gzip, deflate, compress Content-Length: 16 Content-Type: application/json; charset=utf-8 Host: localhost:5555 { "args": [1, 2] } **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 71 Content-Type: application/json; charset=UTF-8 Date: Sun, 13 Apr 2014 15:55:00 GMT { "state": "PENDING", "task-id": "abc300c7-2922-4069-97b6-a635cc2ac47c" } :query args: a list of arguments :query kwargs: a dictionary of arguments :query options: a dictionary of `apply_async` keyword arguments :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task rBrCrDrrErFrGrVN)r%rHrIrJr rKrr8rrLrMr)r/r-rPrQr#r#r$rTs&*  zTaskAsyncApply.postNr=r>r?rrXrTr#r#r#r$rZsrZc@seZdZejddZdS)TaskSendcCsh|\}}}td||||jj|f||d|}d|ji}||rZ|j|jd| |dS)a" Execute a task by name (doesn't require task sources) **Example request**: .. sourcecode:: http POST /api/task/send-task/tasks.add HTTP/1.1 Accept: application/json Accept-Encoding: gzip, deflate, compress Content-Length: 16 Content-Type: application/json; charset=utf-8 Host: localhost:5555 { "args": [1, 2] } **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 71 Content-Type: application/json; charset=UTF-8 { "state": "SUCCESS", "task-id": "c60be250-fe52-48df-befb-ac66174076e6" } :query args: a list of arguments :query kwargs: a dictionary of arguments :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task z%Invoking task '%s' with '%s' and '%s'rFrGrVN) r%rHrIrJZ send_taskrMr)r/r-rP)r rRrrr!r(r1r#r#r$rTs$(  z TaskSend.postNr[r#r#r#r$r\sr\c@seZdZejddZdS) TaskResultcCs|dd}|dk rt|nd}t|}||s:td||jd}|rf|j|dd|||n|rz|||| |dS)a Get a task result **Example request**: .. sourcecode:: http GET /api/task/result/c60be250-fe52-48df-befb-ac66174076e6 HTTP/1.1 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 84 Content-Type: application/json; charset=UTF-8 { "result": 3, "state": "SUCCESS", "task-id": "c60be250-fe52-48df-befb-ac66174076e6" } :query timeout: how long to wait, in seconds, before the operation times out :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 503: result backend is not configured timeoutN)rGr-F)r^rU) get_argumentr7r r)rr-rWr2readyrP)r taskidr^r(r1r#r#r$rWs    zTaskResult.getNr=r>r?rrXrWr#r#r#r$r]sr]c@seZdZejddZdS) TaskAbortcCsFtd|t|}||s&td||td|ddS)a) Abort a running task **Example request**: .. sourcecode:: http POST /api/task/abort/c60be250-fe52-48df-befb-ac66174076e6 HTTP/1.1 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 61 Content-Type: application/json; charset=UTF-8 { "message": "Aborted '1480b55c-b8b2-462c-985e-24af3e9158f9'" } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 503: result backend is not configured zAborting task '%s'r_z Aborted '%s')messageN)rHinfor r)rabortrPr)r rbr(r#r#r$rTGs   zTaskAbort.postNr[r#r#r#r$rdFsrdc@s eZdZejejddZdS)GetQueueLengthsccs|j}|jjj}d}|jdkr.|jjr.|jj}d}|jjjrF|jjj}t|j j dd|||d}| }|st |jjj gt dd|jjjpgDB}|t|V}|d|idS) aH Return length of all active queues **Example request**: .. sourcecode:: http GET /api/queues/length Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 94 Content-Type: application/json; charset=UTF-8 { "active_queues": [ {"name": "celery", "messages": 0}, {"name": "video-queue", "messages": 5} ] } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 503: result backend is not configured NZamqpT)Zinclude_password)http_apibroker_optionsbroker_use_sslcSsg|]}|jr|jqSr#)name).0qr#r#r$ sz'GetQueueLengths.get..Z active_queues) applicationrJconfZBROKER_TRANSPORT_OPTIONS transportr!Z broker_apiZBROKER_USE_SSLr connectionas_uriZget_active_queue_namessetZCELERY_DEFAULT_QUEUEZ CELERY_QUEUESqueuessortedrP)r apprjrirkZbrokerZ queue_namesrvr#r#r$rWps(!   zGetQueueLengths.getN)r=r>r?rrXrrYrWr#r#r#r$rhosrhc@seZdZejddZdS) ListTasksc Cs&|j}|dd}|jddtd}|dd}|dd}|dd}|d d}|d d}|d d} |ott|}t|d}|d kr|nd}|d kr|nd}|d kr|nd}g} tj|j||| |||||d D]@\} } t| } | dd}|dk r|j | d<| | | fq| t | dS)aU List tasks **Example request**: .. sourcecode:: http GET /api/tasks HTTP/1.1 Host: localhost:5555 User-Agent: HTTPie/0.8.0 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 1109 Content-Type: application/json; charset=UTF-8 Etag: "b2478118015c8b825f7b88ce6b660e5449746c37" Server: TornadoServer/3.1.1 { "e42ceb2d-8730-47b5-8b4d-8e0d2a1ef7c9": { "args": "[3, 4]", "client": null, "clock": 1079, "eta": null, "exception": null, "exchange": null, "expires": null, "failed": null, "kwargs": "{}", "name": "tasks.add", "received": 1398505411.107885, "result": "'7'", "retried": null, "retries": 0, "revoked": null, "routing_key": null, "runtime": 0.01610181899741292, "sent": null, "started": 1398505411.108985, "state": "SUCCESS", "succeeded": 1398505411.124802, "timestamp": 1398505411.124802, "traceback": null, "uuid": "e42ceb2d-8730-47b5-8b4d-8e0d2a1ef7c9", "worker": "celery@worker1" }, "f67ea225-ae9e-42a8-90b0-5de0b24507e0": { "args": "[1, 2]", "client": null, "clock": 1042, "eta": null, "exception": null, "exchange": null, "expires": null, "failed": null, "kwargs": "{}", "name": "tasks.add", "received": 1398505395.327208, "result": "'3'", "retried": null, "retries": 0, "revoked": null, "routing_key": null, "runtime": 0.012884548006695695, "sent": null, "started": 1398505395.3289, "state": "SUCCESS", "succeeded": 1398505395.341089, "timestamp": 1398505395.341089, "traceback": null, "uuid": "f67ea225-ae9e-42a8-90b0-5de0b24507e0", "worker": "celery@worker1" } } :query limit: maximum number of tasks :query offset: skip first n tasks :query sort_by: sort tasks by attribute (name, state, received, started) :query workername: filter task by workername :query taskname: filter tasks by taskname :query state: filter tasks by state :query received_start: filter tasks by received date (must be greater than) format %Y-%m-%d %H:%M :query received_end: filter tasks by received date (must be less than) format %Y-%m-%d %H:%M :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request limitNoffsetr)defaulttypeZ workernamerRr-received_start received_endsort_byZAll)rzr{rr}workerr-r~rr) rpr`intmaxr Z iter_taskseventsas_dictrhostnameappendrPr) r rxrzr{rr}r-r~rrr(rMrSr#r#r$rWs@\             z ListTasks.getNrcr#r#r#r$rysryc@seZdZejddZdS) ListTaskTypescCs(|jjj}i}||d<||dS)a List (seen) task types **Example request**: .. sourcecode:: http GET /api/task/types HTTP/1.1 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 44 Content-Type: application/json; charset=UTF-8 { "task-types": [ "tasks.add", "tasks.sleep" ] } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request z task-typesN)rprr-Z task_typesrP)r Zseen_task_typesr1r#r#r$rW%szListTaskTypes.getNrcr#r#r#r$r$src@seZdZejddZdS)TaskInfocCsNt|jj|}|s"tdd||}|jdk r@|jj|d<||dS)a Get a task info **Example request**: .. sourcecode:: http GET /api/task/info/91396550-c228-4111-9da4-9d88cfd5ddc6 HTTP/1.1 Accept: */* Accept-Encoding: gzip, deflate, compress Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 575 Content-Type: application/json; charset=UTF-8 { "args": "[2, 2]", "client": null, "clock": 25, "eta": null, "exception": null, "exchange": null, "expires": null, "failed": null, "kwargs": "{}", "name": "tasks.add", "received": 1400806241.970742, "result": "'4'", "retried": null, "retries": null, "revoked": null, "routing_key": null, "runtime": 2.0037889280356467, "sent": null, "started": 1400806241.972624, "state": "SUCCESS", "succeeded": 1400806243.975336, "task-id": "91396550-c228-4111-9da4-9d88cfd5ddc6", "timestamp": 1400806243.975336, "traceback": null, "worker": "celery@worker1" } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task rCrDNr) r Zget_task_by_idrprrrrrrP)r rbrSr1r#r#r$rWLs9  z TaskInfo.getNrcr#r#r#r$rKsr)+r9loggingrZtornadorrZtornado.iolooprZtornado.escaperZ tornado.webrZceleryrZ celery.resultr Zcelery.contrib.abortabler Zcelery.backends.baser utilsr rZ utils.brokerrZ api.controlr collectionsr getLoggerr=rHrrArZr\r]rdrhryrrr#r#r#r$s6                ;JA41):{'