Allow quality reports to regular users

changelog.d/20241004_201336_mzhiltso_allow_quality_reports_for_regular_users.md

@@ -0,0 +1,4 @@
+### Changed
+
+- \[Server API\] Quality report computation is now allowed to regular users
+  (<https://github.com/cvat-ai/cvat/pull/8511>)

cvat/apps/iam/rules/utils.rego

@@ -58,6 +58,7 @@ IMPORT_DATASET := "import:dataset"
 IMPORT_BACKUP := "import:backup"
 EXPORT_BACKUP := "export:backup"
 UPDATE_ORG := "update:organization"
+VIEW_STATUS := "view:status"
 
 
 get_priority(privilege) := {

cvat/apps/quality_control/permissions.py

@@ -18,6 +18,7 @@
 class QualityReportPermission(OpenPolicyAgentPermission):
     obj: Optional[QualityReport]
     job_owner_id: Optional[int]
+    task_id: Optional[int]
 
     class Scopes(StrEnum):
         LIST = "list"
@@ -29,7 +30,7 @@
     def create_scope_check_status(cls, request, job_owner_id: int, iam_context=None):
         if not iam_context and request:
             iam_context = get_iam_context(request, None)
-        return cls(**iam_context, scope="view:status", job_owner_id=job_owner_id)
+        return cls(**iam_context, scope=cls.Scopes.VIEW_STATUS, job_owner_id=job_owner_id)
 
     @classmethod
     def create_scope_view(cls, request, report: Union[int, QualityReport], iam_context=None):
@@ -59,11 +60,36 @@
                 elif scope == Scopes.LIST and isinstance(obj, Task):
                     permissions.append(TaskPermission.create_scope_view(request, task=obj))
                 elif scope == Scopes.CREATE:
+                    if request.query_params.get("rq_id"):
+                        # There will be another check for this case during request processing
+                        continue
+
                     task_id = request.data.get("task_id")
                     if task_id is not None:
-                        permissions.append(TaskPermission.create_scope_view(request, task_id))
+                        # The request may have a different org or org unset
+                        # Here we need to retrieve iam_context for this user, based on the task_id
+                        try:
+                            task = Task.objects.get(id=task_id)
+                        except Task.DoesNotExist as ex:
+                            raise ValidationError(str(ex))
+
+                        iam_context = get_iam_context(request, task)
+
+                        permissions.append(
+                            TaskPermission.create_scope_view(request, task, iam_context=iam_context)
+                        )
+
+                    permissions.append(
+                        cls.create_base_perm(
+                            request,
+                            view,
+                            scope,
+                            iam_context,
+                            obj,
+                            task_id=task_id,
+                        )
+                    )
 
-                    permissions.append(cls.create_base_perm(request, view, scope, iam_context, obj))
                 else:
                     permissions.append(cls.create_base_perm(request, view, scope, iam_context, obj))
 
@@ -91,15 +117,26 @@
     def get_resource(self):
         data = None
 
-        if self.obj:
-            task = self.obj.get_task()
-            if task.project:
+        if self.obj or self.scope == self.Scopes.CREATE:
+            task = None
+            if self.obj:
+                obj_id = self.obj.id
+                task = self.obj.get_task()
+            elif self.scope == self.Scopes.CREATE:
+                obj_id = None
+
+                if self.task_id:
+                    task = Task.objects.get(id=self.task_id)
+            else:
+                raise AssertionError(self.scope)
+
+            if task and task.project:
                 organization = task.project.organization
             else:
-                organization = task.organization
+                organization = getattr(task, "organization", None)
 
             data = {
-                "id": self.obj.id,
+                "id": obj_id,
                 "organization": {"id": getattr(organization, "id", None)},
                 "task": (
                     {
@@ -114,12 +151,12 @@
                         "owner": {"id": getattr(task.project.owner, "id", None)},
                         "assignee": {"id": getattr(task.project.assignee, "id", None)},
                     }
-                    if task.project
+                    if getattr(task, "project", None)
                     else None
                 ),
             }
         elif self.scope == self.Scopes.VIEW_STATUS:
-            data = {"owner": self.job_owner_id}
+            data = {"owner": {"id": self.job_owner_id}}
 
         return data
 

cvat/apps/quality_control/quality_reports.py

@@ -10,18 +10,22 @@
 from copy import deepcopy
 from datetime import timedelta
 from functools import cached_property, partial
+from time import sleep
 from typing import Any, Callable, Dict, Hashable, List, Optional, Sequence, Tuple, Union, cast
-from uuid import uuid4
 
 import datumaro as dm
 import datumaro.util.mask_tools
 import django_rq
 import numpy as np
+import rq
 from attrs import asdict, define, fields_dict
 from datumaro.util import dump_json, parse_json
 from django.conf import settings
 from django.db import transaction
 from django.utils import timezone
+from django_rq.queues import DjangoRQ as RqQueue
+from rq.job import Job as RqJob
+from rq_scheduler import Scheduler as RqScheduler
 from scipy.optimize import linear_sum_assignment
 
 from cvat.apps.dataset_manager.bindings import (
@@ -48,6 +52,7 @@
     User,
     ValidationMode,
 )
+from cvat.apps.engine.utils import define_dependent_job, get_rq_job_meta, get_rq_lock_by_user
 from cvat.apps.profiler import silk_profile
 from cvat.apps.quality_control import models
 from cvat.apps.quality_control.models import (
@@ -2115,25 +2120,26 @@ def generate_report(self) -> ComparisonReport:
 
 
 class QualityReportUpdateManager:
-    _QUEUE_JOB_PREFIX = "update-quality-metrics-task-"
+    _QUEUE_AUTOUPDATE_JOB_PREFIX = "update-quality-metrics-"
+    _QUEUE_CUSTOM_JOB_PREFIX = "quality-check-"
     _RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE = "custom_quality_check"
     _JOB_RESULT_TTL = 120
 
     @classmethod
     def _get_quality_check_job_delay(cls) -> timedelta:
         return timedelta(seconds=settings.QUALITY_CHECK_JOB_DELAY)
 
-    def _get_scheduler(self):
+    def _get_scheduler(self) -> RqScheduler:
         return django_rq.get_scheduler(settings.CVAT_QUEUES.QUALITY_REPORTS.value)
 
-    def _get_queue(self):
+    def _get_queue(self) -> RqQueue:
         return django_rq.get_queue(settings.CVAT_QUEUES.QUALITY_REPORTS.value)
 
     def _make_queue_job_id_base(self, task: Task) -> str:
-        return f"{self._QUEUE_JOB_PREFIX}{task.id}"
+        return f"{self._QUEUE_AUTOUPDATE_JOB_PREFIX}task-{task.id}"
 
-    def _make_custom_quality_check_job_id(self) -> str:
-        return uuid4().hex
+    def _make_custom_quality_check_job_id(self, task_id: int, user_id: int) -> str:
+        return f"{self._QUEUE_CUSTOM_JOB_PREFIX}task-{task_id}-user-{user_id}"
 
     @classmethod
     def _get_last_report_time(cls, task: Task) -> Optional[timezone.datetime]:
@@ -2186,28 +2192,47 @@ def schedule_quality_autoupdate_job(self, task: Task):
             task_id=task.id,
         )
 
-    def schedule_quality_check_job(self, task: Task, *, user_id: int) -> str:
+    class JobAlreadyExists(QualityReportsNotAvailable):
+        def __str__(self):
+            return "Quality computation job for this task already enqueued"
+
+    def schedule_custom_quality_check_job(self, request, task: Task, *, user_id: int) -> str:
         """
         Schedules a quality report computation job, supposed for updates by a request.
         """
 
         self._check_quality_reporting_available(task)
 
-        rq_id = self._make_custom_quality_check_job_id()
-
         queue = self._get_queue()
-        queue.enqueue(
-            self._check_task_quality,
-            task_id=task.id,
-            job_id=rq_id,
-            meta={"user_id": user_id, "job_type": self._RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE},
-            result_ttl=self._JOB_RESULT_TTL,
-            failure_ttl=self._JOB_RESULT_TTL,
-        )
+
+        with get_rq_lock_by_user(queue, user_id=user_id):
+            rq_id = self._make_custom_quality_check_job_id(task_id=task.id, user_id=user_id)
+            rq_job = queue.fetch_job(rq_id)
+            if rq_job and rq_job.get_status(refresh=False) in (
+                rq.job.JobStatus.QUEUED,
+                rq.job.JobStatus.STARTED,
+                rq.job.JobStatus.SCHEDULED,
+            ):
+                raise self.JobAlreadyExists()
+
+            dependency = define_dependent_job(
+                queue, user_id=user_id, rq_id=rq_id, should_be_dependent=True
+            )
+
+            queue.enqueue(
+                self._check_task_quality,
+                task_id=task.id,
+                job_id=rq_id,
+                meta=get_rq_job_meta(request=request, db_obj=task),
+                result_ttl=self._JOB_RESULT_TTL,
+                failure_ttl=self._JOB_RESULT_TTL,
+                depends_on=dependency,
+            )
+            print(queue.fetch_job(rq_id).get_status())
 
         return rq_id
 
-    def get_quality_check_job(self, rq_id: str):
+    def get_quality_check_job(self, rq_id: str) -> Optional[RqJob]:
         queue = self._get_queue()
         rq_job = queue.fetch_job(rq_id)
 
@@ -2216,8 +2241,8 @@ def get_quality_check_job(self, rq_id: str):
 
         return rq_job
 
-    def is_custom_quality_check_job(self, rq_job) -> bool:
-        return rq_job.meta.get("job_type") == self._RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE
+    def is_custom_quality_check_job(self, rq_job: RqJob) -> bool:
+        return isinstance(rq_job.id, str) and rq_job.id.startswith(self._QUEUE_CUSTOM_JOB_PREFIX)
 
     @classmethod
     @silk_profile()

cvat/apps/quality_control/rules/conflicts.rego

@@ -4,6 +4,7 @@ import rego.v1
 
 import data.utils
 import data.organizations
+import data.tasks
 
 # input: {
 #     "scope": <"list"> or null,
@@ -41,6 +42,42 @@ import data.organizations
 #     }
 # }
 
+is_task_owner if {
+    input.resource.task.owner.id == input.auth.user.id
+}
+
+is_task_assignee if {
+    input.resource.task.assignee.id == input.auth.user.id
+}
+
+is_project_owner if {
+    input.resource.project.owner.id == input.auth.user.id
+}
+
+is_project_assignee if {
+    input.resource.project.assignee.id == input.auth.user.id
+}
+
+is_project_staff if {
+    is_project_owner
+}
+
+is_project_staff if {
+    is_project_assignee
+}
+
+is_task_staff if {
+    is_project_staff
+}
+
+is_task_staff if {
+    is_task_owner
+}
+
+is_task_staff if {
+    is_task_assignee
+}
+
 default allow := false
 
 allow if {
@@ -57,6 +94,28 @@ allow if {
     organizations.is_member
 }
 
+allow if {
+    input.scope == utils.VIEW
+    utils.is_sandbox
+    is_task_staff
+    utils.has_perm(utils.WORKER)
+}
+
+allow if {
+    input.scope == utils.VIEW
+    input.auth.organization.id == input.resource.organization.id
+    utils.has_perm(utils.USER)
+    organizations.has_perm(organizations.MAINTAINER)
+}
+
+allow if {
+    input.scope == utils.VIEW
+    is_task_staff
+    input.auth.organization.id == input.resource.organization.id
+    utils.has_perm(utils.WORKER)
+    organizations.has_perm(organizations.WORKER)
+}
+
 filter := [] if { # Django Q object to filter list of entries
     utils.is_admin
     utils.is_sandbox