<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 flower REST API
This document shows how to use the flower REST API. 
We will use requests for accessing the API. (See here on how to install it.)    
Code
We'll use the following code throughout the documentation.
tasks.py
Step1:  Running
You'll need a celery worker instance and a flower instance running. In one terminal window run
celery worker --loglevel INFO -A proj -E --autoscale 10,3
and in another terminal run
celery flower -A proj
Tasks API
The tasks API is async, meaning calls will return immediatly and you'll need to poll on task status.
Step2:  async-apply
Step3:  We can see that we created a new task and it's pending. Note that the API is async, meaning it won't wait until the task finish.
apply
For create task and wait results you can use 'apply' API.
Step4:  result
Gets the task result. This is async and will return immediatly even if the task didn't finish (with state 'PENDING')
Step5:  revoke
Revoke a running task.
Step6:  rate-limit
Update rate limit for a task.
Step7:  timeout
Set timeout (both hard and soft) for a task.
Step8:  Worker API
Step9:  workers
List workers.
Step10:  pool/shutdown
Shutdown a worker.
Step11:  pool/restart
Restart a worker pool, you need to have CELERYD_POOL_RESTARTS enabled in your configuration).
Step12:  pool/grow
Grows worker pool.
Step13:  pool/shrink
Shrink worker pool.
Step14:  pool/autoscale
Autoscale a pool.
Step15:  queue/add-consumer
Add a consumer to a queue.
Step16:  queue/cancel-consumer
Cancel a consumer queue.
Step17:  Queue API
We assume that we've two queues; the default one 'celery' and 'all'
<ASSISTANT_TASK:>
Python Code:
from celery import Celery
from time import sleep
celery = Celery()
celery.config_from_object({
    'BROKER_URL': 'amqp://localhost',
    'CELERY_RESULT_BACKEND': 'amqp://',
    'CELERYD_POOL_RESTARTS': True,  # Required for /worker/pool/restart API
})
@celery.task
def add(x, y):
    return x + y
@celery.task
def sub(x, y):
    sleep(30)  # Simulate work
    return x -  y
Explanation: flower REST API
This document shows how to use the flower REST API. 
We will use requests for accessing the API. (See here on how to install it.)    
Code
We'll use the following code throughout the documentation.
tasks.py
End of explanation
# Done once for the whole docs
import requests, json
api_root = 'http://localhost:5555/api'
task_api = '{}/task'.format(api_root)
Explanation: Running
You'll need a celery worker instance and a flower instance running. In one terminal window run
celery worker --loglevel INFO -A proj -E --autoscale 10,3
and in another terminal run
celery flower -A proj
Tasks API
The tasks API is async, meaning calls will return immediatly and you'll need to poll on task status.
End of explanation
args = {'args': [1, 2]}
url = '{}/async-apply/tasks.add'.format(task_api)
print(url)
resp = requests.post(url, data=json.dumps(args))
reply = resp.json()
reply
Explanation: async-apply
End of explanation
args = {'args': [1, 2]}
url = '{}/apply/tasks.add'.format(task_api)
print(url)
resp = requests.post(url, data=json.dumps(args))
reply = resp.json()
reply
Explanation: We can see that we created a new task and it's pending. Note that the API is async, meaning it won't wait until the task finish.
apply
For create task and wait results you can use 'apply' API.
End of explanation
url = '{}/result/{}'.format(task_api, reply['task-id'])
print(url)
resp = requests.get(url)
resp.json()
Explanation: result
Gets the task result. This is async and will return immediatly even if the task didn't finish (with state 'PENDING')
End of explanation
# Run a task
args = {'args': [1, 2]}
resp = requests.post('{}/async-apply/tasks.sub'.format(task_api), data=json.dumps(args))
reply = resp.json()
# Now revoke it
url = '{}/revoke/{}'.format(task_api, reply['task-id'])
print(url)
resp = requests.post(url, data='terminate=True')
resp.json()
Explanation: revoke
Revoke a running task.
End of explanation
worker = 'miki-manjaro'  # You'll need to get the worker name from the worker API (seel below)
url = '{}/rate-limit/{}'.format(task_api, worker)
print(url)
resp = requests.post(url, params={'taskname': 'tasks.add', 'ratelimit': '10'})
resp.json()
Explanation: rate-limit
Update rate limit for a task.
End of explanation
url = '{}/timeout/{}'.format(task_api, worker)
print(url)
resp = requests.post(url, params={'taskname': 'tasks.add', 'hard': '3.14', 'soft': '3'})  # You can omit soft or hard
resp.json()
Explanation: timeout
Set timeout (both hard and soft) for a task.
End of explanation
# Once for the documentation
worker_api = '{}/worker'.format(api_root)
Explanation: Worker API
End of explanation
url = '{}/workers'.format(api_root)  # Only one not under /worker
print(url)
resp = requests.get(url)
workers = resp.json()
workers
Explanation: workers
List workers.
End of explanation
worker = workers.keys()[0]
url = '{}/shutdown/{}'.format(worker_api, worker)
print(url)
resp = requests.post(url)
resp.json()
Explanation: pool/shutdown
Shutdown a worker.
End of explanation
pool_api = '{}/pool'.format(worker_api)
url = '{}/restart/{}'.format(pool_api, worker)
print(url)
resp = requests.post(url)
resp.json()
Explanation: pool/restart
Restart a worker pool, you need to have CELERYD_POOL_RESTARTS enabled in your configuration).
End of explanation
url = '{}/grow/{}'.format(pool_api, worker)
print(url)
resp = requests.post(url, params={'n': '10'})
resp.json()
Explanation: pool/grow
Grows worker pool.
End of explanation
url = '{}/shrink/{}'.format(pool_api, worker)
print(url)
resp = requests.post(url, params={'n': '3'})
resp.json()
Explanation: pool/shrink
Shrink worker pool.
End of explanation
url = '{}/autoscale/{}'.format(pool_api, worker)
print(url)
resp = requests.post(url, params={'min': '3', 'max': '10'})
resp.json()
Explanation: pool/autoscale
Autoscale a pool.
End of explanation
queue_api = '{}/queue'.format(worker_api)
url = '{}/add-consumer/{}'.format(queue_api, worker)
print(url)
resp = requests.post(url, params={'queue': 'jokes'})
resp.json()
Explanation: queue/add-consumer
Add a consumer to a queue.
End of explanation
url = '{}/cancel-consumer/{}'.format(queue_api, worker)
print(url)
resp = requests.post(url, params={'queue': 'jokes'})
resp.json()
Explanation: queue/cancel-consumer
Cancel a consumer queue.
End of explanation
url = '{}/queues/length'.format(api_root)
print(url)
resp = requests.get(url)
resp.json()
Explanation: Queue API
We assume that we've two queues; the default one 'celery' and 'all'
End of explanation
<END_TASK>