feat: add custom exceptions

src/application/domain/exceptions/__init__.py

@@ -1 +1,9 @@
-from src.application.domain.exceptions.application_exceptions import ApplicationException
+from src.application.domain.exceptions.application_exception import ApplicationException
+from src.application.domain.exceptions.bad_request_exception import BadRequestException
+from src.application.domain.exceptions.conflict_exception import ConflictException
+from src.application.domain.exceptions.forbidden_exception import ForbiddenException
+from src.application.domain.exceptions.internal_server_exception import InternalServerException
+from src.application.domain.exceptions.not_found_exception import NotFoundException
+from src.application.domain.exceptions.service_unavailable_exception import ServiceUnavailableException
+from src.application.domain.exceptions.too_many_requests_exception import TooManyRequestsException
+from src.application.domain.exceptions.unauthorized_exception import UnauthorizedException

src/application/domain/exceptions/application_exception.py

@@ -15,4 +15,4 @@ class ApplicationException(Exception):
         self.headers = headers
 
     def __str__(self):
-        return f"{self.status_code}: {self.message}"
+        return f'{self.status_code}: {self.message}'

src/application/domain/exceptions/bad_request_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class BadRequestException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Bad Request',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/conflict_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class ConflictException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Conflict',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_409_CONFLICT,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/forbidden_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class ForbiddenException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Forbidden',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/internal_server_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class InternalServerException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Internal Server Error',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/not_found_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class NotFoundException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Not Found',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_404_NOT_FOUND,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/service_unavailable_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class ServiceUnavailableException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Service Unavailable',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/too_many_requests_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class TooManyRequestsException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Too Many Requests',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            message=message,
+            headers=headers,
+        )

src/application/domain/exceptions/unauthorized_exception.py

@@ -0,0 +1,16 @@
+from typing import Mapping
+from starlette import status
+from src.application.domain.exceptions.application_exception import ApplicationException
+
+
+class UnauthorizedException(ApplicationException):
+    def __init__(
+        self,
+        message: str = 'Unauthorized',
+        headers: Mapping[str, str] | None = None,
+    ):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            message=message,
+            headers=headers,
+        )

src/main.py

@@ -4,6 +4,7 @@ import secrets
 from typing import AsyncGenerator
 from fastapi import Depends, FastAPI, status
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
+from fastapi.openapi.utils import get_openapi
 from fastapi.responses import HTMLResponse
 from fastapi.security import HTTPBasic, HTTPBasicCredentials
 from starlette.middleware.cors import CORSMiddleware
@@ -15,13 +16,66 @@ from src.infrastructure.utils import generate_instance_id
 from src.infrastructure.logger import logger
 from src.infrastructure.config import settings
 from src.presentation.dependencies import get_rabbit
-from src.presentation.handlers import application_exception_handler, unhandled_exception_handler
+from src.presentation.handler import application_exception_handler
+from src.presentation.handler import unhandled_exception_handler
 from src.presentation.middleware import TraceIDMiddleware, SecurityHeadersMiddleware
 from src.presentation.routing import v1_router
+from src.presentation.schemas import ErrorResponse
 
 
 security = HTTPBasic()
 
+ERROR_RESPONSES: dict[int, str] = {
+    status.HTTP_400_BAD_REQUEST: 'Bad Request',
+    status.HTTP_401_UNAUTHORIZED: 'Unauthorized',
+    status.HTTP_403_FORBIDDEN: 'Forbidden',
+    status.HTTP_404_NOT_FOUND: 'Not Found',
+    status.HTTP_409_CONFLICT: 'Conflict',
+    status.HTTP_429_TOO_MANY_REQUESTS: 'Too Many Requests',
+    status.HTTP_500_INTERNAL_SERVER_ERROR: 'Internal Server Error',
+    status.HTTP_503_SERVICE_UNAVAILABLE: 'Service Unavailable',
+}
+
+
+def custom_openapi() -> dict:
+    if app.openapi_schema:
+        return app.openapi_schema
+
+    openapi_schema = get_openapi(
+        title=app.title,
+        version=app.version,
+        description=app.description,
+        routes=app.routes,
+        license_info=app.license_info,
+    )
+    components = openapi_schema.setdefault('components', {})
+    schemas = components.setdefault('schemas', {})
+    schemas['ErrorResponse'] = ErrorResponse.model_json_schema()
+
+    for path_item in openapi_schema.get('paths', {}).values():
+        for operation in path_item.values():
+            if not isinstance(operation, dict):
+                continue
+
+            responses = operation.setdefault('responses', {})
+            for status_code, description in ERROR_RESPONSES.items():
+                responses.setdefault(
+                    str(status_code),
+                    {
+                        'description': description,
+                        'content': {
+                            'application/json': {
+                                'schema': {
+                                    '$ref': '#/components/schemas/ErrorResponse',
+                                },
+                            },
+                        },
+                    },
+                )
+
+    app.openapi_schema = openapi_schema
+    return app.openapi_schema
+
 
 async def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)) -> HTTPBasicCredentials:
     user_ok = secrets.compare_digest(credentials.username, settings.DOCS_USERNAME)
@@ -152,3 +206,6 @@ async def ping() -> dict[str, str]:
         'message': 'pong',
         'status': 'ok',
     }
+
+
+app.openapi = custom_openapi

src/presentation/handler/__init__.py

@@ -0,0 +1,2 @@
+from src.presentation.handler.application_exception_handler import application_exception_handler
+from src.presentation.handler.unhandled_exception_handler import unhandled_exception_handler

src/presentation/handler/application_exception_handler.py

@@ -1,17 +1,15 @@
-from fastapi.responses import ORJSONResponse
 from fastapi import Request
+from fastapi.responses import ORJSONResponse
 from src.application.domain.exceptions import ApplicationException
 
 
 async def application_exception_handler(_request: Request, exc: ApplicationException) -> ORJSONResponse:
     detail = exc.message
     if 500 <= exc.status_code:
-        detail = "Internal Server Error"
+        detail = 'Internal Server Error'
 
     return ORJSONResponse(
         status_code=exc.status_code,
-        content={"detail": detail},
+        content={'detail': detail},
         headers=dict(exc.headers) if exc.headers else None,
     )
-
-

src/presentation/handler/unhandled_exception_handler.py

@@ -1,5 +1,5 @@
-from fastapi.responses import ORJSONResponse
 from fastapi import Request
+from fastapi.responses import ORJSONResponse
 from starlette import status
 from src.infrastructure.logger import logger
 

src/presentation/handlers/__init__.py

@@ -1,2 +0,0 @@
-from src.presentation.handlers.unhandled_handler import unhandled_exception_handler
-from src.presentation.handlers.application_handler import application_exception_handler

src/presentation/schemas/__init__.py

@@ -1 +1,5 @@
-from src.presentation.schemas.user import RegistrationStart, RegistrationComplete, UserLogin,  LoginStart
+from src.presentation.schemas.error import ErrorResponse
+from src.presentation.schemas.user import RegistrationComplete
+from src.presentation.schemas.user import RegistrationStart
+from src.presentation.schemas.user import LoginStart
+from src.presentation.schemas.user import UserLogin

src/presentation/schemas/error.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+from pydantic import Field
+
+
+class ErrorResponse(BaseModel):
+    detail: str = Field(title='Detail')