Блог

У меня есть пара задач сельдерея, которые выполняются таким же образом. Доступ к ним осуществляется через конечные точки, и я хочу централизовать их создание.

В идеале я хотел бы определить аргумент класса celery_tasks_endpoints , который содержит набор кортежей с именем конечной точки и задачей сельдерея.

 # assets/views.py

from django.db.models import QuerySet

from rest_framework.viewsets import ModelViewSet

from tasks.views import CeleryTaskViewSetMixin

from .filters import AssetFilterSet
from .models import Asset
from .serializers import AssetSerializer
from .tasks import cei_assets_crawler, cei_passive_incomes_crawler


class AssetViewSet(ModelViewSet, CeleryTaskViewSetMixin):
    serializer_class = AssetSerializer
    filterset_class = AssetFilterSet
    celery_tasks_endpoints = (
        ("fetch_cei_assets", cei_assets_crawler),
        ("fetch_cei_passive_incomes", cei_passive_incomes_crawler),
    )

    def get_queryset(self) -> QuerySet:
        if self.request.user.is_authenticated:
            return self.request.user.assets.all()
        return Asset.objects.none()  # drf-spectatular
 

Создание конечных точек будет зависеть от CeleryTaskViewSetMixin :

 from typing import Callable, Type

from celery import Task as CeleryTask
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.status import HTTP_200_OK
from rest_framework.viewsets import ViewSet


def create_endpoint(task: CeleryTask, endpoint_name: str) -> Callable:
    @action(methods=("GET",), detail=False)
    def task_endpoint(self, request: Request) -> Response:
        task.delay(username=request.user.username)
        return Response(status=HTTP_200_OK)

    task_endpoint.__name__ = endpoint_name
    return task_endpoint


class CeleryTaskViewSetMixin:
    def __new__(cls: Type[ViewSet], *args, **kwargs) -> ViewSet:
        for endpoint_name, task in cls.celery_tasks_endpoints:
            setattr(
                cls,
                endpoint_name,
                create_endpoint(task=task, endpoint_name=endpoint_name),
            )
        return super().__new__(cls)
 

Что меня интригует, так это то, что AssetViewSet.get_extra_actions() возвращает два метода:

 [<function create_endpoint.<locals>.task_endpoint at 0x7fab5e6938b0>, <function create_endpoint.<locals>.task_endpoint at 0x7fab5e6b2160>]
 

And I can actually run the task if I mock a request:

 AssetViewSet().fetch_cei_assets(request=Request(HttpRequest()))  # this executes the task!
 

Но a DefaultRouter не создает никакого URL-адреса для этих действий.

Я попытался перейти __new__ прямо к ViewSet , но это не сработало.

Возможно, мой подход не самый лучший, и у вас есть идея получше, но как я могу динамически создавать такие конечные точки?

Большое спасибо!