U dx>@sddlZddlZddlZddlZddlmZddlmZddlmZddlm Z e e Z Gddde Z Gd d d e ZGd d d e ZGd dde ZGddde ZGddde ZGddde ZGddde ZGddde ZGddde ZGddde ZdS)N)web)gen)util)BaseApiHandlerc@seZdZddZddZdS)ControlHandlercCs|o||jjkS)N)Z applicationworkersself workernamer 6/tmp/pip-unpacked-wheel-3pokl8eb/flower/api/control.py is_workerszControlHandler.is_workerc CsF|D]0}z||ddWStk r2YqXqtd|dS)z$extracts error message from responseerrorzUnknown reasonz(Failed to extract error reason from '%s')getKeyErrorloggerr)r r responserr r r error_reasons zControlHandler.error_reasonN)__name__ __module__ __qualname__rrr r r r rsrc@seZdZejddZdS)WorkerShutDowncCsN||stdd|td||jjjd|gd|t dddS) a Shut down a worker **Example request**: .. sourcecode:: http POST /api/worker/shutdown/celery@worker2 HTTP/1.1 Content-Length: 0 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 29 Content-Type: application/json; charset=UTF-8 { "message": "Shutting down!" } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown worker Unknown worker '%s'zShutting down '%s' workershutdown) destinationzShutting down!messageN) rr HTTPErrorrinfocappcontrol broadcastwritedictr r r r post s   zWorkerShutDown.postNrrrr authenticatedr'r r r r rsrc@seZdZejddZdS)WorkerPoolRestartcCs||stdd|td||jjjdddi|gdd}|rld |d |krl|t d|d n.t || d |d || ||fdS)ac Restart worker's pool **Example request**: .. sourcecode:: http POST /api/worker/pool/restart/celery@worker2 HTTP/1.1 Content-Length: 0 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 56 Content-Type: application/json; charset=UTF-8 { "message": "Restarting 'celery@worker2' worker's pool" } :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: pool restart is not enabled (see CELERYD_POOL_RESTARTS) :statuscode 404: unknown worker rrzRestarting '%s' worker's poolZ pool_restartreloadFT argumentsrreplyokrrz#Failed to restart the '%s' pool: %sN) rrr rr!r"r#r$r%r&r set_statusr)r r rr r r r'Gs&      zWorkerPoolRestart.postNr(r r r r r*Fsr*c@seZdZejddZdS)WorkerPoolGrowcCs||stdd||jddtd}td|||jjj |d|gd}|r|d |d |kr|| t d ||fd n.t || d | d||||fdS)as Grow worker's pool **Example request**: .. sourcecode:: http POST /api/worker/pool/grow/celery@worker2?n=3 HTTP/1.1 Content-Length: 0 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 58 Content-Type: application/json; charset=UTF-8 { "message": "Growing 'celery@worker2' worker's pool by 3" } :query n: number of pool processes to grow, default is 1 :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: failed to grow :statuscode 404: unknown worker rrnrdefaulttypez"Growing '%s' worker's pool by '%s'Tr3r.rr/rz Growing '%s' worker's pool by %srr0z%Failed to grow '%s' worker's pool: %sN)rrr get_argumentintrr!r"r#Z pool_growr%r&rr1rr r r3rr r r r'ys&!     zWorkerPoolGrow.postNr(r r r r r2xsr2c@seZdZejddZdS)WorkerPoolShrinkcCs||stdd||jddtd}td|||jjj |d|gd}|r|d |d |kr|| t d ||fd n.t || d | d||||fdS)ay Shrink worker's pool **Example request**: .. sourcecode:: http POST /api/worker/pool/shrink/celery@worker2 HTTP/1.1 Content-Length: 0 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 60 Content-Type: application/json; charset=UTF-8 { "message": "Shrinking 'celery@worker2' worker's pool by 1" } :query n: number of pool processes to shrink, default is 1 :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: failed to shrink :statuscode 404: unknown worker rrr3rr4z$Shrinking '%s' worker's pool by '%s'Tr7r/rz"Shrinking '%s' worker's pool by %srr0z'Failed to shrink '%s' worker's pool: %sN)rrr r8r9rr!r"r#Z pool_shrinkr%r&rr1rr:r r r r's(!    zWorkerPoolShrink.postNr(r r r r r;sr;c@seZdZejddZdS)WorkerPoolAutoscalecCs||stdd||jdtd}|jdtd}td|||f|jjj d||d|gd d }|rd |d |kr| t d |||fdn.t || d| d||||fdS)a Autoscale worker pool **Example request**: .. sourcecode:: http POST /api/worker/pool/autoscale/celery@worker2?min=3&max=10 HTTP/1.1 Content-Length: 0 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 66 Content-Type: application/json; charset=UTF-8 { "message": "Autoscaling 'celery@worker2' worker (min=3, max=10)" } :query min: minimum number of pool processes :query max: maximum number of pool processes :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: autoscaling is not enabled (see CELERYD_AUTOSCALER) :statuscode 404: unknown worker rrmin)r6maxzAutoscaling '%s' worker by '%s'Z autoscale)r=r>Tr,r/rz(Autoscaling '%s' worker (min=%s, max=%s)rr0z#Failed to autoscale '%s' worker: %sN)rrr r8r9rr!r"r#r$r%r&rr1r)r r r=r>rr r r r's6#     zWorkerPoolAutoscale.postNr(r r r r r<sr<c@seZdZejddZdS)WorkerQueueAddConsumerc Cs||stdd||d}td|||jjjdd|i|gdd}|rd|d |kr| t |d |dd n0t || d | d ||| ||fd S)a Start consuming from a queue **Example request**: .. sourcecode:: http POST /api/worker/queue/add-consumer/celery@worker2?queue=sample-queue Content-Length: 0 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 40 Content-Type: application/json; charset=UTF-8 { "message": "add consumer sample-queue" } :query queue: the name of a new queue :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: failed to add consumer :statuscode 404: unknown worker rrqueuez#Adding consumer '%s' to worker '%s'Z add_consumerTr,r/rrr0z.Failed to add '%s' consumer to '%s' worker: %sNrrr r8rr!r"r#r$r%r&rr1rr r r@rr r r r's,!     zWorkerQueueAddConsumer.postNr(r r r r r?sr?c@seZdZejddZdS)WorkerQueueCancelConsumerc Cs||stdd||d}td|||jjjdd|i|gdd}|rd|d |kr| t |d |dd n0t || d | d ||| ||fd S)a Stop consuming from a queue **Example request**: .. sourcecode:: http POST /api/worker/queue/cancel-consumer/celery@worker2?queue=sample-queue Content-Length: 0 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 52 Content-Type: application/json; charset=UTF-8 { "message": "no longer consuming from sample-queue" } :query queue: the name of queue :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 403: failed to cancel consumer :statuscode 404: unknown worker rrr@z(Canceling consumer '%s' from worker '%s'Zcancel_consumerTr,r/rrr0z3Failed to cancel '%s' consumer from '%s' worker: %sNrArBr r r r'Ss0!     zWorkerQueueCancelConsumer.postNr(r r r r rCRsrCc@seZdZejddZdS) TaskRevokecCsXtd||jddtd}|jddtd}|jjj|||d|t d|d d S) a Revoke a task **Example request**: .. sourcecode:: http POST /api/task/revoke/1480b55c-b8b2-462c-985e-24af3e9158f9?terminate=true Content-Length: 0 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 61 Content-Type: application/json; charset=UTF-8 { "message": "Revoked '1480b55c-b8b2-462c-985e-24af3e9158f9'" } :query terminate: terminate the task if it is running :query signal: name of signal to send to process if terminate (default: 'SIGTERM') :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request zRevoking task '%s' terminateFr4signalSIGTERM)rErFz Revoked '%s'rN) rr!r8boolstrr"r#Zrevoker%r&)r ZtaskidrErFr r r r's zTaskRevoke.postNr(r r r r rDsrDc@seZdZejddZdS) TaskTimoutcCs|d}|jddtd}|jddtd}||jjkrFtdd||dk rh||shtdd|td ||||dk r|gnd}|jj j |d |||d }|rd |d |kr| t |d |d dn*t ||d| d|||dS)a Change soft and hard time limits for a task **Example request**: .. sourcecode:: http POST /api/task/timeout/tasks.sleep HTTP/1.1 Content-Length: 44 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 soft=30&hard=100&workername=celery%40worker1 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 46 Content-Type: application/json; charset=UTF-8 { "message": "time limits set successfully" } :query workername: worker name :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task/worker r hardNr4softrUnknown task '%s'rz'Setting timeouts for '%s' task (%s, %s)T)r.rKrLrr/rrr0zFailed to set timeouts: '%s')r8floatr"tasksrr rrr!r#Z time_limitr%r&rr1r)r tasknamer rKrLrrr r r r's6"     zTaskTimout.postNr(r r r r rJsrJc@seZdZejddZdS) TaskRateLimitcCs|d}|d}||jjkr0tdd||dk rR||sRtdd|td|||dk rn|gnd}|jjj ||d|d }|rd |d |kr| t |d |d d n*t || d | d|||dS)a Change rate limit for a task **Example request**: .. sourcecode:: http POST /api/task/rate-limit/tasks.sleep HTTP/1.1 Content-Length: 41 Content-Type: application/x-www-form-urlencoded; charset=utf-8 Host: localhost:5555 ratelimit=200&workername=celery%40worker1 **Example response**: .. sourcecode:: http HTTP/1.1 200 OK Content-Length: 61 Content-Type: application/json; charset=UTF-8 { "message": "new rate limit set successfully" } :query workername: worker name :reqheader Authorization: optional OAuth token to authenticate :statuscode 200: no error :statuscode 401: unauthorized request :statuscode 404: unknown task/worker r ratelimitrrMNrz%Setting '%s' rate limit for '%s' taskT)r.rr/rrr0zFailed to set rate limit: '%s')r8r"rOrr rrr!r#Z rate_limitr%r&rr1r)r rPr rRrrr r r r's0"      zTaskRateLimit.postNr(r r r r rQsrQ)timeloggingdatetime collectionsZtornadorrrr getLoggerrrrrr*r2r;r<r?rCrDrJrQr r r r s&     '245;67(<