我试图执行一个周期性任务,所以我将 celery 与 Django 1.8 和 Django Rest Framework 以及 Postgres 一起用作数据库。当我尝试将我的 obj 发送到任务时,我得到了 TypeError: foreign_model_obj is not JSON serializable
。如何将我的查询集对象传递给我的任务。
views.py :
class MyModelCreateApiView(generics.CreateAPIView):
queryset = MyModel.objects.all()
serializer_class = MyModelSerializer
authentication_classes = (TokenAuthentication,)
def create(self, request, *args, **kwargs):
data = dict()
data['foreign_model_id'] = kwargs['pk']
foreign_model_obj = MyForeignModel.objects.get(id=data['foreign_model_id'])
obj = MyModel.objects.create(**data)
result = serialize_query(MyModel, {"id": obj.id})
local_time = foreign_model_obj.time
my_celery_task.apply_async([foreign_model_obj], eta=local_time)
return Response(result)
任务.py :
@celery_app.task(name="my_celery_task")
def my_first_celery_task(mymodel_obj):
# ... updating obj attributes
mymodel_obj.save()
最佳答案
您只需发送实例的 id
并在任务中检索对象。
传递实例是一种不好的做法,因为它可以同时更改,特别是当你正在执行你的任务时,它似乎是这样。
views.py :
class MyModelCreateApiView(generics.CreateAPIView):
queryset = MyModel.objects.all()
serializer_class = MyModelSerializer
authentication_classes = (TokenAuthentication,)
def create(self, request, *args, **kwargs):
data = dict()
data['foreign_model_id'] = kwargs['pk']
foreign_model_obj = MyForeignModel.objects.get(id=data['foreign_model_id'])
obj = MyModel.objects.create(**data)
result = serialize_query(MyModel, {"id": obj.id})
local_time = foreign_model_obj.time
my_celery_task.apply_async([foreign_model_obj.id], eta=local_time) # send only the obj id
return Response(result)
任务.py :
@celery_app.task(name="my_celery_task")
def my_celery_task(mymodel_obj_id):
my_model_obj = MyModel.objects.get(id=mymodel_obj_id) # retrieve your object here
# ... updating obj attributes
mymodel_obj.save()
关于python - Celery 在将我的查询集 obj 作为参数传递时引发错误,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/34765276/