Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- @shared_task
- def post_notification(data,url):
- url = "http://posttestserver.com/data/?dir=praful" # when in production, remove this line.
- headers = {'content-type': 'application/json'}
- requests.post(url, data=json.dumps(data), headers=headers)
- @shared_task
- def shipment_server(data,notification_type):
- notification_obj = Notification.objects.get(name = notification_type)
- server_list = ServerNotificationMapping.objects.filter(notification_name=notification_obj)
- for server in server_list:
- task = post_notification.delay(data,server.server_id.url)
- print task.status # it prints 'Nonetype' has no attribute id
- for server in server_list:
- task = group(post_notification.s(data, server.server_id.url))().get()
- print task.status
- TxIsolationWarning: Polling results w│
- ith transaction isolation level repeatable-read within the same transacti│
- on may give outdated results. Be sure to commit the transaction for each │
- poll iteration. │
- 'Polling results with transaction isolation level '
- celery.current_app.send_task('mymodel.tasks.mytask', args=[arg1, arg2, arg3])
- from app.tasks import celery_add_task
- celery_add_task.apply_async(args=[task_name])
- from celery import group
- @shared_task
- def shipment_server(data,notification_type):
- notification_obj = Notification.objects.get(name = notification_type)
- server_list = ServerNotificationMapping.objects.filter(notification_name=notification_obj)
- tasks = [post_notification.s(data, server.server_id.url) for server in server_list]
- results = group(tasks)()
- print results.get() # results.status() what ever you want
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement