Refactor way of working with RQ meta (#9082)

This PR changes the way we work with RQ meta. The previous
implementation tried to standardize the fields saved in the meta and how
to access them, however, we had to remember the structure of nested
objects. This approach explicitly defines the structure of the meta
being used.

cvat/apps/consensus/merging_manager.py

@@ -27,8 +27,9 @@
     User,
     clear_annotations_in_jobs,
 )
+from cvat.apps.engine.rq import BaseRQMeta, define_dependent_job
 from cvat.apps.engine.types import ExtendedRequest
-from cvat.apps.engine.utils import define_dependent_job, get_rq_job_meta, get_rq_lock_by_user
+from cvat.apps.engine.utils import get_rq_lock_by_user
 from cvat.apps.profiler import silk_profile
 from cvat.apps.quality_control.quality_reports import ComparisonParameters, JobDataProvider
 
@@ -224,7 +225,7 @@ def schedule_merge(self, target: Task | Job, *, request: ExtendedRequest) -> str
                 target_type=type(target),
                 target_id=target.id,
                 job_id=rq_id,
-                meta=get_rq_job_meta(request=request, db_obj=target),
+                meta=BaseRQMeta.build(request=request, db_obj=target),
                 result_ttl=self._JOB_RESULT_TTL,
                 failure_ttl=self._JOB_RESULT_TTL,
                 depends_on=dependency,

cvat/apps/consensus/views.py

@@ -25,7 +25,7 @@
 )
 from cvat.apps.engine.mixins import PartialUpdateModelMixin
 from cvat.apps.engine.models import Job, Task
-from cvat.apps.engine.rq_job_handler import RQJobMetaField
+from cvat.apps.engine.rq import BaseRQMeta
 from cvat.apps.engine.serializers import RqIdSerializer
 from cvat.apps.engine.types import ExtendedRequest
 from cvat.apps.engine.utils import process_failed_job
@@ -107,7 +107,7 @@ def create(self, request: ExtendedRequest, *args, **kwargs):
             if (
                 not rq_job
                 or not ConsensusMergePermission.create_scope_check_status(
-                    request, rq_job_owner_id=rq_job.meta[RQJobMetaField.USER]["id"]
+                    request, rq_job_owner_id=BaseRQMeta.for_job(rq_job).user.id
                 )
                 .check_access()
                 .allow

cvat/apps/dataset_manager/bindings.py

@@ -45,7 +45,7 @@
     ShapeType,
     Task,
 )
-from cvat.apps.engine.rq_job_handler import RQJobMetaField
+from cvat.apps.engine.rq import ImportRQMeta
 
 from ..engine.log import ServerLogManager
 from .annotation import AnnotationIR, AnnotationManager, TrackManager
@@ -2452,9 +2452,10 @@ def load_dataset_data(project_annotation, dataset: dm.Dataset, project_data):
                 raise CvatImportError(f'Target project does not have label with name "{label.name}"')
     for subset_id, subset in enumerate(dataset.subsets().values()):
         job = rq.get_current_job()
-        job.meta[RQJobMetaField.STATUS] = 'Task from dataset is being created...'
-        job.meta[RQJobMetaField.PROGRESS] = (subset_id + job.meta.get(RQJobMetaField.TASK_PROGRESS, 0.)) / len(dataset.subsets().keys())
-        job.save_meta()
+        job_meta = ImportRQMeta.for_job(job)
+        job_meta.status = 'Task from dataset is being created...'
+        job_meta.progress = (subset_id + (job_meta.task_progress or 0.)) / len(dataset.subsets().keys())
+        job_meta.save()
 
         task_fields = {
             'project': project_annotation.db_project,

cvat/apps/dataset_manager/project.py

@@ -18,7 +18,7 @@
 from cvat.apps.engine import models
 from cvat.apps.engine.log import DatasetLogManager
 from cvat.apps.engine.model_utils import bulk_create
-from cvat.apps.engine.rq_job_handler import RQJobMetaField
+from cvat.apps.engine.rq import ImportRQMeta
 from cvat.apps.engine.serializers import DataSerializer, TaskWriteSerializer
 from cvat.apps.engine.task import _create_thread as create_task
 
@@ -198,9 +198,10 @@ def data(self) -> dict:
 @transaction.atomic
 def import_dataset_as_project(src_file, project_id, format_name, conv_mask_to_poly):
     rq_job = rq.get_current_job()
-    rq_job.meta[RQJobMetaField.STATUS] = 'Dataset import has been started...'
-    rq_job.meta[RQJobMetaField.PROGRESS] = 0.
-    rq_job.save_meta()
+    rq_job_meta = ImportRQMeta.for_job(rq_job)
+    rq_job_meta.status = 'Dataset import has been started...'
+    rq_job_meta.progress = 0.
+    rq_job_meta.save()
 
     project = ProjectAnnotationAndData(project_id)
     project.init_from_db()

cvat/apps/dataset_manager/views.py

@@ -20,7 +20,7 @@
 import cvat.apps.dataset_manager.task as task
 from cvat.apps.engine.log import ServerLogManager
 from cvat.apps.engine.models import Job, Project, Task
-from cvat.apps.engine.rq_job_handler import RQMeta
+from cvat.apps.engine.rq import ExportRQMeta
 from cvat.apps.engine.utils import get_rq_lock_by_user
 
 from .formats.registry import EXPORT_FORMATS, IMPORT_FORMATS
@@ -88,7 +88,8 @@ def _patched_retry(*_1, **_2):
             settings.CVAT_QUEUES.EXPORT_DATA.value
         )
 
-        user_id = current_rq_job.meta.get('user', {}).get('id') or -1
+        rq_job_meta = ExportRQMeta.for_job(current_rq_job)
+        user_id = rq_job_meta.user.id or -1
 
         with get_rq_lock_by_user(settings.CVAT_QUEUES.EXPORT_DATA.value, user_id):
             scheduled_rq_job: rq.job.Job = scheduler.enqueue_in(
@@ -97,7 +98,7 @@ def _patched_retry(*_1, **_2):
                 *current_rq_job.args,
                 **current_rq_job.kwargs,
                 job_id=current_rq_job.id,
-                meta=RQMeta.reset_meta_on_retry(current_rq_job.meta),
+                meta=rq_job_meta.get_meta_on_retry(),
                 job_ttl=current_rq_job.ttl,
                 job_result_ttl=current_rq_job.result_ttl,
                 job_description=current_rq_job.description,

cvat/apps/engine/background.py

@@ -37,14 +37,12 @@
     Task,
 )
 from cvat.apps.engine.permissions import get_cloud_storage_for_import_or_export
-from cvat.apps.engine.rq_job_handler import RQId, RQJobMetaField
+from cvat.apps.engine.rq import ExportRQMeta, RQId, define_dependent_job
 from cvat.apps.engine.serializers import RqIdSerializer
 from cvat.apps.engine.types import ExtendedRequest
 from cvat.apps.engine.utils import (
     build_annotations_file_name,
     build_backup_file_name,
-    define_dependent_job,
-    get_rq_job_meta,
     get_rq_lock_by_user,
     get_rq_lock_for_job,
     sendfile,
@@ -229,7 +227,7 @@ def _handle_rq_job_v1(
     ) -> Optional[Response]:
 
         def is_result_outdated() -> bool:
-            return rq_job.meta[RQJobMetaField.REQUEST]["timestamp"] < instance_update_time
+            return ExportRQMeta.for_job(rq_job).request.timestamp < instance_update_time
 
         def handle_local_download() -> Response:
             with dm.util.get_export_cache_lock(
@@ -342,7 +340,7 @@ def handle_local_download() -> Response:
                     f"Export to {self.export_args.location} location is not implemented yet"
                 )
         elif rq_job_status == RQJobStatus.FAILED:
-            exc_info = rq_job.meta.get(RQJobMetaField.FORMATTED_EXCEPTION, str(rq_job.exc_info))
+            exc_info = ExportRQMeta.for_job(rq_job).formatted_exception or str(rq_job.exc_info)
             rq_job.delete()
             return Response(exc_info, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
         elif (
@@ -478,16 +476,19 @@ def setup_background_job(
             result_url = self.make_result_url()
 
         with get_rq_lock_by_user(queue, user_id):
+            meta = ExportRQMeta.build_for(
+                request=self.request,
+                db_obj=self.db_instance,
+                result_url=result_url,
+            )
             queue.enqueue_call(
                 func=func,
                 args=func_args,
                 kwargs={
                     "server_url": server_address,
                 },
                 job_id=rq_id,
-                meta=get_rq_job_meta(
-                    request=self.request, db_obj=self.db_instance, result_url=result_url
-                ),
+                meta=meta,
                 depends_on=define_dependent_job(queue, user_id, rq_id=rq_id),
                 result_ttl=cache_ttl.total_seconds(),
                 failure_ttl=cache_ttl.total_seconds(),
@@ -548,7 +549,7 @@ def _handle_rq_job_v1(
     ) -> Optional[Response]:
 
         def is_result_outdated() -> bool:
-            return rq_job.meta[RQJobMetaField.REQUEST]["timestamp"] < last_instance_update_time
+            return ExportRQMeta.for_job(rq_job).request.timestamp < last_instance_update_time
 
         last_instance_update_time = timezone.localtime(self.db_instance.updated_date)
         timestamp = self.get_timestamp(last_instance_update_time)
@@ -644,7 +645,7 @@ def is_result_outdated() -> bool:
                     f"Export to {self.export_args.location} location is not implemented yet"
                 )
         elif rq_job_status == RQJobStatus.FAILED:
-            exc_info = rq_job.meta.get(RQJobMetaField.FORMATTED_EXCEPTION, str(rq_job.exc_info))
+            exc_info = ExportRQMeta.for_job(rq_job).formatted_exception or str(rq_job.exc_info)
             rq_job.delete()
             return Response(exc_info, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
         elif (
@@ -756,13 +757,16 @@ def setup_background_job(
         user_id = self.request.user.id
 
         with get_rq_lock_by_user(queue, user_id):
+            meta = ExportRQMeta.build_for(
+                request=self.request,
+                db_obj=self.db_instance,
+                result_url=result_url,
+            )
             queue.enqueue_call(
                 func=func,
                 args=func_args,
                 job_id=rq_id,
-                meta=get_rq_job_meta(
-                    request=self.request, db_obj=self.db_instance, result_url=result_url
-                ),
+                meta=meta,
                 depends_on=define_dependent_job(queue, user_id, rq_id=rq_id),
                 result_ttl=cache_ttl.total_seconds(),
                 failure_ttl=cache_ttl.total_seconds(),

cvat/apps/engine/backup.py

@@ -67,7 +67,7 @@
     StorageMethodChoice,
 )
 from cvat.apps.engine.permissions import get_cloud_storage_for_import_or_export
-from cvat.apps.engine.rq_job_handler import RQId, RQJobMetaField
+from cvat.apps.engine.rq import ImportRQMeta, RQId, define_dependent_job
 from cvat.apps.engine.serializers import (
     AnnotationGuideWriteSerializer,
     AssetWriteSerializer,
@@ -89,8 +89,6 @@
 from cvat.apps.engine.types import ExtendedRequest
 from cvat.apps.engine.utils import (
     av_scan_paths,
-    define_dependent_job,
-    get_rq_job_meta,
     get_rq_lock_by_user,
     import_resource_with_clean_up_after,
     process_failed_job,
@@ -1180,6 +1178,7 @@ def create_backup(
         log_exception(logger)
         raise
 
+
 def _import(
     importer: TaskImporter | ProjectImporter,
     request: ExtendedRequest,
@@ -1192,9 +1191,6 @@ def _import(
 ):
     rq_job = queue.fetch_job(rq_id)
 
-    if (user_id_from_meta := getattr(rq_job, 'meta', {}).get(RQJobMetaField.USER, {}).get('id')) and user_id_from_meta != request.user.id:
-        return Response(status=status.HTTP_403_FORBIDDEN)
-
     if not rq_job:
         org_id = getattr(request.iam_context['organization'], 'id', None)
         location = location_conf.get('location')
@@ -1239,19 +1235,25 @@ def _import(
         user_id = request.user.id
 
         with get_rq_lock_by_user(queue, user_id):
+            meta = ImportRQMeta.build_for(
+                request=request,
+                db_obj=None,
+                tmp_file=filename,
+            )
             rq_job = queue.enqueue_call(
                 func=func,
                 args=func_args,
                 job_id=rq_id,
-                meta={
-                    'tmp_file': filename,
-                    **get_rq_job_meta(request=request, db_obj=None)
-                },
+                meta=meta,
                 depends_on=define_dependent_job(queue, user_id),
                 result_ttl=settings.IMPORT_CACHE_SUCCESS_TTL.total_seconds(),
                 failure_ttl=settings.IMPORT_CACHE_FAILED_TTL.total_seconds()
             )
     else:
+        rq_job_meta = ImportRQMeta.for_job(rq_job)
+        if rq_job_meta.user.id != request.user.id:
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
         if rq_job.is_finished:
             project_id = rq_job.return_value()
             rq_job.delete()

cvat/apps/engine/cache.py

@@ -53,7 +53,7 @@
     ZipCompressedChunkWriter,
     load_image,
 )
-from cvat.apps.engine.rq_job_handler import RQJobMetaField
+from cvat.apps.engine.rq import RQMetaWithFailureInfo
 from cvat.apps.engine.utils import (
     CvatChunkTimestampMismatchError,
     format_list,
@@ -107,9 +107,10 @@ def wait_for_rq_job(rq_job: rq.job.Job):
         if job_status in ("finished",):
             return
         elif job_status in ("failed",):
-            job_meta = rq_job.get_meta()
-            exc_type = job_meta.get(RQJobMetaField.EXCEPTION_TYPE, Exception)
-            exc_args = job_meta.get(RQJobMetaField.EXCEPTION_ARGS, ("Cannot create chunk",))
+            rq_job.get_meta()  # refresh from Redis
+            job_meta = RQMetaWithFailureInfo.for_job(rq_job)
+            exc_type = job_meta.exc_type or Exception
+            exc_args = job_meta.exc_args or ("Cannot create chunk",)
             raise exc_type(*exc_args)
 
         time.sleep(settings.CVAT_CHUNK_CREATE_CHECK_INTERVAL)

cvat/apps/engine/mixins.py

@@ -41,7 +41,7 @@
     RequestTarget,
     Task,
 )
-from cvat.apps.engine.rq_job_handler import RQId
+from cvat.apps.engine.rq import RQId
 from cvat.apps.engine.serializers import DataSerializer, RqIdSerializer
 from cvat.apps.engine.types import ExtendedRequest
 from cvat.apps.engine.utils import is_dataset_export

cvat/apps/engine/permissions.py

@@ -14,7 +14,7 @@
 from rest_framework.exceptions import PermissionDenied, ValidationError
 from rq.job import Job as RQJob
 
-from cvat.apps.engine.rq_job_handler import is_rq_job_owner
+from cvat.apps.engine.rq import is_rq_job_owner
 from cvat.apps.engine.types import ExtendedRequest
 from cvat.apps.engine.utils import is_dataset_export
 from cvat.apps.iam.permissions import (