From 51a1554222ffc99e861568d9d8f1db970f73e0f4 Mon Sep 17 00:00:00 2001
From: Maria Khrustaleva <maria@cvat.ai>
Date: Mon, 10 Feb 2025 17:56:41 +0100
Subject: [PATCH] Apply comments

---
 cvat/apps/engine/background.py               |  6 +-
 cvat/apps/engine/backup.py                   | 12 +--
 cvat/apps/engine/filters.py                  | 13 ++--
 cvat/apps/engine/mixins.py                   | 28 +++----
 cvat/apps/engine/pagination.py               |  4 +-
 cvat/apps/engine/permissions.py              | 62 ++++++++--------
 cvat/apps/engine/task.py                     |  4 +-
 cvat/apps/engine/types.py                    |  8 +-
 cvat/apps/engine/utils.py                    | 12 +--
 cvat/apps/engine/view_utils.py               |  4 +-
 cvat/apps/engine/views.py                    | 78 ++++++++++----------
 cvat/apps/lambda_manager/views.py            |  2 +-
 cvat/apps/quality_control/quality_reports.py |  2 +-
 13 files changed, 120 insertions(+), 115 deletions(-)

diff --git a/cvat/apps/engine/background.py b/cvat/apps/engine/background.py
index 03a9d74a715f..8fc944babb01 100644
--- a/cvat/apps/engine/background.py
+++ b/cvat/apps/engine/background.py
@@ -39,7 +39,7 @@
 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.serializers import RqIdSerializer
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import (
     build_annotations_file_name,
     build_backup_file_name,
@@ -180,7 +180,7 @@ def location(self) -> Location:
     def __init__(
         self,
         db_instance: Union[models.Project, models.Task, models.Job],
-        request: PatchedRequest,
+        request: Request,
         export_callback: Callable,
         save_images: Optional[bool] = None,
         *,
@@ -525,7 +525,7 @@ def location(self) -> Location:
     def __init__(
         self,
         db_instance: Union[models.Project, models.Task],
-        request: PatchedRequest,
+        request: Request,
         *,
         version: int = 2,
     ) -> None:
diff --git a/cvat/apps/engine/backup.py b/cvat/apps/engine/backup.py
index ca2ba36fa547..c8df7095f4d1 100644
--- a/cvat/apps/engine/backup.py
+++ b/cvat/apps/engine/backup.py
@@ -79,7 +79,7 @@
     ValidationParamsSerializer,
 )
 from cvat.apps.engine.task import JobFileMapping, _create_thread
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import (
     av_scan_paths,
     define_dependent_job,
@@ -1150,10 +1150,10 @@ def create_backup(
 
 def _import(
     importer: TaskImporter | ProjectImporter,
-    request: PatchedRequest,
-    queue: django_rq.queues.Queue,
+    request: Request,
+    queue: django_rq.queues.DjangoRQ,
     rq_id: str,
-    Serializer: TaskFileSerializer | ProjectFileSerializer,
+    Serializer: type[TaskFileSerializer] | type[ProjectFileSerializer],
     file_field_name: str,
     location_conf: dict,
     filename: str | None = None,
@@ -1245,7 +1245,7 @@ def _import(
 def get_backup_dirname():
     return TmpDirManager.TMP_ROOT
 
-def import_project(request: PatchedRequest, queue_name: str, filename: str | None = None):
+def import_project(request: Request, queue_name: str, filename: str | None = None):
     if 'rq_id' in request.data:
         rq_id = request.data['rq_id']
     else:
@@ -1274,7 +1274,7 @@ def import_project(request: PatchedRequest, queue_name: str, filename: str | Non
         filename=filename
     )
 
-def import_task(request: PatchedRequest, queue_name: str, filename: str | None = None):
+def import_task(request: Request, queue_name: str, filename: str | None = None):
     rq_id = request.data.get('rq_id', RQId(
         RequestAction.IMPORT, RequestTarget.TASK, uuid.uuid4(),
         subresource=RequestSubresource.BACKUP,
diff --git a/cvat/apps/engine/filters.py b/cvat/apps/engine/filters.py
index b1045476540a..df924db77bf9 100644
--- a/cvat/apps/engine/filters.py
+++ b/cvat/apps/engine/filters.py
@@ -21,7 +21,8 @@
 from rest_framework import filters
 from rest_framework.compat import coreapi, coreschema
 from rest_framework.exceptions import ValidationError
-from rest_framework.request import Request
+
+from cvat.apps.engine.types import Request
 
 DEFAULT_FILTER_FIELDS_ATTR = 'filter_fields'
 DEFAULT_LOOKUP_MAP_ATTR = 'lookup_fields'
@@ -39,7 +40,7 @@ def get_lookup_fields(view, fields: Optional[Iterator[str]] = None) -> dict[str,
 
 
 class SearchFilter(filters.SearchFilter):
-    def get_search_fields(self, view, request):
+    def get_search_fields(self, view, request: Request):
         search_fields = getattr(view, 'search_fields') or []
         return get_lookup_fields(view, search_fields).values()
 
@@ -82,7 +83,7 @@ class OrderingFilter(filters.OrderingFilter):
     ordering_param = 'sort'
     reverse_flag = "-"
 
-    def get_ordering(self, request, queryset, view):
+    def get_ordering(self, request: Request, queryset, view):
         ordering = []
         lookup_fields = self._get_lookup_fields(request, queryset, view)
         for term in super().get_ordering(request, queryset, view):
@@ -94,7 +95,7 @@ def get_ordering(self, request, queryset, view):
 
         return ordering
 
-    def _get_lookup_fields(self, request, queryset, view):
+    def _get_lookup_fields(self, request: Request, queryset: QuerySet, view):
         ordering_fields = self.get_valid_fields(queryset, view, {'request': request})
         ordering_fields = [v[0] for v in ordering_fields]
         return get_lookup_fields(view, ordering_fields)
@@ -201,7 +202,7 @@ def apply_filter(self,
 
         return queryset.filter(q_object)
 
-    def filter_queryset(self, request, queryset, view):
+    def filter_queryset(self, request: Request, queryset: QuerySet, view):
         json_rules = request.query_params.get(self.filter_param)
         if json_rules:
             parsed_rules = self._parse_query(json_rules)
@@ -455,7 +456,7 @@ class NonModelOrderingFilter(OrderingFilter, _NestedAttributeHandler):
     ?sort=-field1,-field2
     """
 
-    def get_ordering(self, request, queryset, view) -> tuple[list[str], bool]:
+    def get_ordering(self, request: Request, queryset: Iterable, view) -> tuple[list[str], bool]:
         ordering = super().get_ordering(request, queryset, view)
         result, reverse = [], False
         for field in ordering:
diff --git a/cvat/apps/engine/mixins.py b/cvat/apps/engine/mixins.py
index ec2a7b0336be..30305f80dd16 100644
--- a/cvat/apps/engine/mixins.py
+++ b/cvat/apps/engine/mixins.py
@@ -43,7 +43,7 @@
 )
 from cvat.apps.engine.rq_job_handler import RQId
 from cvat.apps.engine.serializers import DataSerializer, RqIdSerializer
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import is_dataset_export
 
 slogger = ServerLogManager(__name__)
@@ -167,7 +167,7 @@ def create_file(metadata, file_size, upload_dir):
         return tus_file
 
 class TusChunk:
-    def __init__(self, request: PatchedRequest):
+    def __init__(self, request: Request):
         self.META = request.META
         self.offset = int(request.META.get("HTTP_UPLOAD_OFFSET", 0))
         self.size = int(request.META.get("CONTENT_LENGTH", settings.TUS_DEFAULT_CHUNK_SIZE))
@@ -223,7 +223,7 @@ def _tus_response(self, status, data=None, extra_headers=None):
                 response.__setitem__(key, value)
         return response
 
-    def _get_metadata(self, request: PatchedRequest):
+    def _get_metadata(self, request: Request):
         metadata = {}
         if request.META.get("HTTP_UPLOAD_METADATA"):
             for kv in request.META.get("HTTP_UPLOAD_METADATA").split(","):
@@ -238,7 +238,7 @@ def _get_metadata(self, request: PatchedRequest):
                     metadata[splited_metadata[0]] = ""
         return metadata
 
-    def upload_data(self, request: PatchedRequest):
+    def upload_data(self, request: Request):
         tus_request = request.headers.get('Upload-Length', None) is not None or request.method == 'OPTIONS'
         bulk_file_upload = request.headers.get('Upload-Multiple', None) is not None
         start_upload = request.headers.get('Upload-Start', None) is not None
@@ -255,7 +255,7 @@ def upload_data(self, request: PatchedRequest):
         else: # backward compatibility case - no upload headers were found
             return self.upload_finished(request)
 
-    def init_tus_upload(self, request: PatchedRequest):
+    def init_tus_upload(self, request: Request):
         if request.method == 'OPTIONS':
             return self._tus_response(status=status.HTTP_204_NO_CONTENT)
         else:
@@ -329,7 +329,7 @@ def init_tus_upload(self, request: PatchedRequest):
                 extra_headers={'Location': urljoin(location, tus_file.file_id),
                                'Upload-Filename': tus_file.filename})
 
-    def append_tus_chunk(self, request: PatchedRequest, file_id: str):
+    def append_tus_chunk(self, request: Request, file_id: str):
         tus_file = TusFile(str(file_id), self.get_upload_dir())
         if request.method == 'HEAD':
             if tus_file.exists():
@@ -413,7 +413,7 @@ class PartialUpdateModelMixin:
     Almost the same as UpdateModelMixin, but has no public PUT / update() method.
     """
 
-    def _update(self, request: PatchedRequest, *args, **kwargs):
+    def _update(self, request: Request, *args, **kwargs):
         # This method must not be named "update" not to be matched with the PUT method
         return mixins.UpdateModelMixin.update(self, request, *args, **kwargs)
 
@@ -428,7 +428,7 @@ def partial_update(self, request, *args, **kwargs):
 class DatasetMixin:
     def export_dataset_v1(
         self,
-        request: PatchedRequest,
+        request: Request,
         save_images: bool,
         *,
         get_data: Optional[Callable[[int], dict[str, Any]]] = None,
@@ -474,7 +474,7 @@ def export_dataset_v1(
         },
     )
     @action(detail=True, methods=['POST'], serializer_class=None, url_path='dataset/export')
-    def export_dataset_v2(self, request: PatchedRequest, pk: int):
+    def export_dataset_v2(self, request: Request, pk: int):
         self._object = self.get_object() # force call of check_object_permissions()
 
         save_images = is_dataset_export(request)
@@ -486,11 +486,11 @@ def export_dataset_v2(self, request: PatchedRequest, pk: int):
     # FUTURE-TODO: migrate to new API
     def import_annotations(
         self,
-        request: PatchedRequest,
+        request: Request,
         db_obj: Project | Task | Job,
         import_func: Callable[..., None],
         rq_func: Callable[..., None],
-        rq_id_factory: RQId,
+        rq_id_factory: Callable[..., RQId],
     ):
         is_tus_request = request.headers.get('Upload-Length', None) is not None or \
             request.method == 'OPTIONS'
@@ -523,7 +523,7 @@ def import_annotations(
 
 
 class BackupMixin:
-    def export_backup_v1(self, request: PatchedRequest) -> Response:
+    def export_backup_v1(self, request: Request) -> Response:
         db_object = self.get_object() # force to call check_object_permissions
 
         export_backup_manager = BackupExportManager(db_object, request, version=1)
@@ -535,7 +535,7 @@ def export_backup_v1(self, request: PatchedRequest) -> Response:
         return response
 
     # FUTURE-TODO: migrate to new API
-    def import_backup_v1(self, request: PatchedRequest, import_func: Callable) -> Response:
+    def import_backup_v1(self, request: Request, import_func: Callable) -> Response:
         location = request.query_params.get("location", Location.LOCAL)
         if location == Location.CLOUD_STORAGE:
             file_name = request.query_params.get("filename", "")
@@ -569,7 +569,7 @@ def import_backup_v1(self, request: PatchedRequest, import_func: Callable) -> Re
         },
     )
     @action(detail=True, methods=['POST'], serializer_class=None, url_path='backup/export')
-    def export_backup_v2(self, request: PatchedRequest, pk: int):
+    def export_backup_v2(self, request: Request, pk: int):
         db_object = self.get_object() # force to call check_object_permissions
 
         export_backup_manager = BackupExportManager(db_object, request, version=2)
diff --git a/cvat/apps/engine/pagination.py b/cvat/apps/engine/pagination.py
index 90d6c9dd9248..bda58b0ac7e4 100644
--- a/cvat/apps/engine/pagination.py
+++ b/cvat/apps/engine/pagination.py
@@ -6,13 +6,13 @@
 
 from rest_framework.pagination import PageNumberPagination
 
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 
 
 class CustomPagination(PageNumberPagination):
     page_size_query_param = "page_size"
 
-    def get_page_size(self, request: PatchedRequest):
+    def get_page_size(self, request: Request):
         page_size = 0
         try:
             value = request.query_params[self.page_size_query_param]
diff --git a/cvat/apps/engine/permissions.py b/cvat/apps/engine/permissions.py
index 3ad54e774907..facd37bea00b 100644
--- a/cvat/apps/engine/permissions.py
+++ b/cvat/apps/engine/permissions.py
@@ -14,7 +14,7 @@
 from rq.job import Job as RQJob
 
 from cvat.apps.engine.rq_job_handler import is_rq_job_owner
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import is_dataset_export
 from cvat.apps.iam.permissions import (
     OpenPolicyAgentPermission,
@@ -50,7 +50,7 @@ class Scopes(StrEnum):
         LIST_CONTENT = 'list:content'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'server':
             for scope in cls.get_scopes(request, view, obj):
@@ -64,7 +64,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/server/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: None):
+    def get_scopes(request: Request, view: ViewSet, obj: None):
         Scopes = __class__.Scopes
         return [{
             ('annotation_formats', 'GET'): Scopes.VIEW,
@@ -86,7 +86,7 @@ class Scopes(StrEnum):
         DELETE = 'delete'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: User | None, iam_context: dict[str, Any]):
+    def create(cls, request: Request, view: ViewSet, obj: User | None, iam_context: dict[str, Any]):
         permissions = []
         if view.basename == 'user':
             for scope in cls.get_scopes(request, view, obj):
@@ -100,7 +100,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/users/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: User | None):
+    def get_scopes(request: Request, view: ViewSet, obj: User | None):
         Scopes = __class__.Scopes
         return [{
             'list': Scopes.LIST,
@@ -149,7 +149,7 @@ class Scopes(StrEnum):
         DELETE = 'delete'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: CloudStorage | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: CloudStorage | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'cloudstorage':
             for scope in cls.get_scopes(request, view, obj):
@@ -159,7 +159,7 @@ def create(cls, request: PatchedRequest, view: ViewSet, obj: CloudStorage | None
         return permissions
 
     @classmethod
-    def create_scope_view(cls, iam_context: dict[str, Any], storage_id: int, request: PatchedRequest | None = None):
+    def create_scope_view(cls, iam_context: dict[str, Any], storage_id: int, request: Request | None = None):
         try:
             obj = CloudStorage.objects.get(id=storage_id)
         except CloudStorage.DoesNotExist as ex:
@@ -175,7 +175,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/cloudstorages/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: CloudStorage | None):
+    def get_scopes(request: Request, view: ViewSet, obj: CloudStorage | None):
         Scopes = __class__.Scopes
         return [{
             'list': Scopes.LIST,
@@ -231,7 +231,7 @@ class Scopes(StrEnum):
         IMPORT_BACKUP = 'import:backup'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Project | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Project | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'project':
             assignee_id = request.data.get('assignee_id') or request.data.get('assignee')
@@ -273,7 +273,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/projects/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Project | None):
+    def get_scopes(request: Request, view: ViewSet, obj: Project | None):
         Scopes = __class__.Scopes
         scope = {
             ('list', 'GET'): Scopes.LIST,
@@ -314,7 +314,7 @@ def get_scopes(request: PatchedRequest, view: ViewSet, obj: Project | None):
         return scopes
 
     @classmethod
-    def create_scope_view(cls, request: PatchedRequest, project: int | Project, iam_context: dict[str, Any] | None = None):
+    def create_scope_view(cls, request: Request, project: int | Project, iam_context: dict[str, Any] | None = None):
         if isinstance(project, int):
             try:
                 project = Project.objects.get(id=project)
@@ -327,7 +327,7 @@ def create_scope_view(cls, request: PatchedRequest, project: int | Project, iam_
         return cls(**iam_context, obj=project, scope=__class__.Scopes.VIEW)
 
     @classmethod
-    def create_scope_create(cls, request: PatchedRequest, org_id: int | None):
+    def create_scope_create(cls, request: Request, org_id: int | None):
         organization = None
         membership = None
         privilege = request.iam_context['privilege']
@@ -403,7 +403,7 @@ class Scopes(StrEnum):
         UPDATE_VALIDATION_LAYOUT = 'update:validation_layout'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Task | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Task | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'task':
             project_id = request.data.get('project_id') or request.data.get('project')
@@ -461,7 +461,7 @@ def create(cls, request: PatchedRequest, view: ViewSet, obj: Task | None, iam_co
         return permissions
 
     @classmethod
-    def create_scope_view(cls, request: PatchedRequest, task: int | Task, iam_context: dict[str, Any] | None = None):
+    def create_scope_view(cls, request: Request, task: int | Task, iam_context: dict[str, Any] | None = None):
         if isinstance(task, int):
             try:
                 task = Task.objects.get(id=task)
@@ -478,7 +478,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/tasks/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Task | None) -> list[Scopes]:
+    def get_scopes(request: Request, view: ViewSet, obj: Task | None) -> list[Scopes]:
         Scopes = __class__.Scopes
         scope = {
             ('list', 'GET'): Scopes.LIST,
@@ -633,7 +633,7 @@ class Scopes(StrEnum):
         UPDATE_VALIDATION_LAYOUT = 'update:validation_layout'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Job | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Job | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'job':
             task_id = request.data.get('task_id')
@@ -686,7 +686,7 @@ def create_scope_view_data(cls, iam_context: dict[str, Any], job_id: int):
         return cls(**iam_context, obj=obj, scope='view:data')
 
     @classmethod
-    def create_scope_view(cls, request: PatchedRequest, job: int | Job, iam_context: dict[str, Any] | None = None):
+    def create_scope_view(cls, request: Request, job: int | Job, iam_context: dict[str, Any] | None = None):
         if isinstance(job, int):
             try:
                 job = Job.objects.get(id=job)
@@ -704,7 +704,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/jobs/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Job | None):
+    def get_scopes(request: Request, view: ViewSet, obj: Job | None):
         Scopes = __class__.Scopes
         scope = {
             ('list', 'GET'): Scopes.LIST,
@@ -809,7 +809,7 @@ class Scopes(StrEnum):
         VIEW = 'view'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Comment | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Comment | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'comment':
             for scope in cls.get_scopes(request, view, obj):
@@ -824,7 +824,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/comments/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Comment | None):
+    def get_scopes(request: Request, view: ViewSet, obj: Comment | None):
         Scopes = __class__.Scopes
         return [{
             'list': Scopes.LIST,
@@ -894,7 +894,7 @@ class Scopes(StrEnum):
         VIEW = 'view'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Issue | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Issue | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'issue':
             assignee_id = request.data.get('assignee')
@@ -915,7 +915,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/issues/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Issue | None):
+    def get_scopes(request: Request, view: ViewSet, obj: Issue | None):
         Scopes = __class__.Scopes
         return [{
             'list': Scopes.LIST,
@@ -986,7 +986,7 @@ class Scopes(StrEnum):
         VIEW = 'view'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: Label | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: Label | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         Scopes = __class__.Scopes
 
         permissions = []
@@ -1041,7 +1041,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/labels/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: Label | None):
+    def get_scopes(request: Request, view: ViewSet, obj: Label | None):
         Scopes = __class__.Scopes
         return [{
             'list': Scopes.LIST,
@@ -1084,7 +1084,7 @@ class Scopes(StrEnum):
         CREATE  = 'create'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: AnnotationGuide | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: AnnotationGuide | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         permissions = []
 
         if view.basename == 'annotationguide':
@@ -1103,7 +1103,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/annotationguides/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: AnnotationGuide | None):
+    def get_scopes(request: Request, view: ViewSet, obj: AnnotationGuide | None):
         Scopes = __class__.Scopes
         return [{
             'create': Scopes.CREATE,
@@ -1155,7 +1155,7 @@ class Scopes(StrEnum):
         CREATE  = 'create'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: AnnotationGuide | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: AnnotationGuide | None, iam_context: dict[str, Any]) -> list[OpenPolicyAgentPermission]:
         Scopes = __class__.Scopes
         permissions = []
 
@@ -1182,7 +1182,7 @@ def create(cls, request: PatchedRequest, view: ViewSet, obj: AnnotationGuide | N
         return permissions
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: AnnotationGuide | None):
+    def get_scopes(request: Request, view: ViewSet, obj: AnnotationGuide | None):
         Scopes = __class__.Scopes
         return [{
             'create': Scopes.CREATE,
@@ -1198,7 +1198,7 @@ class Scopes(StrEnum):
         CANCEL = 'cancel'
 
     @classmethod
-    def create(cls, request: PatchedRequest, view: ViewSet, obj: RQJob | None, iam_context: dict) -> list[OpenPolicyAgentPermission]:
+    def create(cls, request: Request, view: ViewSet, obj: RQJob | None, iam_context: dict) -> list[OpenPolicyAgentPermission]:
         permissions = []
         if view.basename == 'request':
             for scope in cls.get_scopes(request, view, obj):
@@ -1214,7 +1214,7 @@ def __init__(self, **kwargs):
         self.url = settings.IAM_OPA_DATA_URL + '/requests/allow'
 
     @staticmethod
-    def get_scopes(request: PatchedRequest, view: ViewSet, obj: RQJob | None) -> list[Scopes]:
+    def get_scopes(request: Request, view: ViewSet, obj: RQJob | None) -> list[Scopes]:
         Scopes = __class__.Scopes
         return [{
             ('list', 'GET'): Scopes.LIST,
@@ -1227,7 +1227,7 @@ def get_resource(self):
         return None
 
 def get_cloud_storage_for_import_or_export(
-    storage_id: int, *, request: PatchedRequest, is_default: bool = False
+    storage_id: int, *, request: Request, is_default: bool = False
 ) -> CloudStorage:
     perm = CloudStoragePermission.create_scope_view(None, storage_id=storage_id, request=request)
     result = perm.check_access()
diff --git a/cvat/apps/engine/task.py b/cvat/apps/engine/task.py
index 669e3b543d9d..c707c126907a 100644
--- a/cvat/apps/engine/task.py
+++ b/cvat/apps/engine/task.py
@@ -48,7 +48,7 @@
 from cvat.apps.engine.models import RequestAction, RequestTarget
 from cvat.apps.engine.rq_job_handler import RQId
 from cvat.apps.engine.task_validation import HoneypotFrameSelector
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import (
     av_scan_paths,
     define_dependent_job,
@@ -71,7 +71,7 @@
 def create(
     db_task: models.Task,
     data: models.Data,
-    request: PatchedRequest,
+    request: Request,
 ) -> str:
     """Schedule a background job to create a task and return that job's identifier"""
     q = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
diff --git a/cvat/apps/engine/types.py b/cvat/apps/engine/types.py
index 9cb28b426718..f816abaed8ef 100644
--- a/cvat/apps/engine/types.py
+++ b/cvat/apps/engine/types.py
@@ -1,9 +1,13 @@
+# Copyright (C) CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
 from django.http import HttpRequest
-from rest_framework.request import Request
+from rest_framework.request import Request as RestFrameworkRequest
 
 from cvat.apps.engine.middleware import WithUUID
 from cvat.apps.iam.middleware import WithIAMContext
 
 
-class PatchedRequest(HttpRequest, Request, WithUUID, WithIAMContext):
+class Request(HttpRequest, RestFrameworkRequest, WithUUID, WithIAMContext):
     pass
diff --git a/cvat/apps/engine/utils.py b/cvat/apps/engine/utils.py
index e7d25c19ea60..f84065e4e50e 100644
--- a/cvat/apps/engine/utils.py
+++ b/cvat/apps/engine/utils.py
@@ -37,7 +37,7 @@
 from rest_framework.reverse import reverse as _reverse
 from rq.job import Dependency, Job
 
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 
 Import = namedtuple("Import", ["module", "name", "alias"])
 
@@ -222,7 +222,7 @@ def get_rq_lock_for_job(queue: DjangoRQ, rq_id: str, *, timeout: int = 60, block
     )
 
 def get_rq_job_meta(
-    request: PatchedRequest,
+    request: Request,
     db_obj: Any,
     *,
     result_url: Optional[str] = None,
@@ -262,7 +262,7 @@ def get_rq_job_meta(
 
 def reverse(viewname, *, args=None, kwargs=None,
     query_params: Optional[dict[str, str]] = None,
-    request: PatchedRequest | None = None,
+    request: Request | None = None,
 ) -> str:
     """
     The same as rest_framework's reverse(), but adds custom query params support.
@@ -277,7 +277,7 @@ def reverse(viewname, *, args=None, kwargs=None,
 
     return url
 
-def get_server_url(request: PatchedRequest) -> str:
+def get_server_url(request: Request) -> str:
     return request.build_absolute_uri('/')
 
 def build_field_filter_params(field: str, value: Any) -> dict[str, str]:
@@ -348,7 +348,7 @@ def make_attachment_file_name(filename: str) -> str:
     return filename
 
 def sendfile(
-    request: PatchedRequest, filename,
+    request: Request, filename,
     attachment=False, attachment_filename=None, mimetype=None, encoding=None
 ):
     """
@@ -421,7 +421,7 @@ def directory_tree(path, max_depth=None) -> str:
             tree += f"{indent}-{file}\n"
     return tree
 
-def is_dataset_export(request: PatchedRequest) -> bool:
+def is_dataset_export(request: Request) -> bool:
     return to_bool(request.query_params.get('save_images', False))
 
 _T = TypeVar('_T')
diff --git a/cvat/apps/engine/view_utils.py b/cvat/apps/engine/view_utils.py
index f25606f80c0b..bc654ff752d9 100644
--- a/cvat/apps/engine/view_utils.py
+++ b/cvat/apps/engine/view_utils.py
@@ -16,7 +16,7 @@
 
 from cvat.apps.engine.mixins import UploadMixin
 from cvat.apps.engine.parsers import TusUploadParser
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 
 
 def make_paginated_response(
@@ -25,7 +25,7 @@ def make_paginated_response(
     viewset: GenericViewSet,
     response_type: Optional[type[HttpResponse]] = None,
     serializer_type: Optional[type[Serializer]] = None,
-    request: Optional[type[PatchedRequest]] = None,
+    request: Optional[type[Request]] = None,
     **serializer_params
 ):
     # Adapted from the mixins.ListModelMixin.list()
diff --git a/cvat/apps/engine/views.py b/cvat/apps/engine/views.py
index 0b4154cea2fe..bfec95672652 100644
--- a/cvat/apps/engine/views.py
+++ b/cvat/apps/engine/views.py
@@ -162,7 +162,7 @@
     TaskWriteSerializer,
     UserSerializer,
 )
-from cvat.apps.engine.types import PatchedRequest
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import (
     av_scan_paths,
     define_dependent_job,
@@ -208,7 +208,7 @@ def get_serializer(self, *args, **kwargs):
     @action(detail=False, methods=['GET'], serializer_class=AboutSerializer,
         permission_classes=[] # This endpoint is available for everyone
     )
-    def about(request: PatchedRequest):
+    def about(request: Request):
         from cvat import __version__ as cvat_version
         about = {
             "name": "Computer Vision Annotation Tool",
@@ -238,7 +238,7 @@ def about(request: PatchedRequest):
             '200' : FileInfoSerializer(many=True)
         })
     @action(detail=False, methods=['GET'], serializer_class=FileInfoSerializer)
-    def share(request: PatchedRequest):
+    def share(request: Request):
         directory_param = request.query_params.get('directory', '/')
         search_param = request.query_params.get('search', '')
 
@@ -283,7 +283,7 @@ def share(request: PatchedRequest):
             '200': DatasetFormatsSerializer,
         })
     @action(detail=False, methods=['GET'], url_path='annotation/formats')
-    def annotation_formats(request: PatchedRequest):
+    def annotation_formats(request: Request):
         data = dm.views.get_all_formats()
         return Response(DatasetFormatsSerializer(data).data)
 
@@ -294,7 +294,7 @@ def annotation_formats(request: PatchedRequest):
             '200': PluginsSerializer,
         })
     @action(detail=False, methods=['GET'], url_path='plugins', serializer_class=PluginsSerializer)
-    def plugins(request: PatchedRequest):
+    def plugins(request: Request):
         data = {
             'GIT_INTEGRATION': False, # kept for backwards compatibility
             'ANALYTICS': settings.ANALYTICS_ENABLED,
@@ -471,7 +471,7 @@ def get_export_callback(self, save_images: bool) -> Callable:
         url_path=r'dataset/?$', parser_classes=_UPLOAD_PARSER_CLASSES,
         csrf_workaround_is_needed=lambda qp:
             csrf_workaround_is_needed_for_export(qp) and qp.get("action") != "import_status")
-    def dataset(self, request: PatchedRequest, pk: int):
+    def dataset(self, request: Request, pk: int):
         self._object = self.get_object() # force call of check_object_permissions()
 
         if request.method in {'POST', 'OPTIONS'}:
@@ -523,7 +523,7 @@ def dataset(self, request: PatchedRequest, pk: int):
                 return self.export_dataset_v1(request=request, save_images=True)
 
     @tus_chunk_action(detail=True, suffix_base="dataset")
-    def append_dataset_chunk(self, request: PatchedRequest, pk: int, file_id: str):
+    def append_dataset_chunk(self, request: Request, pk: int, file_id: str):
         self._object = self.get_object()
         return self.append_tus_chunk(request, file_id)
 
@@ -611,7 +611,7 @@ def upload_finished(self, request):
     @action(detail=True, methods=['GET'],
         serializer_class=LabeledDataSerializer,
         csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
-    def annotations(self, request: PatchedRequest, pk: int):
+    def annotations(self, request: Request, pk: int):
         # FUTURE-TODO: mark exporting dataset using this endpoint as deprecated when new API for result file downloading will be implemented
         self._object = self.get_object() # force call of check_object_permissions()
         return self.export_dataset_v1(request=request, save_images=False)
@@ -648,7 +648,7 @@ def annotations(self, request: PatchedRequest, pk: int):
         })
     @action(methods=['GET'], detail=True, url_path='backup',
         csrf_workaround_is_needed=csrf_workaround_is_needed_for_backup)
-    def export_backup(self, request: PatchedRequest, pk: int):
+    def export_backup(self, request: Request, pk: int):
         # FUTURE-TODO: mark this endpoint as deprecated when new API for result file downloading will be implemented
         return self.export_backup_v1(request)
 
@@ -693,11 +693,11 @@ def export_backup(self, request: PatchedRequest, pk: int):
     @action(detail=False, methods=['OPTIONS', 'POST'], url_path=r'backup/?$',
         serializer_class=None,
         parser_classes=_UPLOAD_PARSER_CLASSES)
-    def import_backup(self, request: PatchedRequest):
+    def import_backup(self, request: Request):
         return self.import_backup_v1(request, backup.import_project)
 
     @tus_chunk_action(detail=False, suffix_base="backup")
-    def append_backup_chunk(self, request: PatchedRequest, file_id: str):
+    def append_backup_chunk(self, request: Request, file_id: str):
         return self.append_tus_chunk(request, file_id)
 
     @extend_schema(summary='Get a preview image for a project',
@@ -707,7 +707,7 @@ def append_backup_chunk(self, request: PatchedRequest, file_id: str):
 
         })
     @action(detail=True, methods=['GET'], url_path='preview')
-    def preview(self, request: PatchedRequest, pk: int):
+    def preview(self, request: Request, pk: int):
         self._object = self.get_object() # call check_object_permissions as well
 
         first_task: Optional[models.Task] = self._object.tasks.order_by('-id').first()
@@ -1048,11 +1048,11 @@ def get_queryset(self):
     @action(detail=False, methods=['OPTIONS', 'POST'], url_path=r'backup/?$',
         serializer_class=None,
         parser_classes=_UPLOAD_PARSER_CLASSES)
-    def import_backup(self, request: PatchedRequest):
+    def import_backup(self, request: Request):
         return self.import_backup_v1(request, backup.import_task)
 
     @tus_chunk_action(detail=False, suffix_base="backup")
-    def append_backup_chunk(self, request: PatchedRequest, file_id: str):
+    def append_backup_chunk(self, request: Request, file_id: str):
         return self.append_tus_chunk(request, file_id)
 
     @extend_schema(summary='Back up a task',
@@ -1087,7 +1087,7 @@ def append_backup_chunk(self, request: PatchedRequest, file_id: str):
         })
     @action(methods=['GET'], detail=True, url_path='backup',
         csrf_workaround_is_needed=csrf_workaround_is_needed_for_backup)
-    def export_backup(self, request: PatchedRequest, pk: int):
+    def export_backup(self, request: Request, pk: int):
         # FUTURE-TODO: mark this endpoint as deprecated when new API for result file downloading will be implemented
         if self.get_object().data is None:
             return Response(
@@ -1433,7 +1433,7 @@ def _handle_upload_backup(request):
         })
     @action(detail=True, methods=['OPTIONS', 'POST', 'GET'], url_path=r'data/?$',
         parser_classes=_UPLOAD_PARSER_CLASSES)
-    def data(self, request: PatchedRequest, pk: int):
+    def data(self, request: Request, pk: int):
         self._object = self.get_object() # call check_object_permissions as well
         if request.method == 'POST' or request.method == 'OPTIONS':
             with transaction.atomic():
@@ -1466,7 +1466,7 @@ def data(self, request: PatchedRequest, pk: int):
             return data_getter()
 
     @tus_chunk_action(detail=True, suffix_base="data")
-    def append_data_chunk(self, request: PatchedRequest, pk: int, file_id: str):
+    def append_data_chunk(self, request: Request, pk: int, file_id: str):
         self._object = self.get_object()
         return self.append_tus_chunk(request, file_id)
 
@@ -1586,7 +1586,7 @@ def get_export_callback(self, save_images: bool) -> Callable:
     @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH', 'POST', 'OPTIONS'], url_path=r'annotations/?$',
         serializer_class=None, parser_classes=_UPLOAD_PARSER_CLASSES,
         csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
-    def annotations(self, request: PatchedRequest, pk: int):
+    def annotations(self, request: Request, pk: int):
         self._object = self.get_object() # force call of check_object_permissions()
         if request.method == 'GET':
             if self._object.data:
@@ -1647,7 +1647,7 @@ def annotations(self, request: PatchedRequest, pk: int):
                 return Response(data)
 
     @tus_chunk_action(detail=True, suffix_base="annotations")
-    def append_annotations_chunk(self, request: PatchedRequest, pk: int, file_id: str):
+    def append_annotations_chunk(self, request: Request, pk: int, file_id: str):
         self._object = self.get_object()
         return self.append_tus_chunk(request, file_id)
 
@@ -1709,7 +1709,7 @@ def _get_rq_response(queue, job_id):
         })
     @action(detail=True, methods=['GET', 'PATCH'], serializer_class=DataMetaReadSerializer,
         url_path='data/meta')
-    def metadata(self, request: PatchedRequest, pk: int):
+    def metadata(self, request: Request, pk: int):
         self.get_object() #force to call check_object_permissions
         db_task = models.Task.objects.prefetch_related(
             'segment_set',
@@ -1782,7 +1782,7 @@ def metadata(self, request: PatchedRequest, pk: int):
     )
     @action(detail=True, methods=['GET'], serializer_class=None,
         url_path='dataset', csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
-    def dataset_export(self, request: PatchedRequest, pk: int):
+    def dataset_export(self, request: Request, pk: int):
         # FUTURE-TODO: mark this endpoint as deprecated when new API for result file downloading will be implemented
         self._object = self.get_object() # force call of check_object_permissions()
 
@@ -1848,7 +1848,7 @@ def preview(self, request, pk):
         ])
     @action(detail=True, methods=["GET", "PATCH"], url_path='validation_layout')
     @transaction.atomic
-    def validation_layout(self, request: PatchedRequest, pk: int):
+    def validation_layout(self, request: Request, pk: int):
         db_task = cast(models.Task, self.get_object()) # call check_object_permissions as well
 
         validation_layout = getattr(db_task.data, 'validation_layout', None)
@@ -2166,7 +2166,7 @@ def upload_finished(self, request):
     @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH', 'POST', 'OPTIONS'], url_path=r'annotations/?$',
         serializer_class=LabeledDataSerializer, parser_classes=_UPLOAD_PARSER_CLASSES,
         csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
-    def annotations(self, request: PatchedRequest, pk: int):
+    def annotations(self, request: Request, pk: int):
         self._object: models.Job = self.get_object() # force call of check_object_permissions()
         if request.method == 'GET':
             # FUTURE-TODO: mark as deprecated using this endpoint to export annotations when new API for result file downloading will be implemented
@@ -2228,7 +2228,7 @@ def annotations(self, request: PatchedRequest, pk: int):
 
 
     @tus_chunk_action(detail=True, suffix_base="annotations")
-    def append_annotations_chunk(self, request: PatchedRequest, pk: int, file_id: str):
+    def append_annotations_chunk(self, request: Request, pk: int, file_id: str):
         self._object = self.get_object()
         return self.append_tus_chunk(request, file_id)
 
@@ -2269,7 +2269,7 @@ def append_annotations_chunk(self, request: PatchedRequest, pk: int, file_id: st
     )
     @action(detail=True, methods=['GET'], serializer_class=None,
         url_path='dataset', csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
-    def dataset_export(self, request: PatchedRequest, pk: int):
+    def dataset_export(self, request: Request, pk: int):
         # FUTURE-TODO: mark this endpoint as deprecated when new API for result file downloading will be implemented
         self._object = self.get_object() # force call of check_object_permissions()
 
@@ -2301,7 +2301,7 @@ def get_export_callback(self, save_images: bool) -> Callable:
     @action(detail=True, methods=['GET'],
         simple_filters=[] # type query parameter conflicts with the filter
     )
-    def data(self, request: PatchedRequest, pk: int):
+    def data(self, request: Request, pk: int):
         db_job = self.get_object() # call check_object_permissions as well
         data_type = request.query_params.get('type', None)
         data_num = request.query_params.get('number', None)
@@ -2327,7 +2327,7 @@ def data(self, request: PatchedRequest, pk: int):
         }, versions=['2.0'])
     @action(detail=True, methods=['GET', 'PATCH'], serializer_class=DataMetaReadSerializer,
         url_path='data/meta')
-    def metadata(self, request: PatchedRequest, pk: int):
+    def metadata(self, request: Request, pk: int):
         self.get_object() # force call of check_object_permissions()
 
         db_job = models.Job.objects.select_related(
@@ -2414,7 +2414,7 @@ def metadata(self, request: PatchedRequest, pk: int):
             '200': OpenApiResponse(description='Job image preview'),
         })
     @action(detail=True, methods=['GET'], url_path='preview')
-    def preview(self, request: PatchedRequest, pk: int):
+    def preview(self, request: Request, pk: int):
         self._object = self.get_object() # call check_object_permissions as well
 
         data_getter = _JobDataGetter(
@@ -2453,7 +2453,7 @@ def preview(self, request: PatchedRequest, pk: int):
         ])
     @action(detail=True, methods=["GET", "PATCH"], url_path='validation_layout')
     @transaction.atomic
-    def validation_layout(self, request: PatchedRequest, pk: int):
+    def validation_layout(self, request: Request, pk: int):
         self.get_object() # call check_object_permissions as well
 
         db_job = models.Job.objects.prefetch_related(
@@ -2857,7 +2857,7 @@ def get_serializer_class(self):
                 ], resource_type_field_name=None),
         })
     @action(detail=False, methods=['GET'])
-    def self(self, request: PatchedRequest):
+    def self(self, request: Request):
         """
         Method returns an instance of a user who is currently authenticated
         """
@@ -2940,7 +2940,7 @@ def perform_create(self, serializer):
             owner=self.request.user,
             organization=self.request.iam_context['organization'])
 
-    def create(self, request: PatchedRequest, *args, **kwargs):
+    def create(self, request: Request, *args, **kwargs):
         try:
             response = super().create(request, *args, **kwargs)
         except ValidationError as exceptions:
@@ -2971,7 +2971,7 @@ def create(self, request: PatchedRequest, *args, **kwargs):
         },
     )
     @action(detail=True, methods=['GET'], url_path='content-v2')
-    def content_v2(self, request: PatchedRequest, pk: int):
+    def content_v2(self, request: Request, pk: int):
         storage = None
         try:
             db_storage = self.get_object()
@@ -3038,7 +3038,7 @@ def content_v2(self, request: PatchedRequest, pk: int):
             '404': OpenApiResponse(description='Cloud Storage preview not found'),
         })
     @action(detail=True, methods=['GET'], url_path='preview')
-    def preview(self, request: PatchedRequest, pk: int):
+    def preview(self, request: Request, pk: int):
         try:
             db_storage = self.get_object()
             cache = MediaCache()
@@ -3077,7 +3077,7 @@ def preview(self, request: PatchedRequest, pk: int):
             '200': OpenApiResponse(response=OpenApiTypes.STR, description='Cloud Storage status (AVAILABLE | NOT_FOUND | FORBIDDEN)'),
         })
     @action(detail=True, methods=['GET'], url_path='status')
-    def status(self, request: PatchedRequest, pk: int):
+    def status(self, request: Request, pk: int):
         try:
             db_storage = self.get_object()
             storage = db_storage_to_storage_instance(db_storage)
@@ -3096,7 +3096,7 @@ def status(self, request: PatchedRequest, pk: int):
             '200': OpenApiResponse(response=OpenApiTypes.STR, description='Cloud Storage actions (GET | PUT | DELETE)'),
         })
     @action(detail=True, methods=['GET'], url_path='actions')
-    def actions(self, request: PatchedRequest, pk: int):
+    def actions(self, request: Request, pk: int):
         '''
         Method return allowed actions for cloud storage. It's required for reading/writing
         '''
@@ -3167,7 +3167,7 @@ def get_serializer_class(self):
         else:
             return AssetWriteSerializer
 
-    def create(self, request: PatchedRequest, *args, **kwargs):
+    def create(self, request: Request, *args, **kwargs):
         file = request.data.get('file', None)
         if not file:
             raise ValidationError('Asset file was not provided')
@@ -3725,7 +3725,7 @@ def wrapper(*args, **kwargs):
 
     @method_decorator(never_cache)
     @_handle_redis_exceptions
-    def retrieve(self, request: PatchedRequest, pk: str):
+    def retrieve(self, request: Request, pk: str):
         job = self._get_rq_job_by_id(pk)
 
         if not job:
@@ -3738,7 +3738,7 @@ def retrieve(self, request: PatchedRequest, pk: str):
 
     @method_decorator(never_cache)
     @_handle_redis_exceptions
-    def list(self, request: PatchedRequest):
+    def list(self, request: Request):
         user_id = request.user.id
         user_jobs = self._get_rq_jobs(user_id)
 
@@ -3762,7 +3762,7 @@ def list(self, request: PatchedRequest):
     @method_decorator(never_cache)
     @action(detail=True, methods=['POST'], url_path='cancel')
     @_handle_redis_exceptions
-    def cancel(self, request: PatchedRequest, pk: str):
+    def cancel(self, request: Request, pk: str):
         rq_job = self._get_rq_job_by_id(pk)
 
         if not rq_job:
diff --git a/cvat/apps/lambda_manager/views.py b/cvat/apps/lambda_manager/views.py
index 9edb5889ae75..981c75c229ed 100644
--- a/cvat/apps/lambda_manager/views.py
+++ b/cvat/apps/lambda_manager/views.py
@@ -31,7 +31,6 @@
     inline_serializer,
 )
 from rest_framework import serializers, status, viewsets
-from rest_framework.request import Request
 from rest_framework.response import Response
 
 import cvat.apps.dataset_manager as dm
@@ -48,6 +47,7 @@
 )
 from cvat.apps.engine.rq_job_handler import RQId, RQJobMetaField
 from cvat.apps.engine.serializers import LabeledDataSerializer
+from cvat.apps.engine.types import Request
 from cvat.apps.engine.utils import define_dependent_job, get_rq_job_meta, get_rq_lock_by_user
 from cvat.apps.events.handlers import handle_function_call
 from cvat.apps.iam.filters import ORGANIZATION_OPEN_API_PARAMETERS
diff --git a/cvat/apps/quality_control/quality_reports.py b/cvat/apps/quality_control/quality_reports.py
index 8d16c47c7af3..91b8361fcf5b 100644
--- a/cvat/apps/quality_control/quality_reports.py
+++ b/cvat/apps/quality_control/quality_reports.py
@@ -25,7 +25,6 @@
 from django.conf import settings
 from django.db import transaction
 from django_rq.queues import DjangoRQ as RqQueue
-from rest_framework.request import Request
 from rq.job import Job as RqJob
 from rq_scheduler import Scheduler as RqScheduler
 from scipy.optimize import linear_sum_assignment
@@ -54,6 +53,7 @@
     User,
     ValidationMode,
 )
+from cvat.apps.engine.types import Request
 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