Http codes with map 774

app/api/audit/adapter.py

@@ -4,19 +4,11 @@
 License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
 """
 
-from typing import ParamSpec, TypeVar
-
-from fastapi import status
-
 from api.base_adapter import BaseAdapter
 from ldap_protocol.policies.audit.dataclasses import (
     AuditDestinationDTO,
     AuditPolicyDTO,
 )
-from ldap_protocol.policies.audit.exception import (
-    AuditAlreadyExistsError,
-    AuditNotFoundError,
-)
 from ldap_protocol.policies.audit.schemas import (
     AuditDestinationResponse,
     AuditDestinationSchemaRequest,
@@ -25,18 +17,10 @@
 )
 from ldap_protocol.policies.audit.service import AuditService
 
-P = ParamSpec("P")
-R = TypeVar("R")
-
 
 class AuditPoliciesAdapter(BaseAdapter[AuditService]):
     """Adapter for audit policies."""
 
-    _exceptions_map: dict[type[Exception], int] = {
-        AuditNotFoundError: status.HTTP_404_NOT_FOUND,
-        AuditAlreadyExistsError: status.HTTP_409_CONFLICT,
-    }
-
     async def get_policies(self) -> list[AuditPolicyResponse]:
         """Get all audit policies."""
         return [

app/api/audit/router.py

@@ -5,10 +5,21 @@
 """
 
 from dishka import FromDishka
-from dishka.integrations.fastapi import DishkaRoute
-from fastapi import APIRouter, Depends, status
+from fastapi import Depends, status
+from fastapi_error_map.routing import ErrorAwareRouter
+from fastapi_error_map.rules import rule
 
 from api.auth.utils import verify_auth
+from api.error_routing import (
+    ERROR_MAP_TYPE,
+    DishkaErrorAwareRoute,
+    DomainErrorTranslator,
+)
+from enums import DomainCodes
+from ldap_protocol.policies.audit.exception import (
+    AuditAlreadyExistsError,
+    AuditNotFoundError,
+)
 from ldap_protocol.policies.audit.schemas import (
     AuditDestinationResponse,
     AuditDestinationSchemaRequest,
@@ -18,23 +29,37 @@
 
 from .adapter import AuditPoliciesAdapter
 
-audit_router = APIRouter(
+translator = DomainErrorTranslator(DomainCodes.AUDIT)
+
+
+error_map: ERROR_MAP_TYPE = {
+    AuditNotFoundError: rule(
+        status=status.HTTP_400_BAD_REQUEST,
+        translator=translator,
+    ),
+    AuditAlreadyExistsError: rule(
+        status=status.HTTP_400_BAD_REQUEST,
+        translator=translator,
+    ),
+}
+
+audit_router = ErrorAwareRouter(
     prefix="/audit",
     tags=["Audit policy"],
     dependencies=[Depends(verify_auth)],
-    route_class=DishkaRoute,
+    route_class=DishkaErrorAwareRoute,
 )
 
 
-@audit_router.get("/policies")
+@audit_router.get("/policies", error_map=error_map)
 async def get_audit_policies(
     audit_adapter: FromDishka[AuditPoliciesAdapter],
 ) -> list[AuditPolicyResponse]:
     """Get all audit policies."""
     return await audit_adapter.get_policies()
 
 
-@audit_router.put("/policy/{policy_id}")
+@audit_router.put("/policy/{policy_id}", error_map=error_map)
 async def update_audit_policy(
     policy_id: int,
     policy_data: AuditPolicySchemaRequest,
@@ -44,15 +69,19 @@ async def update_audit_policy(
     return await audit_adapter.update_policy(policy_id, policy_data)
 
 
-@audit_router.get("/destinations")
+@audit_router.get("/destinations", error_map=error_map)
 async def get_audit_destinations(
     audit_adapter: FromDishka[AuditPoliciesAdapter],
 ) -> list[AuditDestinationResponse]:
     """Get all audit destinations."""
     return await audit_adapter.get_destinations()
 
 
-@audit_router.post("/destination", status_code=status.HTTP_201_CREATED)
+@audit_router.post(
+    "/destination",
+    status_code=status.HTTP_201_CREATED,
+    error_map=error_map,
+)
 async def create_audit_destination(
     destination_data: AuditDestinationSchemaRequest,
     audit_adapter: FromDishka[AuditPoliciesAdapter],
@@ -61,7 +90,7 @@ async def create_audit_destination(
     return await audit_adapter.create_destination(destination_data)
 
 
-@audit_router.delete("/destination/{destination_id}")
+@audit_router.delete("/destination/{destination_id}", error_map=error_map)
 async def delete_audit_destination(
     destination_id: int,
     audit_adapter: FromDishka[AuditPoliciesAdapter],
@@ -70,7 +99,7 @@ async def delete_audit_destination(
     await audit_adapter.delete_destination(destination_id)
 
 
-@audit_router.put("/destination/{destination_id}")
+@audit_router.put("/destination/{destination_id}", error_map=error_map)
 async def update_audit_destination(
     destination_id: int,
     destination_data: AuditDestinationSchemaRequest,

app/api/auth/adapters/auth.py

@@ -7,54 +7,24 @@
 from ipaddress import IPv4Address, IPv6Address
 
 from adaptix.conversion import get_converter
-from fastapi import Request, status
+from fastapi import Request
 
 from api.base_adapter import BaseAdapter
 from ldap_protocol.auth import AuthManager
 from ldap_protocol.auth.dto import SetupDTO
-from ldap_protocol.auth.exceptions.mfa import (
-    MFAAPIError,
-    MFAConnectError,
-    MFARequiredError,
-    MissingMFACredentialsError,
-)
 from ldap_protocol.auth.schemas import (
     MFAChallengeResponse,
     OAuth2Form,
     SetupRequest,
 )
 from ldap_protocol.dialogue import UserSchema
-from ldap_protocol.identity.exceptions import (
-    AlreadyConfiguredError,
-    AuthValidationError,
-    LoginFailedError,
-    PasswordPolicyError,
-    UnauthorizedError,
-    UserNotFoundError,
-)
-from ldap_protocol.kerberos.exceptions import KRBAPIChangePasswordError
 
 _convert_request_to_dto = get_converter(SetupRequest, SetupDTO)
 
 
 class AuthFastAPIAdapter(BaseAdapter[AuthManager]):
     """Adapter for using IdentityManager with FastAPI."""
 
-    _exceptions_map: dict[type[Exception], int] = {
-        UnauthorizedError: status.HTTP_401_UNAUTHORIZED,
-        LoginFailedError: status.HTTP_403_FORBIDDEN,
-        MFARequiredError: status.HTTP_426_UPGRADE_REQUIRED,
-        PasswordPolicyError: status.HTTP_422_UNPROCESSABLE_ENTITY,
-        AuthValidationError: status.HTTP_422_UNPROCESSABLE_ENTITY,
-        PermissionError: status.HTTP_403_FORBIDDEN,
-        UserNotFoundError: status.HTTP_404_NOT_FOUND,
-        KRBAPIChangePasswordError: status.HTTP_424_FAILED_DEPENDENCY,
-        AlreadyConfiguredError: status.HTTP_423_LOCKED,
-        MissingMFACredentialsError: status.HTTP_403_FORBIDDEN,
-        MFAAPIError: status.HTTP_406_NOT_ACCEPTABLE,
-        MFAConnectError: status.HTTP_406_NOT_ACCEPTABLE,
-    }
-
     async def login(
         self,
         form: OAuth2Form,

app/api/auth/adapters/mfa.py

@@ -11,33 +11,14 @@
 
 from api.base_adapter import BaseAdapter
 from ldap_protocol.auth import MFAManager
-from ldap_protocol.auth.exceptions.mfa import (
-    ForbiddenError,
-    InvalidCredentialsError,
-    MFAAPIError,
-    MFAConnectError,
-    MFATokenError,
-    MissingMFACredentialsError,
-    NetworkPolicyError,
-    NotFoundError,
-)
+from ldap_protocol.auth.exceptions.mfa import MFATokenError
 from ldap_protocol.auth.schemas import MFACreateRequest, MFAGetResponse
 from ldap_protocol.multifactor import MFA_HTTP_Creds, MFA_LDAP_Creds
 
 
 class MFAFastAPIAdapter(BaseAdapter[MFAManager]):
     """Adapter for using MFAManager with FastAPI."""
 
-    _exceptions_map: dict[type[Exception], int] = {
-        MissingMFACredentialsError: status.HTTP_403_FORBIDDEN,
-        NetworkPolicyError: status.HTTP_403_FORBIDDEN,
-        ForbiddenError: status.HTTP_403_FORBIDDEN,
-        InvalidCredentialsError: status.HTTP_422_UNPROCESSABLE_ENTITY,
-        NotFoundError: status.HTTP_404_NOT_FOUND,
-        MFAAPIError: status.HTTP_406_NOT_ACCEPTABLE,
-        MFAConnectError: status.HTTP_406_NOT_ACCEPTABLE,
-    }
-
     async def setup_mfa(self, mfa: MFACreateRequest) -> bool:
         """Create or update MFA keys.
 

app/api/auth/adapters/session_gateway.py

@@ -5,8 +5,6 @@
 from ipaddress import IPv4Address, IPv6Address
 from typing import Literal, ParamSpec, TypeVar
 
-from fastapi import status
-
 from api.base_adapter import BaseAdapter
 from ldap_protocol.session_storage import SessionRepository
 
@@ -54,9 +52,9 @@ class UserSessionsResponseSchema:
 class SessionFastAPIGateway(BaseAdapter[SessionRepository]):
     """Base class for session storage."""
 
-    _exceptions_map: dict[type[Exception], int] = {
-        LookupError: status.HTTP_404_NOT_FOUND,
-    }
+    def __init__(self, repository: SessionRepository) -> None:
+        """Initialize the session gateway with a repository."""
+        self._service = repository
 
     async def get_user_sessions(
         self,