o RDiC@sZddlZddlZddlmZddlmZddlmZddlmZddl m Z ddl m Z ddl mZdd lmZdd lmZdd lmZd d lmZd dlmZddlmZeeZGdddeZGdddeZGdddeZ GdddeZ!GdddeZ"GdddeZ#GdddeZ$Gdd d eZ%Gd!d"d"eZ&Gd#d$d$eZ'dS)%N) OrderedDict)datetime)states)DisabledBackend)AbortableAsyncResult) AsyncResult)web) json_decode)IOLoop) HTTPError)tasks)Broker)BaseApiHandlerc@sDeZdZdZddZeddZddZdd Zd d Z d d Z dS)BaseTaskHandlerz%Y-%m-%d %H:%M:%S.%fc Csz|jj}|r t|ni}Wnty"}ztdt||d}~wwt|ts-tdd|dg}|di}t|t t fsEtdd|||fS)Nzinvalid optionsargskwargszargs must be an array) requestbodyr ValueErrorr str isinstancedictpoplisttuple)selfroptionserrr!I/var/www/Datamplify/venv/lib/python3.10/site-packages/flower/api/tasks.py get_task_argss      zBaseTaskHandler.get_task_argscCst|jt SN)rbackendr)resultr!r!r"backend_configured+sz"BaseTaskHandler.backend_configuredcKs||dSr$) set_status)r status_coderr!r!r" write_error/szBaseTaskHandler.write_errorcCsD|jtjkr|||j|jddS|d||jidS)N)r& tracebackr&)staterFAILUREupdate safe_resultr&r+)rresponser&r!r!r"update_response_result2s  z&BaseTaskHandler.update_response_resultcCsd|vrt|d|j|d<d|vrt|d|d<d|vr@|d}zt|}Wnty9t||j}Ynw||d<dSdS)Neta countdownexpires)rstrptime DATE_FORMATfloatr)rrr4r!r!r"normalize_options9s    z!BaseTaskHandler.normalize_optionscCs,zt|W|Styt|YSw)zreturns json encodable result)jsondumps TypeErrorrepr)rr&r!r!r"r/Gs   zBaseTaskHandler.safe_resultN) __name__ __module__ __qualname__r6r# staticmethodr'r*r1r8r/r!r!r!r"rs  rc@s"eZdZejddZddZdS) TaskApplyc s|\}}}td|||z|jj|}Wnty.}z tdd|d|d}~wwz||WntyH}ztdd|d}~ww|j d ||d|}d |j i}t d|j||IdH}||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 ''NrInvalid optionrrtask-idr!)r#loggerdebugcappr KeyErrorr r8r apply_asynctask_idr currentrun_in_executor wait_resultswrite rtasknamerrrtaskexcr&r0r!r!r"postQs.)    zTaskApply.postcCs4|jdd|||||r|j|jd|S)NF) propagater,)getr1r'r.r,)rr&r0r!r!r"rQs   zTaskApply.wait_resultsN)r=r>r?r authenticatedrWrQr!r!r!r"rAPs  =rAc@eZdZejddZdS)TaskAsyncApplyc Cs|\}}}td|||z|jj|}Wnty-}z tdd|d|d}~wwz||WntyG}ztdd|d}~ww|j d ||d|}d |j i}| |rd|j |j d ||dS) 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 rBrCrDrENrrFrGrHrYr!)r#rIrJrKr rLr r8rrMrNr'r.r,rRrSr!r!r"rWs**   zTaskAsyncApply.postNr=r>r?rr[rWr!r!r!r"r]sr]c@r\)TaskSendcCsh|\}}}td||||jj|f||d|}d|ji}||r-|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'rGrHrYN) r#rIrJrK send_taskrNr'r.r,rR)rrTrrrr&r0r!r!r"rWs(  z TaskSend.postNr^r!r!r!r"r_r_c@r\) TaskResultcCs|dd}|durt|nd}t|}||std||jd}|r3|j|dd|||n |r=|||| |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)rHr,F)rcrX) get_argumentr7rr'r r,rZr1readyrR)rtaskidrcr&r0r!r!r"rZs    zTaskResult.getNr=r>r?rr[rZr!r!r!r"rbrarbc@r\) TaskAbortcCsJtd|t|}||std||td|dddS)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'rdz Aborted 'rE)messageN)rIinforr'r abortrRr)rrgr&r!r!r"rW@s  zTaskAbort.postNr^r!r!r!r"ri?raric@r\)GetQueueLengthscsv|j}d}|jdkr|jjr|jj}t|jjdd||jjj |jjj d}| | IdH}| 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 NamqpT)include_password)http_apibroker_optionsbroker_use_ssl active_queues) application transportr broker_apirrK connectionas_uriconfbroker_transport_optionsrrqueuesget_active_queue_namesrR)rapprpbrokerr{r!r!r"rZis  zGetQueueLengths.getNrhr!r!r!r"rmhrarmc@r\) ListTasksc Cs2|j}|dd}|jddtd}|dd}|dd}|dd}|d d}|d d}|d d} |d d} |o@t|}t|d}|d krL|nd}|d krT|nd}|d kr\|nd}g} tj|j||| |||||| d D]\} } t| } | dd}|dur|j | d<| | | fqp| 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)defaulttype workernamerTr,received_start received_endsort_bysearchAll) rrrrworkerr,rrrr) rtreintmaxr iter_taskseventsas_dictrhostnameappendrRr)rr}rrrrr,rrrrr&rNrUr!r!r"rZs:\              z ListTasks.getNrhr!r!r!r"rrarc@r\) 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)rtrr, task_typesrR)rseen_task_typesr0r!r!r"rZszListTaskTypes.getNrhr!r!r!r"rrarc@r\)TaskInfocCsRt|jj|}|stdd|d|}|jdur"|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 rCrDrENr) r get_task_by_idrtrr rrrrR)rrgrUr0r!r!r"rZ=s9  z TaskInfo.getNrhr!r!r!r"r<rar)(r9logging collectionsrrceleryrcelery.backends.basercelery.contrib.abortabler celery.resultrtornadortornado.escaper tornado.ioloopr tornado.webr utilsr utils.brokerrr getLoggerr=rIrrAr]r_rbrirmrrrr!r!r!r"s4              :IA41)/~'