我正在尝试将芹菜应用于 django 中基于类的视图 (apis)。我该怎么做？

Question

将芹菜应用于基于类的视图是正确的方法吗？如果是的话，我如何将芹菜应用于基于类的视图？我不能只在类中标记@app.task以上函数。

class ScheduleByFranchiseIdView(generics.RetrieveAPIView):
permission_classes = (IsAdmin,)
serializer_class = ScheduleSerializer
@app2.task
def get(self, request, franchise_id, start = None, end = None):
if start != None and end != None:
query1 = Q(student__profile__franchise__exact=franchise_id)
query2 = Q(start_time__gte=start)
query3 = Q(end_time__lt=end)
queryset = Schedule.objects.filter(query1 & query2 & query3).exclude(status=ScheduleStatus.DELETED).order_by('-id')
serializer = ScheduleSerializer(queryset, many=True)
else:
query1 = Q(student__profile__franchise__exact=franchise_id)
queryset = Schedule.objects.filter(query1).exclude(status=ScheduleStatus.DELETED).order_by('-id')
serializer = ScheduleSerializer(queryset, many=True)
return Response(serializer.data)

我正在尝试测试这个api，当我调用HTTP GET方法来调用这个api时，我得到下面的错误：

Traceback (most recent call last):
File "C:UsersTonyscodingDesktopTOCOLTOCOL_backendapitestingtest_pagination.py", line 154, in test_admin_schedule_pagination
response = self.client.get('/api/schedule/by/franchise/simple/1/')
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworktest.py", line 286, in get
response = super().get(path, data=data, **extra)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworktest.py", line 203, in get
return self.generic('GET', path, **r)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworktest.py", line 232, in generic
method, path, data, content_type, secure, **extra)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangotestclient.py", line 422, in generic
return self.request(**r)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworktest.py", line 283, in request
return super().request(**kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworktest.py", line 235, in request
request = super().request(**kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangotestclient.py", line 503, in request
raise exc_value
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangocorehandlersexception.py", line 34, in inner
response = get_response(request)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangocorehandlersbase.py", line 115, in _get_response
response = self.process_exception_by_middleware(e, request)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangocorehandlersbase.py", line 113, in _get_response
response = wrapped_callback(request, *callback_args, **callback_kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangoviewsdecoratorscsrf.py", line 54, in wrapped_view
return view_func(*args, **kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesdjangoviewsgenericbase.py", line 71, in view
return self.dispatch(request, *args, **kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworkviews.py", line 505, in dispatch
response = self.handle_exception(exc)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworkviews.py", line 465, in handle_exception
self.raise_uncaught_exception(exc)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworkviews.py", line 476, in raise_uncaught_exception
raise exc
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesrest_frameworkviews.py", line 502, in dispatch
response = handler(request, *args, **kwargs)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagescelerylocal.py", line 191, in __call__
return self._get_current_object()(*a, **kw)
File "C:UsersTonyscodingDesktopTOCOLvenvlibsite-packagesceleryapptask.py", line 392, in __call__
return self.run(*args, **kwargs)
TypeError: get() missing 1 required positional argument: 'request'

我的芹菜工人得到了任务。我认为这不是工人的问题。。

Answer 1

在您的情况下，下一个场景可能会工作。为繁重的工作量创建任务：

@app.task
def schedule_by_franchise(franchise_id, start=None, end=None):
# Do some slow workload, filtering by non-indexed fields or something.
if start is not None and end is not None:  # is not None ~20% faster than != None
query1 = Q(student__profile__franchise__exact=franchise_id)
query2 = Q(start_time__gte=start)
query3 = Q(end_time__lt=end)
queryset = Schedule.objects.filter(query1 & query2 & query3).exclude(status=ScheduleStatus.DELETED).order_by('-id')
else:
query1 = Q(student__profile__franchise__exact=franchise_id)
queryset = Schedule.objects.filter(query1).exclude(status=ScheduleStatus.DELETED).order_by('-id')
# Returns something serializable and what could be used for more faster DB search (founded object primary keys might fits)
return tuple(queryset.values_list('id', flat=True))

当第一次执行GET时，您应该创建Celery任务，然后将其task_ID保存在某个位置，以便稍后获得结果：

from celery.result import AsyncResult

class ScheduleByFranchiseIdView(generics.RetrieveAPIView):
permission_classes = (IsAdmin,)
serializer_class = ScheduleSerializer
def get(self, request, franchise_id, start=None, end=None, task_id=None):
if not task_id:
task = schedule_by_franchise.delay(franchise_id, start, end)
return Response({
'task': task.task_id,
'status': 'processing',
'message': f'Please, try again in 10 seconds with following task_id={task.task_id}',
})
else:
result = AsyncResult(task_id)
if result.ready():
ids = result.result
queryset = Schedule.objects.filter(id__in=ids)
serializer = ScheduleSerializer(queryset, many=True)
return Response(serializer.data)
else:
return Response({
'status': 'not_ready_yet',
'message': 'Please, try again in 5 seconds',
})