@dog_stats_api.timed('status.service.celery.ping')
def celery_ping(_):
 start = time.time()
 result = delayed_ping.apply_async(('ping', 0.1))
 task_id = result.id
 try:
  value = result.get(timeout=4.0)
  success = True
 except TimeoutError:
  value = None
  success = False
 output = {'success': success, 'task_id': task_id, 'value': value, 'time': (time.time() - start)}
 return HttpResponse(json.dumps(output, indent=4), content_type='application/json')

@ dog_stats_api . timed ( 'status.service.celery.ping' ) def celery_ping ( _ ) : start = time . time ( ) result = delayed_ping . apply_async ( ( 'ping' , 0.1 ) ) task_id = result . id try : value = result . get ( timeout = 4.0 ) success = True except TimeoutError : value = None success = False output = { 'success' : success , 'task_id' : task_id , 'value' : value , 'time' : ( time . time ( ) - start ) } return HttpResponse ( json . dumps ( output , indent = 4 ) , content_type = 'application/json' )
a simple view that checks if celery can process a simple task .