refactor: user validation

api/api/endpoints/account.py

@@ -17,7 +17,7 @@ from api.schemas.account.account import User
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.account import AllUserResponse, UserCreate, UserFilterDTO, UserUpdate
 from api.services.auth import get_current_user
-from api.services.user_role_validation import db_user_role_validation
+from api.services.user_role_validation import UserRoleValidator
 
 api_router = APIRouter(
     prefix="/account",
@@ -38,7 +38,8 @@ async def get_all_account_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     filters = {
         **({"status": status_filter} if status_filter else {}),
@@ -67,7 +68,8 @@ async def get_account_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     user = await get_user_by_id(connection, user_id)
 
@@ -83,7 +85,8 @@ async def create_account_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     user_validation = await get_user_by_login(connection, user.login)
 
@@ -104,7 +107,8 @@ async def update_account_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     user = await get_user_by_id(connection, user_id)
     if user is None:
@@ -131,7 +135,8 @@ async def delete_account_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     user = await get_user_by_id(connection, user_id)
     if user is None:

api/api/endpoints/keyring.py

@@ -13,7 +13,7 @@ from api.schemas.account.account_keyring import AccountKeyring
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.account_keyring import AccountKeyringUpdate
 from api.services.auth import get_current_user
-from api.services.user_role_validation import db_user_role_validation
+from api.services.user_role_validation import UserRoleValidator
 
 api_router = APIRouter(
     prefix="/keyring",
@@ -25,7 +25,8 @@ api_router = APIRouter(
 async def get_keyring_endpoint(
     key_id: str, connection: AsyncConnection = Depends(get_connection_dep), current_user=Depends(get_current_user)
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     keyring = await get_key_by_id(connection, key_id)
 
@@ -43,7 +44,8 @@ async def create_keyring_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     keyring = await get_key_by_id(connection, key_id)
 
@@ -69,7 +71,8 @@ async def update_keyring_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     keyring = await get_key_by_id(connection, key_id)
     if keyring is None:
@@ -94,7 +97,8 @@ async def delete_keyring_endpoint(
     connection: AsyncConnection = Depends(get_connection_dep),
     current_user=Depends(get_current_user),
 ):
-    authorize_user = await db_user_role_validation(connection, current_user)
+    validator = UserRoleValidator(connection)
+    authorize_user = await validator.validate_admin(current_user)
 
     keyring = await get_key_by_id(connection, key_id)
     if keyring is None:

api/api/endpoints/list_events.py

@@ -17,10 +17,7 @@ from api.schemas.base import bearer_schema
 from api.schemas.endpoints.list_events import AllListEventResponse, ListEventFilterDTO, ListEventUpdate
 from api.schemas.events.list_events import ListEvent
 from api.services.auth import get_current_user
-from api.services.user_role_validation import (
+from api.services.user_role_validation import UserRoleValidator
-    db_user_role_validation_for_list_events_and_process_schema,
-    db_user_role_validation_for_list_events_and_process_schema_by_list_event_id,
-)
 
 api_router = APIRouter(
     prefix="/list_events",
@@ -54,9 +51,8 @@ async def get_all_list_events_endpoint(
         filters=filters if filters else None,
     )
 
-    authorize_user, page_flag = await db_user_role_validation_for_list_events_and_process_schema(
+    validator = UserRoleValidator(connection)
-        connection, current_user
+    authorize_user, page_flag = await validator.get_user(current_user)
-    )
 
     if not page_flag:
         if filter_dto.filters is None:
@@ -82,9 +78,8 @@ async def get_list_events_endpoint(
     if list_events_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, list_events_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
-    )
 
     if list_events_id is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
@@ -124,9 +119,8 @@ async def update_list_events(
     if list_events_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, list_events_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
-    )
 
     updated_values = list_events_update.model_dump(by_alias=True, exclude_none=True)
 
@@ -151,9 +145,8 @@ async def delete_list_events_endpoint(
     if list_events_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, list_events_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
-    )
 
     list_events_update = ListEventUpdate(status=EventStatus.DELETED.value)
 

api/api/endpoints/process_schema.py

@@ -18,10 +18,7 @@ from api.schemas.process.process_schema import ProcessSchema, ProcessSchemaSetti
 from api.schemas.process.ps_node import Ps_NodeFrontResponseNode
 from api.schemas.process.ps_node import Ps_NodeFrontResponse
 from api.services.auth import get_current_user
-from api.services.user_role_validation import (
+from api.services.user_role_validation import UserRoleValidator
-    db_user_role_validation_for_list_events_and_process_schema,
-    db_user_role_validation_for_list_events_and_process_schema_by_list_event_id,
-)
 
 
 from api.db.logic.ps_node import create_ps_node_schema
@@ -78,9 +75,8 @@ async def get_all_process_schema_endpoint(
         filters=filters if filters else None,
     )
 
-    authorize_user, page_flag = await db_user_role_validation_for_list_events_and_process_schema(
+    validator = UserRoleValidator(connection)
-        connection, current_user
+    authorize_user, page_flag = await validator.get_user(current_user)
-    )
 
     if not page_flag:
         if filter_dto.filters is None:
@@ -106,9 +102,8 @@ async def get_process_schema_endpoint(
     if process_schema_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, process_schema_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
-    )
 
     if process_schema_id is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
@@ -190,9 +185,8 @@ async def update_process_schema_endpoint(
     if process_schema_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, process_schema_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
-    )
 
     updated_values = process_schema_update.model_dump(by_alias=True, exclude_none=True)
 
@@ -217,9 +211,8 @@ async def delete_process_schema_endpoint(
     if process_schema_validation is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
 
-    authorize_user = await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    validator = UserRoleValidator(connection)
-        connection, current_user, process_schema_validation.creator_id
+    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
-    )
 
     process_schema_update = ProcessSchemaUpdate(status=ProcessStatus.DELETED.value)
 

api/api/endpoints/ps_node.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, status
 
 from sqlalchemy.ext.asyncio import AsyncConnection
 
@@ -22,9 +22,7 @@ from api.db.logic.ps_node import (
 from api.db.logic.node_link import get_last_link_name_by_node_id, create_node_link_schema
 
 from api.db.logic.process_schema import update_process_schema_settings_by_id, get_process_schema_by_id
-from api.services.user_role_validation import (
+from api.services.user_role_validation import UserRoleValidator
-    db_user_role_validation_for_list_events_and_process_schema_by_list_event_id,
-)
 
 from core import VorkNodeRegistry, VorkNodeLink
 
@@ -53,10 +51,9 @@ async def delete_ps_node_endpoint(
             details={"schema_id": ps_node_delete_data.schema_id},
         )
 
+    validator = UserRoleValidator(connection)
     try:
-        await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+        await validator.validate_ownership(current_user, process_schema.creator_id)
-            connection, current_user, process_schema.creator_id
-        )
     except Exception as e:
         raise create_access_error(
             message="Access denied",
@@ -127,10 +124,9 @@ async def create_ps_node_endpoint(
             details={"schema_id": ps_node.data["ps_id"]},
         )
 
+    validator = UserRoleValidator(connection)
     try:
-        await db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+        await validator.validate_ownership(current_user, process_schema.creator_id)
-            connection, current_user, process_schema.creator_id
-        )
     except Exception as e:
         raise create_access_error(
             message="Access denied",

api/api/services/user_role_validation.py

@@ -1,32 +1,75 @@
-from fastapi import (
+from fastapi import status
-    HTTPException,
-    status,
-)
 from orm.tables.account import AccountRole
-
 from api.db.logic.account import get_user_by_login
+from api.error import create_operation_error, create_access_error
 
 
-async def db_user_role_validation(connection, current_user):
+class UserRoleValidator:
-    authorize_user = await get_user_by_login(connection, current_user)
+    """Валидатор ролей пользователей"""
-    if authorize_user.role not in {AccountRole.OWNER, AccountRole.ADMIN}:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You do not have enough permissions")
-    return authorize_user
 
+    def __init__(self, connection):
+        self.connection = connection
 
-async def db_user_role_validation_for_list_events_and_process_schema_by_list_event_id(
+    async def validate_admin(self, current_user: int):
-    connection, current_user, current_listevents_creator_id
+        """Проверяет права администратора или владельца"""
-):
+        try:
-    authorize_user = await get_user_by_login(connection, current_user)
+            authorize_user = await get_user_by_login(self.connection, current_user)
-    if authorize_user.role not in {AccountRole.OWNER, AccountRole.ADMIN}:
+        except Exception as e:
-        if authorize_user.id != current_listevents_creator_id:
+            raise create_operation_error(
-            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You do not have enough permissions")
+                message="User not found",
-    return authorize_user
+                status_code=status.HTTP_404_NOT_FOUND,
+                details={"user_id": current_user, "error": str(e)},
+            )
 
+        if authorize_user.role not in {AccountRole.OWNER, AccountRole.ADMIN}:
+            raise create_access_error(
+                message="Insufficient permissions",
+                status_code=status.HTTP_403_FORBIDDEN,
+                details={
+                    "user_id": current_user,
+                    "user_role": authorize_user.role.value,
+                    "required_roles": [AccountRole.OWNER.value, AccountRole.ADMIN.value],
+                },
+            )
 
-async def db_user_role_validation_for_list_events_and_process_schema(connection, current_user):
+        return authorize_user
-    authorize_user = await get_user_by_login(connection, current_user)
+
-    if authorize_user.role not in {AccountRole.OWNER, AccountRole.ADMIN}:
+    async def validate_ownership(self, current_user: int, resource_owner_id: int):
-        return authorize_user, False
+        """Проверяет владение ресурсом или права администратора"""
-    else:
+        try:
-        return authorize_user, True
+            authorize_user = await get_user_by_login(self.connection, current_user)
+        except Exception as e:
+            raise create_operation_error(
+                message="User not found",
+                status_code=status.HTTP_404_NOT_FOUND,
+                details={"user_id": current_user, "error": str(e)},
+            )
+
+        if authorize_user.role not in {AccountRole.OWNER, AccountRole.ADMIN}:
+            if authorize_user.id != resource_owner_id:
+                raise create_access_error(
+                    message="Access denied",
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    details={
+                        "user_id": current_user,
+                        "resource_owner_id": resource_owner_id,
+                        "user_role": authorize_user.role.value,
+                        "reason": "User is not the owner and does not have admin privileges",
+                    },
+                )
+
+        return authorize_user
+
+    async def get_user(self, current_user: int):
+        """Получает пользователя с админ-статусом"""
+        try:
+            authorize_user = await get_user_by_login(self.connection, current_user)
+        except Exception as e:
+            raise create_operation_error(
+                message="User not found",
+                status_code=status.HTTP_404_NOT_FOUND,
+                details={"user_id": current_user, "error": str(e)},
+            )
+
+        is_admin = authorize_user.role in {AccountRole.OWNER, AccountRole.ADMIN}
+        return authorize_user, is_admin