From 611e10620e2bb4b6c407482fa636ab873bcf9001 Mon Sep 17 00:00:00 2001 From: Kitos Date: Wed, 18 Feb 2026 13:44:47 +0100 Subject: [PATCH] refactor(domain): introduce domain exceptions boundary - Create domain/errors.py as canonical error hierarchy: DomainError, InvalidStateTransition, PermissionViolation, BusinessRuleViolation, EntityNotFoundError, DuplicateEntityError - InvalidOperationError now inherits from BusinessRuleViolation for semantic consistency - Convert domain/exceptions.py to backward-compatible re-export shim with legacy aliases (DomainException, InvalidTransitionError, AuthorizationError) - Update error_handler.py to import from domain/errors.py and map all new error types - Update main.py to register DomainError (new base) as the exception handler root --- backend/app/domain/errors.py | 96 +++++++++++++++++++++++++ backend/app/domain/exceptions.py | 81 +++++---------------- backend/app/main.py | 4 +- backend/app/middleware/error_handler.py | 26 +++---- 4 files changed, 130 insertions(+), 77 deletions(-) create mode 100644 backend/app/domain/errors.py diff --git a/backend/app/domain/errors.py b/backend/app/domain/errors.py new file mode 100644 index 0000000..e23f0d5 --- /dev/null +++ b/backend/app/domain/errors.py @@ -0,0 +1,96 @@ +"""Canonical domain error hierarchy for Aegis. + +Every service-layer error should be a subclass of :class:`DomainError`. +The global exception handler in ``app.middleware.error_handler`` maps +each concrete subclass to an appropriate HTTP status code so that +services never depend on FastAPI. + +Existing code that imports from ``app.domain.exceptions`` continues to +work — that module re-exports everything defined here. +""" + +from __future__ import annotations + + +class DomainError(Exception): + """Base for all domain errors.""" + + def __init__(self, message: str, *, code: str = "DOMAIN_ERROR") -> None: + self.message = message + self.code = code + super().__init__(message) + + +# ── Entity lifecycle ────────────────────────────────────────────────── + + +class EntityNotFoundError(DomainError): + """A requested entity does not exist.""" + + def __init__(self, entity: str, identifier: str) -> None: + super().__init__(f"{entity} not found: {identifier}", code="NOT_FOUND") + self.entity = entity + self.identifier = identifier + + +class DuplicateEntityError(DomainError): + """Creating an entity that already exists.""" + + def __init__(self, entity: str, field: str, value: str) -> None: + super().__init__( + f"{entity} with {field}='{value}' already exists", + code="DUPLICATE", + ) + + +# ── State machine ──────────────────────────────────────────────────── + + +class InvalidStateTransition(DomainError): + """A state-machine transition is not allowed.""" + + def __init__( + self, + current_state: str, + target_state: str, + valid_transitions: list[str] | None = None, + ) -> None: + msg = f"Cannot transition from '{current_state}' to '{target_state}'" + if valid_transitions: + msg += f". Valid transitions: {valid_transitions}" + super().__init__(msg, code="INVALID_TRANSITION") + self.current_state = current_state + self.target_state = target_state + self.valid_transitions = valid_transitions or [] + + +# ── Business rules ──────────────────────────────────────────────────── + + +class BusinessRuleViolation(DomainError): + """An operation violates a business invariant.""" + + def __init__(self, message: str) -> None: + super().__init__(message, code="BUSINESS_RULE_VIOLATION") + + +class InvalidOperationError(BusinessRuleViolation): + """An operation is invalid in the current context. + + Kept for backward compatibility; new code should prefer + :class:`BusinessRuleViolation` directly. + """ + + def __init__(self, message: str) -> None: + super().__init__(message) + self.code = "INVALID_OPERATION" + + +# ── Authorization ──────────────────────────────────────────────────── + + +class PermissionViolation(DomainError): + """The user lacks permissions for an action.""" + + def __init__(self, message: str = "Insufficient permissions") -> None: + super().__init__(message, code="FORBIDDEN") diff --git a/backend/app/domain/exceptions.py b/backend/app/domain/exceptions.py index f37c04a..7564750 100644 --- a/backend/app/domain/exceptions.py +++ b/backend/app/domain/exceptions.py @@ -1,67 +1,22 @@ -"""Domain exceptions for Aegis business logic. +"""Backward-compatible re-exports from :mod:`app.domain.errors`. -These exceptions are raised by service-layer code and automatically -mapped to HTTP responses by the error-handler middleware registered -in ``app.main``. This keeps the service layer free from any HTTP -or framework coupling. +All domain errors now live in ``errors.py``. This module preserves the +old import paths so that existing code keeps working without changes:: + + from app.domain.exceptions import InvalidTransitionError # still works """ +from app.domain.errors import ( # noqa: F401 + BusinessRuleViolation, + DomainError, + DuplicateEntityError, + EntityNotFoundError, + InvalidOperationError, + InvalidStateTransition, + PermissionViolation, +) -class DomainException(Exception): - """Base for all domain exceptions.""" - - def __init__(self, message: str, code: str = "DOMAIN_ERROR"): - self.message = message - self.code = code - super().__init__(message) - - -class EntityNotFoundError(DomainException): - """Raised when a requested entity does not exist.""" - - def __init__(self, entity: str, identifier: str): - super().__init__(f"{entity} not found: {identifier}", "NOT_FOUND") - self.entity = entity - self.identifier = identifier - - -class DuplicateEntityError(DomainException): - """Raised when creating an entity that already exists.""" - - def __init__(self, entity: str, field: str, value: str): - super().__init__( - f"{entity} with {field}='{value}' already exists", - "DUPLICATE", - ) - - -class InvalidTransitionError(DomainException): - """Raised when a state-machine transition is not allowed.""" - - def __init__( - self, - current_state: str, - target_state: str, - valid_transitions: list[str] | None = None, - ): - msg = f"Cannot transition from '{current_state}' to '{target_state}'" - if valid_transitions: - msg += f". Valid transitions: {valid_transitions}" - super().__init__(msg, "INVALID_TRANSITION") - self.current_state = current_state - self.target_state = target_state - self.valid_transitions = valid_transitions or [] - - -class InvalidOperationError(DomainException): - """Raised when an operation is invalid in the current context.""" - - def __init__(self, message: str): - super().__init__(message, "INVALID_OPERATION") - - -class AuthorizationError(DomainException): - """Raised when the user lacks permissions for an action.""" - - def __init__(self, message: str = "Insufficient permissions"): - super().__init__(message, "FORBIDDEN") +# Legacy aliases — old name → new name +DomainException = DomainError +InvalidTransitionError = InvalidStateTransition +AuthorizationError = PermissionViolation diff --git a/backend/app/main.py b/backend/app/main.py index 81685ed..65ad950 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -38,7 +38,7 @@ from app.routers import professional_reports as professional_reports_router from app.routers import analytics as analytics_router from app.routers import advanced_metrics as advanced_metrics_router from app.routers import osint as osint_router -from app.domain.exceptions import DomainException +from app.domain.errors import DomainError from app.middleware.error_handler import domain_exception_handler from app.storage import ensure_bucket_exists from app.jobs.mitre_sync_job import start_scheduler, scheduler @@ -77,7 +77,7 @@ app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # ── Domain exception → HTTP mapping ────────────────────────────────────── -app.add_exception_handler(DomainException, domain_exception_handler) +app.add_exception_handler(DomainError, domain_exception_handler) # ── CORS ────────────────────────────────────────────────────────────────── from app.config import settings as _settings diff --git a/backend/app/middleware/error_handler.py b/backend/app/middleware/error_handler.py index ecd62d6..b815da9 100644 --- a/backend/app/middleware/error_handler.py +++ b/backend/app/middleware/error_handler.py @@ -1,41 +1,43 @@ -"""Domain exception → HTTP response mapping. +"""Domain error → HTTP response mapping. This module provides a single exception handler that converts -domain-layer exceptions into structured JSON responses, keeping +domain-layer errors into structured JSON responses, keeping the service layer free from FastAPI's ``HTTPException``. """ from fastapi import Request from fastapi.responses import JSONResponse -from app.domain.exceptions import ( - AuthorizationError, - DomainException, +from app.domain.errors import ( + BusinessRuleViolation, + DomainError, DuplicateEntityError, EntityNotFoundError, InvalidOperationError, - InvalidTransitionError, + InvalidStateTransition, + PermissionViolation, ) -EXCEPTION_STATUS_MAP: dict[type[DomainException], int] = { +EXCEPTION_STATUS_MAP: dict[type[DomainError], int] = { EntityNotFoundError: 404, DuplicateEntityError: 409, - InvalidTransitionError: 400, + InvalidStateTransition: 400, InvalidOperationError: 400, - AuthorizationError: 403, + BusinessRuleViolation: 400, + PermissionViolation: 403, } async def domain_exception_handler( request: Request, - exc: DomainException, + exc: DomainError, ) -> JSONResponse: - """Convert a :class:`DomainException` into a JSON error response.""" + """Convert a :class:`DomainError` into a JSON error response.""" status_code = EXCEPTION_STATUS_MAP.get(type(exc), 400) content: dict = {"detail": exc.message, "code": exc.code} - if isinstance(exc, InvalidTransitionError): + if isinstance(exc, InvalidStateTransition): content["current_state"] = exc.current_state content["target_state"] = exc.target_state content["valid_transitions"] = exc.valid_transitions