use the approach from check_permissions to avoid n plus 1 query issue (#814)

Co-authored-by: burnettk <burnettk@users.noreply.github.com>

spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py

@@ -15,10 +15,8 @@ from sqlalchemy import or_
 
 from spiffworkflow_backend.exceptions.api_error import ApiError
 from spiffworkflow_backend.exceptions.process_entity_not_found_error import ProcessEntityNotFoundError
-from spiffworkflow_backend.models.db import db
 from spiffworkflow_backend.models.human_task import HumanTaskModel
 from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
-from spiffworkflow_backend.models.permission_assignment import PermissionAssignmentModel
 from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
 from spiffworkflow_backend.models.process_instance_file_data import ProcessInstanceFileDataModel
@@ -31,7 +29,6 @@ from spiffworkflow_backend.services.git_service import GitService
 from spiffworkflow_backend.services.process_caller_service import ProcessCallerService
 from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
 from spiffworkflow_backend.services.process_model_service import ProcessModelService
-from spiffworkflow_backend.services.user_service import UserService
 
 process_api_blueprint = Blueprint("process_api", __name__)
 
@@ -49,14 +46,7 @@ def permissions_check(body: dict[str, dict[str, list[str]]]) -> flask.wrappers.R
     requests_to_check = body["requests_to_check"]
 
     user = g.user
-    principals = UserService.all_principals_for_user(user)
+    permission_assignments = AuthorizationService.all_permission_assignments_for_user(user=user)
-    principal_ids = [p.id for p in principals]
-
-    permission_assignments = (
-        PermissionAssignmentModel.query.filter(PermissionAssignmentModel.principal_id.in_(principal_ids))
-        .options(db.joinedload(PermissionAssignmentModel.permission_target))
-        .all()
-    )
 
     for target_uri, http_methods in requests_to_check.items():
         if target_uri not in response_dict:

spiffworkflow-backend/src/spiffworkflow_backend/services/authorization_service.py

@@ -29,7 +29,6 @@ from spiffworkflow_backend.models.group import GroupModel
 from spiffworkflow_backend.models.human_task import HumanTaskModel
 from spiffworkflow_backend.models.permission_assignment import PermissionAssignmentModel
 from spiffworkflow_backend.models.permission_target import PermissionTargetModel
-from spiffworkflow_backend.models.principal import MissingPrincipalError
 from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.service_account import SPIFF_SERVICE_ACCOUNT_AUTH_SERVICE
 from spiffworkflow_backend.models.task import TaskModel  # noqa: F401
@@ -118,18 +117,20 @@ class AuthorizationService:
 
     @classmethod
     def user_has_permission(cls, user: UserModel, permission: str, target_uri: str) -> bool:
-        if user.principal is None:
+        principals = UserService.all_principals_for_user(user)
-            raise MissingPrincipalError(f"Missing principal for user with id: {user.id}")
-
-        principals = [user.principal]
-
-        for group in user.groups:
-            if group.principal is None:
-                raise MissingPrincipalError(f"Missing principal for group with id: {group.id}")
-            principals.append(group.principal)
-
         return cls.has_permission(principals, permission, target_uri)
 
+    @classmethod
+    def all_permission_assignments_for_user(cls, user: UserModel) -> list[PermissionAssignmentModel]:
+        principals = UserService.all_principals_for_user(user)
+        principal_ids = [p.id for p in principals]
+        permission_assignments: list[PermissionAssignmentModel] = (
+            PermissionAssignmentModel.query.filter(PermissionAssignmentModel.principal_id.in_(principal_ids))
+            .options(db.joinedload(PermissionAssignmentModel.permission_target))
+            .all()
+        )
+        return permission_assignments
+
     @classmethod
     def permission_assignments_include(
         cls, permission_assignments: list[PermissionAssignmentModel], permission: str, target_uri: str

spiffworkflow-backend/src/spiffworkflow_backend/services/process_model_service.py

@@ -267,14 +267,22 @@ class ProcessModelService(FileSystemService):
             permission=permission_to_check,
             target_uri=f"{permission_base_uri}/{guid_of_non_existent_item_to_check_perms_against}",
         )
+
+        # if user has access to uri/* with that permission then there's no reason to check each one individually
         if has_permission:
             return process_model_identifiers
 
+        permission_assignments = AuthorizationService.all_permission_assignments_for_user(user=user)
+
         permitted_process_model_identifiers = []
         for process_model_identifier in process_model_identifiers:
             modified_process_model_id = ProcessModelInfo.modify_process_identifier_for_path_param(process_model_identifier)
             uri = f"{permission_base_uri}/{modified_process_model_id}"
-            has_permission = AuthorizationService.user_has_permission(user=user, permission=permission_to_check, target_uri=uri)
+            has_permission = AuthorizationService.permission_assignments_include(
+                permission_assignments=permission_assignments,
+                permission=permission_to_check,
+                target_uri=uri,
+            )
             if has_permission:
                 permitted_process_model_identifiers.append(process_model_identifier)
 

spiffworkflow-backend/src/spiffworkflow_backend/services/user_service.py

@@ -212,6 +212,7 @@ class UserService:
     def all_principals_for_user(cls, user: UserModel) -> list[PrincipalModel]:
         if user.principal is None:
             raise MissingPrincipalError(f"Missing principal for user with id: {user.id}")
+
         principals = [user.principal]
 
         for group in user.groups: