o RDiK=@sddlZddlmZddlmZeeZGdddeZGdddeZ Gd d d eZ Gd d d eZ Gd ddeZ GdddeZ GdddeZGdddeZGdddeZGdddeZGdddeZdS)N)web)BaseApiHandlerc@seZdZddZddZdS)ControlHandlercCs|o||jjvS)N) applicationworkersself workernamer K/var/www/Datamplify/venv/lib/python3.10/site-packages/flower/api/control.py is_worker szControlHandler.is_workerc CsB|D]}z ||ddWStyYqwtd|dS)z$extracts error message from responseerrorzUnknown reasonz(Failed to extract error reason from '%s')getKeyErrorloggerr)r r responseresr r r error_reasons  zControlHandler.error_reasonN)__name__ __module__ __qualname__r rr r r r r s rc@eZdZejddZdS)WorkerShutDowncCsR||stdd|dtd||jjjd|gd|t ddd S) 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 ''zShutting down '%s' workershutdown) destinationzShutting down!messageN) r r HTTPErrorrinfocappcontrol broadcastwritedictrr r r posts  zWorkerShutDown.postNrrrr authenticatedr(r r r r rrc@r)WorkerPoolRestartcCs||stdd|dtd||jjjdddi|gdd }|r;d |d |vr;|t d |d ddSt || d| ||}|d|d|dS)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 rrrzRestarting '%s' worker's pool pool_restartreloadFT argumentsrreplyokrz Restarting 'z' worker's poolrzFailed to restart the 'z' pool: N) r rr!rr"r#r$r%r&r'r set_statusr)r r rreasonr r r r(As     zWorkerPoolRestart.postNr)r r r r r,@r+r,c@r)WorkerPoolGrowcC||stdd|d|jddtd}td|||jjj |d|gd }|rCd |d |vrC| t d |d |ddSt || d|||}| d|d|dS)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 rrrnrdefaulttypez"Growing '%s' worker's pool by '%s'Tr8r1rr2rz Growing '' worker's pool by rr3zFailed to grow '' worker's pool: N)r rr! get_argumentintrr"r#r$ pool_growr&r'rr4rr r r8rr5r r r r(q !    zWorkerPoolGrow.postNr)r r r r r6pr+r6c@r)WorkerPoolShrinkcCr7)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 rrrr8rr9z$Shrinking '%s' worker's pool by '%s'Tr<r2rz Shrinking 'r=rr3zFailed to shrink 'r>N)r rr!r?r@rr"r#r$ pool_shrinkr&r'rr4rrBr r r r(rCzWorkerPoolShrink.postNr)r r r r rDr+rDc@r)WorkerPoolAutoscalecCs||stdd|d|jdtd}|jdtd}td|||f|jjj d||d |gd d }|rMd |d |vrM| t d|dddSt || d|||}| d|d|dS)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 rrrmin)r;maxzAutoscaling '%s' worker by '%s' autoscale)rGrHTr/r2rz Autoscaling 'z' worker (min={min}, max={max})rr3zFailed to autoscale ' ' worker: N)r rr!r?r@rr"r#r$r%r&r'rr4r)r r rGrHrr5r r r r(s" #    zWorkerPoolAutoscale.postNr)r r r r rFr+rFc@r)WorkerQueueAddConsumercC||stdd|d|d}td|||jjjdd|i|gdd}|rCd |d |vrC| t |d |d d dSt || d | ||}| d |d|d|dS)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 rrrqueuez#Adding consumer '%s' to worker '%s' add_consumerTr/r2rrr3zFailed to add 'z' consumer to 'rJNr rr!r?rr"r#r$r%r&r'rr4rr r rMrr5r r r r( !      zWorkerQueueAddConsumer.postNr)r r r r rKr+rKc@r)WorkerQueueCancelConsumercCrL)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 rrrrMz(Canceling consumer '%s' from worker '%s'cancel_consumerTr/r2rrr3zFailed to cancel 'z' consumer from 'rJNrOrPr r r r(ErQzWorkerQueueCancelConsumer.postNr)r r r r rRDr+rRc@r) TaskRevokecCs\td||jddtd}|jddtd}|jjj|||d|t d|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' terminateFr9signalSIGTERM)rUrVz Revoked 'rrN) rr"r?boolstrr#r$revoker&r')r taskidrUrVr r r r(zs zTaskRevoke.postNr)r r r r rTyr+rTc@r) TaskTimoutcCs|d}|jddtd}|jddtd}||jjvr%tdd|d|dur8||s8tdd |dtd ||||durG|gnd}|jj j |d |||d }|rod |d|vro| t |d|d ddSt ||d|||}| d|ddS)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 hardNr9softrUnknown task 'rrz'Setting timeouts for '%s' task (%s, %s)T)r1r]r^rr2rrr3zFailed to set timeouts: ')r?floatr#tasksrr!r rr"r$ time_limitr&r'rr4r)r tasknamer r]r^rrr5r r r r(s* "     zTaskTimout.postNr)r r r r r\r+r\c@r) TaskRateLimitcCs|d}|d}||jjvrtdd|d|dur-||s-tdd|dtd|||dur;|gnd}|jjj ||d |d }|rbd |d |vrb| t |d |d d dSt || d|||}| d|ddS)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 ratelimitrr_rNrz%Setting '%s' rate limit for '%s' taskT)r1rr2rrr3zFailed to set rate limit: ')r?r#rarr!r rr"r$ rate_limitr&r'rr4r)r rcr rerrr5r r r r(s& "      zTaskRateLimit.postNr)r r r r rdr+rd)loggingtornadorr getLoggerrrrrr,r6rDrFrKrRrTr\rdr r r r s   '033955(<