Refactor RQId / RQIdManager

* Merge these into one class. These classes clearly deal with the same
  concept, so it doesn't make sense to divide the logic into two classes.

* Turn `build` into an instance method (`render`). That way, the validation
  logic can be reused between it and the `RQId` constructor. Adjust the
  fields so that the first three fields can be specified as positional
  arguments.

* Make the class frozen (I don't see a compelling case to mutate it).

* Change string fields into corresponding enums. This reduces the amount
  of hardcoded string literals everywhere. Note that I had to move the
  enums into the `models` module to avoid a circular import.

* Rename `resource` to `target`, because that's the name of the enum
  and the corresponding field in the API.

Author: SpecLad

cvat/apps/engine/background_operations.py

@@ -28,9 +28,11 @@
 from cvat.apps.engine.cloud_provider import export_resource_to_cloud_storage
 from cvat.apps.engine.location import StorageType, get_location_configuration
 from cvat.apps.engine.log import ServerLogManager
-from cvat.apps.engine.models import Location, Project, Task
+from cvat.apps.engine.models import (
+    Location, Project, Task, RequestAction, RequestTarget, RequestSubresource,
+)
 from cvat.apps.engine.permissions import get_cloud_storage_for_import_or_export
-from cvat.apps.engine.rq_job_handler import RQIdManager
+from cvat.apps.engine.rq_job_handler import RQId
 from cvat.apps.engine.serializers import RqIdSerializer
 from cvat.apps.engine.utils import (
     build_annotations_file_name,
@@ -342,14 +344,15 @@ def export(self) -> Response:
             return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
 
         queue: DjangoRQ = django_rq.get_queue(self.QUEUE_NAME)
-        rq_id = RQIdManager.build(
-            "export",
-            self.resource,
+        rq_id = RQId(
+            RequestAction.EXPORT,
+            RequestTarget(self.resource),
             self.db_instance.pk,
-            subresource="dataset" if self.export_args.save_images else "annotations",
-            anno_format=self.export_args.format,
+            subresource=RequestSubresource.DATASET
+                if self.export_args.save_images else RequestSubresource.ANNOTATIONS,
+            format=self.export_args.format,
             user_id=self.request.user.id,
-        )
+        ).render()
 
         rq_job = queue.fetch_job(rq_id)
 
@@ -580,13 +583,13 @@ def _handle_rq_job_v1(
 
     def export(self) -> Response:
         queue: DjangoRQ = django_rq.get_queue(self.QUEUE_NAME)
-        rq_id = RQIdManager.build(
-            "export",
-            self.resource,
+        rq_id = RQId(
+            RequestAction.EXPORT,
+            RequestTarget(self.resource),
             self.db_instance.pk,
-            subresource="backup",
+            subresource=RequestSubresource.BACKUP,
             user_id=self.request.user.id,
-        )
+        ).render()
         rq_job = queue.fetch_job(rq_id)
 
         if rq_job:

cvat/apps/engine/backup.py

@@ -39,9 +39,11 @@
     get_rq_job_meta, import_resource_with_clean_up_after,
     define_dependent_job, get_rq_lock_by_user,
 )
-from cvat.apps.engine.rq_job_handler import RQIdManager, RQJobMetaField
+from cvat.apps.engine.rq_job_handler import RQId, RQJobMetaField
 from cvat.apps.engine.models import (
-    StorageChoice, StorageMethodChoice, DataChoice, Project, Location)
+    StorageChoice, StorageMethodChoice, DataChoice, Project, Location,
+    RequestAction, RequestTarget, RequestSubresource,
+)
 from cvat.apps.engine.task import JobFileMapping, _create_thread
 from cvat.apps.engine.cloud_provider import import_resource_from_cloud_storage
 from cvat.apps.engine.location import StorageType, get_location_configuration
@@ -1013,7 +1015,10 @@ def import_project(request, queue_name, filename=None):
     if 'rq_id' in request.data:
         rq_id = request.data['rq_id']
     else:
-        rq_id = RQIdManager.build('import', 'project', uuid.uuid4(), subresource='backup')
+        rq_id = RQId(
+            RequestAction.IMPORT, RequestTarget.PROJECT, uuid.uuid4(),
+            subresource=RequestSubresource.BACKUP,
+        ).render()
     Serializer = ProjectFileSerializer
     file_field_name = 'project_file'
 
@@ -1036,7 +1041,10 @@ def import_project(request, queue_name, filename=None):
     )
 
 def import_task(request, queue_name, filename=None):
-    rq_id = request.data.get('rq_id', RQIdManager.build('import', 'task', uuid.uuid4(), subresource='backup'))
+    rq_id = request.data.get('rq_id', RQId(
+        RequestAction.IMPORT, RequestTarget.TASK, uuid.uuid4(),
+        subresource=RequestSubresource.BACKUP,
+    ).render())
     Serializer = TaskFileSerializer
     file_field_name = 'task_file'
 

cvat/apps/engine/mixins.py

@@ -33,8 +33,8 @@
 from cvat.apps.engine.handlers import clear_import_cache
 from cvat.apps.engine.location import StorageType, get_location_configuration
 from cvat.apps.engine.log import ServerLogManager
-from cvat.apps.engine.models import Location
-from cvat.apps.engine.rq_job_handler import RQIdManager
+from cvat.apps.engine.models import Location, RequestAction, RequestTarget, RequestSubresource
+from cvat.apps.engine.rq_job_handler import RQId
 from cvat.apps.engine.serializers import DataSerializer, RqIdSerializer
 from cvat.apps.engine.utils import is_dataset_export
 
@@ -275,7 +275,10 @@ def init_tus_upload(self, request):
             if file_exists:
                 # check whether the rq_job is in progress or has been finished/failed
                 object_class_name = self._object.__class__.__name__.lower()
-                template = RQIdManager.build('import', object_class_name, self._object.pk, subresource=import_type)
+                template = RQId(
+                    RequestAction.IMPORT, RequestTarget(object_class_name), self._object.pk,
+                    subresource=RequestSubresource(import_type)
+                ).render()
                 queue = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
                 finished_job_ids = queue.finished_job_registry.get_job_ids()
                 failed_job_ids = queue.failed_job_registry.get_job_ids()
@@ -473,7 +476,7 @@ def export_dataset_v2(self, request: HttpRequest, pk: int):
         return dataset_export_manager.export()
 
     # FUTURE-TODO: migrate to new API
-    def import_annotations(self, request, db_obj, import_func, rq_func, rq_id_template):
+    def import_annotations(self, request, db_obj, import_func, rq_func, rq_id_factory):
         is_tus_request = request.headers.get('Upload-Length', None) is not None or \
             request.method == 'OPTIONS'
         if is_tus_request:
@@ -492,7 +495,7 @@ def import_annotations(self, request, db_obj, import_func, rq_func, rq_id_templa
 
             return import_func(
                 request=request,
-                rq_id_template=rq_id_template,
+                rq_id_factory=rq_id_factory,
                 rq_func=rq_func,
                 db_obj=self._object,
                 format_name=format_name,

cvat/apps/engine/models.py

@@ -19,7 +19,7 @@
 from django.core.exceptions import ValidationError
 from django.db import IntegrityError, models, transaction
 from django.db.models.fields import FloatField
-from django.db.models import Q
+from django.db.models import Q, TextChoices
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema_field
 from cvat.apps.engine.lazy_list import LazyList
@@ -1196,3 +1196,24 @@ def organization_id(self):
 
     def get_asset_dir(self):
         return os.path.join(settings.ASSETS_ROOT, str(self.uuid))
+
+class RequestStatus(TextChoices):
+    QUEUED = "queued"
+    STARTED = "started"
+    FAILED = "failed"
+    FINISHED = "finished"
+
+class RequestAction(TextChoices):
+    CREATE = "create"
+    IMPORT = "import"
+    EXPORT = "export"
+
+class RequestTarget(TextChoices):
+    PROJECT = "project"
+    TASK = "task"
+    JOB = "job"
+
+class RequestSubresource(TextChoices):
+    ANNOTATIONS = "annotations"
+    DATASET = "dataset"
+    BACKUP = "backup"

cvat/apps/engine/permissions.py

@@ -1227,7 +1227,7 @@ def create(cls, request, view, obj: Optional[RQJob], iam_context: Dict):
                         ('export', 'task', 'backup'): (TaskPermission, TaskPermission.Scopes.EXPORT_BACKUP),
                         ('export', 'job', 'annotations'): (JobPermission, JobPermission.Scopes.EXPORT_ANNOTATIONS),
                         ('export', 'job', 'dataset'): (JobPermission, JobPermission.Scopes.EXPORT_DATASET),
-                    }[(parsed_rq_id.action, parsed_rq_id.resource, parsed_rq_id.subresource)]
+                    }[(parsed_rq_id.action, parsed_rq_id.target, parsed_rq_id.subresource)]
 
 
                     resource = None
@@ -1236,12 +1236,12 @@ def create(cls, request, view, obj: Optional[RQJob], iam_context: Dict):
                             'project': Project,
                             'task': Task,
                             'job': Job,
-                        }[parsed_rq_id.resource]
+                        }[parsed_rq_id.target]
 
                         try:
                             resource = resource_model.objects.get(id=resource_id)
                         except resource_model.DoesNotExist as ex:
-                            raise NotFound(f'The {parsed_rq_id.resource!r} with specified id#{resource_id} does not exist') from ex
+                            raise NotFound(f'The {parsed_rq_id.target!r} with specified id#{resource_id} does not exist') from ex
 
                     permissions.append(permission_class.create_base_perm(request, view, scope=resource_scope, iam_context=iam_context, obj=resource))
 

cvat/apps/engine/rq_job_handler.py

@@ -2,12 +2,16 @@
 #
 # SPDX-License-Identifier: MIT
 
+from __future__ import annotations
+
 import attrs
 
 from typing import Optional, Union
 from uuid import UUID
 from rq.job import Job as RQJob
 
+from .models import RequestAction, RequestTarget, RequestSubresource
+
 class RQJobMetaField:
     # common fields
     FORMATTED_EXCEPTION = "formatted_exception"
@@ -28,82 +32,85 @@ class RQJobMetaField:
 def is_rq_job_owner(rq_job: RQJob, user_id: int) -> bool:
     return rq_job.meta.get(RQJobMetaField.USER, {}).get('id') == user_id
 
-@attrs.define(kw_only=True)
+@attrs.frozen()
 class RQId:
-    action: str = attrs.field(
-        validator=attrs.validators.instance_of(str)
+    action: RequestAction = attrs.field(
+        validator=attrs.validators.instance_of(RequestAction)
     )
-    resource: str = attrs.field(
-        validator=attrs.validators.instance_of(str)
+    target: RequestTarget = attrs.field(
+        validator=attrs.validators.instance_of(RequestTarget)
     )
     identifier: Union[int, UUID] = attrs.field(
         validator=attrs.validators.instance_of((int, UUID))
     )
-    subresource: Optional[str] = attrs.field(
+    subresource: Optional[RequestSubresource] = attrs.field(
         validator=attrs.validators.optional(
-            attrs.validators.instance_of(str)
-        )
+            attrs.validators.instance_of(RequestSubresource)
+        ),
+        kw_only=True, default=None,
     )
     user_id: Optional[int] = attrs.field(
-        validator=attrs.validators.optional(attrs.validators.instance_of(int))
+        validator=attrs.validators.optional(attrs.validators.instance_of(int)),
+        kw_only=True, default=None,
     )
     format: Optional[str] = attrs.field(
-        validator=attrs.validators.optional(attrs.validators.instance_of(str))
+        validator=attrs.validators.optional(attrs.validators.instance_of(str)),
+        kw_only=True, default=None,
     )
 
+    def __attrs_post_init__(self) -> None:
+        if self.action == RequestAction.CREATE and self.target != RequestTarget.TASK:
+            raise ValueError(f"Unsupported target {self.target!r} for the {self.action!r} action")
 
-class RQIdManager:
     # RQ ID templates:
     # import:<task|project|job>-<id|uuid>-<annotations|dataset|backup>
     # create:task-<tid>
     # export:<project|task|job>-<id>-<annotations|dataset>-in-<format>-format-by-<user_id>
     # export:<project|task>-<id>-backup-by-<user_id>
 
-    @staticmethod
-    def build(
-        action: str,
-        resource: str,
-        identifier: Union[int, UUID],
-        *,
-        subresource: Optional[str] = None,
-        user_id: Optional[int] = None,
-        anno_format: Optional[str] = None,
+    def render(
+        self,
     ) -> str:
-        if "import" == action:
-            return f"{action}:{resource}-{identifier}-{subresource}"
-        elif "export" == action:
-            if anno_format is None:
+        common_prefix = f"{self.action}:{self.target}-{self.identifier}"
+
+        if RequestAction.IMPORT == self.action:
+            return f"{common_prefix}-{self.subresource}"
+        elif RequestAction.EXPORT == self.action:
+            if self.format is None:
                 return (
-                    f"{action}:{resource}-{identifier}-{subresource}-by-{user_id}"
+                    f"{common_prefix}-{self.subresource}-by-{self.user_id}"
                 )
-            format_to_be_used_in_urls = anno_format.replace(" ", "_").replace(".", "@")
-            return f"{action}:{resource}-{identifier}-{subresource}-in-{format_to_be_used_in_urls}-format-by-{user_id}"
-        elif "create" == action:
-            assert "task" == resource
-            return f"{action}:{resource}-{identifier}"
+
+            format_to_be_used_in_urls = self.format.replace(" ", "_").replace(".", "@")
+            return f"{common_prefix}-{self.subresource}-in-{format_to_be_used_in_urls}-format-by-{self.user_id}"
+        elif RequestAction.CREATE == self.action:
+            return common_prefix
         else:
-            raise ValueError(f"Unsupported action {action!r} was found")
+            assert False, f"Unsupported action {self.action!r} was found"
 
     @staticmethod
     def parse(rq_id: str) -> RQId:
-        action: Optional[str] = None
-        resource: Optional[str] = None
         identifier: Optional[Union[UUID, int]] = None
-        subresource: Optional[str] = None
+        subresource: Optional[RequestSubresource] = None
         user_id: Optional[int] = None
         anno_format: Optional[str] = None
 
         try:
             action_and_resource, unparsed = rq_id.split("-", maxsplit=1)
-            action, resource = action_and_resource.split(":")
+            action_str, target_str = action_and_resource.split(":")
+            action = RequestAction(action_str)
+            target = RequestTarget(target_str)
 
-            if "create" == action:
+            if RequestAction.CREATE == action:
                 identifier = unparsed
-            elif "import" == action:
-                identifier, subresource = unparsed.rsplit("-", maxsplit=1)
+            elif RequestAction.IMPORT == action:
+                identifier, subresource_str = unparsed.rsplit("-", maxsplit=1)
+                subresource = RequestSubresource(subresource_str)
             else: # action == export
-                identifier, subresource, unparsed = unparsed.split("-", maxsplit=2)
-                if "backup" == subresource:
+                identifier, subresource_str, unparsed = unparsed.split("-", maxsplit=2)
+                subresource = RequestSubresource(subresource_str)
+
+                if RequestSubresource.BACKUP == subresource:
                     _, user_id = unparsed.split("-")
                 else:
                     unparsed, _, user_id = unparsed.rsplit("-", maxsplit=2)
@@ -122,7 +129,7 @@ def parse(rq_id: str) -> RQId:
 
             return RQId(
                 action=action,
-                resource=resource,
+                target=target,
                 identifier=identifier,
                 subresource=subresource,
                 user_id=user_id,