refactor to reduce complexity (sonarcloud)

Author: pcrespov

services/web/server/src/simcore_service_webserver/folders/_folders_repository.py

@@ -95,40 +95,17 @@ async def create(
         return FolderDB.model_validate(row)
 
 
-async def list_(  # pylint: disable=too-many-arguments,too-many-branches
-    app: web.Application,
-    connection: AsyncConnection | None = None,
-    *,
+def _create_private_workspace_query(
     product_name: ProductName,
     user_id: UserID,
-    # hierarchy filters
-    folder_query: FolderQuery,
-    workspace_query: WorkspaceQuery,
-    # attribute filters
-    filter_trashed: bool | None,
-    filter_by_text: str | None,
-    # pagination
-    offset: NonNegativeInt,
-    limit: int,
-    # order
-    order_by: OrderBy,
-) -> tuple[int, list[UserFolderAccessRightsDB]]:
-    """
-    folder_query - Used to filter in which folder we want to list folders.
-    trashed - If set to true, it returns folders **explicitly** trashed, if false then non-trashed folders.
-    """
-
-    workspace_access_rights_subquery = create_my_workspace_access_rights_subquery(
-        user_id=user_id
-    )
-
-    if workspace_query.workspace_scope is not WorkspaceScope.SHARED:
-        assert workspace_query.workspace_scope in (  # nosec
+    workspace_scope: WorkspaceScope,
+):
+    if workspace_scope is not WorkspaceScope.SHARED:
+        assert workspace_scope in (  # nosec
             WorkspaceScope.PRIVATE,
             WorkspaceScope.ALL,
         )
-
-        private_workspace_query = (
+        return (
             select(
                 *_SELECTION_ARGS,
                 func.json_build_object(
@@ -146,15 +123,25 @@ async def list_(  # pylint: disable=too-many-arguments,too-many-branches
                 & (folders_v2.c.user_id == user_id)
             )
         )
-    else:
-        private_workspace_query = None
+    return None
 
-    if workspace_query.workspace_scope is not WorkspaceScope.PRIVATE:
-        assert workspace_query.workspace_scope in (  # nosec
+
+def _create_shared_workspace_query(
+    product_name: ProductName,
+    user_id: UserID,
+    workspace_scope: WorkspaceScope,
+    workspace_id: WorkspaceID | None,
+):
+    if workspace_scope is not WorkspaceScope.PRIVATE:
+        assert workspace_scope in (  # nosec
             WorkspaceScope.SHARED,
             WorkspaceScope.ALL,
         )
 
+        workspace_access_rights_subquery = create_my_workspace_access_rights_subquery(
+            user_id=user_id
+        )
+
         shared_workspace_query = (
             select(
                 *_SELECTION_ARGS, workspace_access_rights_subquery.c.my_access_rights
@@ -172,14 +159,52 @@ async def list_(  # pylint: disable=too-many-arguments,too-many-branches
             )
         )
 
-        if workspace_query.workspace_scope == WorkspaceScope.SHARED:
+        if workspace_scope == WorkspaceScope.SHARED:
             shared_workspace_query = shared_workspace_query.where(
-                folders_v2.c.workspace_id == workspace_query.workspace_id
+                folders_v2.c.workspace_id == workspace_id
             )
 
     else:
         shared_workspace_query = None
 
+    return shared_workspace_query
+
+
+async def list_(  # pylint: disable=too-many-arguments,too-many-branches
+    app: web.Application,
+    connection: AsyncConnection | None = None,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    # hierarchy filters
+    folder_query: FolderQuery,
+    workspace_query: WorkspaceQuery,
+    # attribute filters
+    filter_trashed: bool | None,
+    filter_by_text: str | None,
+    # pagination
+    offset: NonNegativeInt,
+    limit: int,
+    # order
+    order_by: OrderBy,
+) -> tuple[int, list[UserFolderAccessRightsDB]]:
+    """
+    folder_query - Used to filter in which folder we want to list folders.
+    trashed - If set to true, it returns folders **explicitly** trashed, if false then non-trashed folders.
+    """
+
+    private_workspace_query = _create_private_workspace_query(
+        workspace_scope=workspace_query.workspace_scope,
+        product_name=product_name,
+        user_id=user_id,
+    )
+    shared_workspace_query = _create_shared_workspace_query(
+        workspace_scope=workspace_query.workspace_scope,
+        product_name=product_name,
+        user_id=user_id,
+        workspace_id=workspace_query.workspace_id,
+    )
+
     attributes_filters: list[ColumnElement] = []
 
     if filter_trashed is not None: