python - 我正在尝试将 celery 应用于 django 中基于类的视图(apis)。我怎样才能做到这一点?
问题描述
将芹菜应用于基于类的视图是否正确?而且,如果是,我如何将 celery 应用于基于类的视图?我不能只在类内的函数上方标记@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:\Users\Tonyscoding\Desktop\TOCOL\TOCOL_backend\api\testing\test_pagination.py", line 154, in test_admin_schedule_pagination
response = self.client.get('/api/schedule/by/franchise/simple/1/')
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\test.py", line 286, in get
response = super().get(path, data=data, **extra)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\test.py", line 203, in get
return self.generic('GET', path, **r)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\test.py", line 232, in generic
method, path, data, content_type, secure, **extra)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\test\client.py", line 422, in generic
return self.request(**r)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\test.py", line 283, in request
return super().request(**kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\test.py", line 235, in request
request = super().request(**kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\test\client.py", line 503, in request
raise exc_value
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\core\handlers\exception.py", line 34, in inner
response = get_response(request)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\core\handlers\base.py", line 115, in _get_response
response = self.process_exception_by_middleware(e, request)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\core\handlers\base.py", line 113, in _get_response
response = wrapped_callback(request, *callback_args, **callback_kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\views\decorators\csrf.py", line 54, in wrapped_view
return view_func(*args, **kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\django\views\generic\base.py", line 71, in view
return self.dispatch(request, *args, **kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\views.py", line 505, in dispatch
response = self.handle_exception(exc)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\views.py", line 465, in handle_exception
self.raise_uncaught_exception(exc)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\views.py", line 476, in raise_uncaught_exception
raise exc
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\rest_framework\views.py", line 502, in dispatch
response = handler(request, *args, **kwargs)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\celery\local.py", line 191, in __call__
return self._get_current_object()(*a, **kw)
File "C:\Users\Tonyscoding\Desktop\TOCOL\venv\lib\site-packages\celery\app\task.py", line 392, in __call__
return self.run(*args, **kwargs)
TypeError: get() missing 1 required positional argument: 'request'
我的芹菜工人得到了任务。我认为这不是工人的问题..
解决方案
在您的情况下,下一个场景可能会起作用。为繁重的工作量创建一个任务:
@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',
})
推荐阅读
- .net - 如何获得两个列表,每个列表都包含 F# 记录的必需属性和可选属性的名称?
- r - 如何使用串扰小部件过滤 plot_mapbox() 中的点?
- javascript - VSCode 尝试创建新模块时对装饰器的实验性支持警告
- ios - 根据字符串数组配置collectionView:cellForRowAt
- c# - System.Drawing.Graphics.DrawString 在另一个之上。阴影效果问题
- javascript - 如何隐藏物品?
- css - 为什么即使我清除了缓存,我的 style.css 也没有加载?
- image - 每个像素都将彩色图像转换为 [1,1,1]
- javascript - 使用 MongoDB 和 Mongoose 填充子集合的问题
- python - 用于检查异步代码路径是否执行同步 I/O 的构建时工具