HEX

File: //home/arjun/projects/env/lib/python3.10/site-packages/flower/api/__pycache__/tasks.cpython-310.pyc
o

/we�C�@sZddlZddlZddlmZddlmZddlmZddlmZddl	m
Z
ddlmZddl
mZdd	lmZdd
lmZddlmZdd
lmZddlmZddlmZe�e�ZGdd�de�ZGdd�de�ZGdd�de�Z Gdd�de�Z!Gdd�de�Z"Gdd�de�Z#Gdd�de�Z$Gdd �d e�Z%Gd!d"�d"e�Z&Gd#d$�d$e�Z'dS)%�N)�OrderedDict)�datetime)�states)�DisabledBackend)�AbortableAsyncResult)�AsyncResult)�web)�json_decode)�IOLoop)�	HTTPError�)�tasks)�Broker�)�BaseApiHandlerc@sDeZdZdZdd�Zedd��Zdd�Zdd	�Zd
d�Z	dd
�Z
dS)�BaseTaskHandlerz%Y-%m-%d %H:%M:%S.%fc
Cs�z|jj}|rt|�ni}Wnty"}ztdt|��|�d}~wwt|t�s-tdd��|�dg�}|�di�}t|t	t
f�sEtdd��|||fS)N�zinvalid options�args�kwargszargs must be an array)�request�bodyr	�
ValueErrorr�str�
isinstance�dict�pop�list�tuple)�selfr�options�err�r!�I/home/arjun/projects/env/lib/python3.10/site-packages/flower/api/tasks.py�
get_task_argss��



zBaseTaskHandler.get_task_argscCst|jt�S�N)r�backendr)�resultr!r!r"�backend_configured+sz"BaseTaskHandler.backend_configuredcKs|�|�dSr$)�
set_status)r�status_coderr!r!r"�write_error/szBaseTaskHandler.write_errorcCsD|jtjkr|�|�|j�|jd��dS|�d|�|j�i�dS)N)r&�	tracebackr&)�stater�FAILURE�update�safe_resultr&r+)r�responser&r!r!r"�update_response_result2s
�z&BaseTaskHandler.update_response_resultcCs�d|vrt�|d|j�|d<d|vrt|d�|d<d|vr@|d}zt|�}Wnty9t�||j�}Ynw||d<dSdS)N�eta�	countdown�expires)r�strptime�DATE_FORMAT�floatr)rrr4r!r!r"�normalize_options9s
���z!BaseTaskHandler.normalize_optionscCs,zt�|�W|Styt|�YSw)zreturns json encodable result)�json�dumps�	TypeError�repr)rr&r!r!r"r/Gs��zBaseTaskHandler.safe_resultN)�__name__�
__module__�__qualname__r6r#�staticmethodr'r*r1r8r/r!r!r!r"rs
rc@s"eZdZejdd��Zdd�ZdS)�	TaskApplyc	
�s��|��\}}}t�d|||�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 '�'Nr�Invalid option�rr�task-idr!)r#�logger�debug�cappr
�KeyErrorrr8r�apply_async�task_idr
�current�run_in_executor�wait_results�write�	r�tasknamerrr�task�excr&r0r!r!r"�postQs.�)�����


�zTaskApply.postcCs4|jdd�|�||�|�|�r|j|jd�|S)NF)�	propagate�r,)�getr1r'r.r,)rr&r0r!r!r"rQ�s

zTaskApply.wait_resultsN)r=r>r?r�
authenticatedrWrQr!r!r!r"rAPs
=rAc@�eZdZejdd��ZdS)�TaskAsyncApplyc	
Cs�|��\}}}t�d|||�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
rLrr8rrMrNr'r.r,rRrSr!r!r"rW�s**�����

zTaskAsyncApply.postN�r=r>r?rr[rWr!r!r!r"r]�sr]c@r\)�TaskSendcCsh|��\}}}t�d|||�|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"rW�s(����

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'rr,rZr1�readyrR)r�taskidrcr&r0r!r!r"rZs 
zTaskResult.getN�r=r>r?rr[rZr!r!r!r"rbrarbc@r\)�	TaskAbortcCsJt�d|�t|�}|�|�std��|��|�td|�d�d��dS)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)rI�inforr'r�abortrRr)rrgr&r!r!r"rW@s
zTaskAbort.postNr^r!r!r!r"ri?raric@r\)�GetQueueLengthsc�sv�|j}d}|jdkr|jjr|jj}t|j��jdd�||jjj	|jjj
d�}|�|���IdH}|�
d|i�dS)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
        N�amqpT)�include_password)�http_api�broker_options�broker_use_ssl�
active_queues)�application�	transportr�
broker_apirrK�
connection�as_uri�conf�broker_transport_optionsrr�queues�get_active_queue_namesrR)r�apprp�brokerr{r!r!r"rZis� 
�zGetQueueLengths.getNrhr!r!r!r"rmhrarmc@r\)�	ListTaskscCs2|j}|�dd�}|jddtd�}|�dd�}|�dd�}|�dd�}|�d	d�}|�d
d�}|�dd�}	|�dd�}
|o@t|�}t|d�}|d
krL|nd}|d
krT|nd}|d
kr\|nd}g}tj|j|||	||||||
d�
D]\}}
t�|
�}
|
�dd�}|dur�|j	|
d<|�
||
f�qp|�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
        �limitN�offsetr)�default�type�
workernamerTr,�received_start�received_end�sort_by�search�All)	r�r�r�r��workerr,r�r�r�r�)
rtre�int�maxr
�
iter_tasks�events�as_dictr�hostname�appendrRr)rr}r�r�r�r�r,r�r�r�r�r&rNrUr!r!r"rZ�s:\
�

z
ListTasks.getNrhr!r!r!r"r�rarc@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)rtr�r,�
task_typesrR)r�seen_task_typesr0r!r!r"rZszListTaskTypes.getNrhr!r!r!r"r�rar�c@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_idrtr�rr�r�r�rR)rrgrUr0r!r!r"rZ=s9
zTaskInfo.getNrhr!r!r!r"r�<rar�)(r9�logging�collectionsrr�celeryr�celery.backends.baser�celery.contrib.abortabler�
celery.resultr�tornador�tornado.escaper	�tornado.ioloopr
�tornado.webr�utilsr
�utils.brokerr�r�	getLoggerr=rIrrAr]r_rbrirmrr�r�r!r!r!r"�<module>s4
:IA41)/~'