"""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. """ # Enable future language features for compatibility from __future__ import annotations # Define class DomainError class DomainError(Exception): """Base for all domain errors.""" # Define function __init__ def __init__(self, message: str, *, code: str = "DOMAIN_ERROR") -> None: """Initialise the domain error with a human-readable message and error code. Args: message (str): Human-readable description of the error. code (str): Machine-readable error code used by the HTTP error handler. Returns: None """ # Assign self.message = message self.message = message # Assign self.code = code self.code = code # Call super() super().__init__(message) # ── Entity lifecycle ────────────────────────────────────────────────── class EntityNotFoundError(DomainError): """A requested entity does not exist.""" # Define function __init__ def __init__(self, entity: str, identifier: str) -> None: """Initialise an entity-not-found error. Args: entity (str): Name of the entity type that was not found (e.g. "Technique"). identifier (str): The ID or key used in the failed lookup. Returns: None """ # Call super() super().__init__(f"{entity} not found: {identifier}", code="NOT_FOUND") # Assign self.entity = entity self.entity = entity # Assign self.identifier = identifier self.identifier = identifier # Define class DuplicateEntityError class DuplicateEntityError(DomainError): """Creating an entity that already exists.""" # Define function __init__ def __init__(self, entity: str, field: str, value: str) -> None: """Initialise a duplicate-entity error. Args: entity (str): Name of the entity type that already exists (e.g. "Campaign"). field (str): Name of the field whose value conflicts (e.g. "name"). value (str): The conflicting value that is already in use. Returns: None """ # Call super() super().__init__( f"{entity} with {field}='{value}' already exists", # Keyword argument: code code="DUPLICATE", ) # ── State machine ──────────────────────────────────────────────────── class InvalidStateTransition(DomainError): # noqa: N818 — DDD term, renaming would break 96 call sites """A state-machine transition is not allowed.""" # Define function __init__ def __init__( self, # Entry: current_state current_state: str, # Entry: target_state target_state: str, # Entry: valid_transitions valid_transitions: list[str] | None = None, ) -> None: """Initialise an invalid state-transition error. Args: current_state (str): The entity's present state (e.g. "draft"). target_state (str): The state that was illegally requested. valid_transitions (list[str] | None): Allowed target states from the current state; included in the error message when provided. Returns: None """ # Assign msg = f"Cannot transition from '{current_state}' to '{target_state}'" msg = f"Cannot transition from '{current_state}' to '{target_state}'" # Check: valid_transitions if valid_transitions: # Assign msg = f". Valid transitions: {valid_transitions}" msg += f". Valid transitions: {valid_transitions}" # Call super() super().__init__(msg, code="INVALID_TRANSITION") # Assign self.current_state = current_state self.current_state = current_state # Assign self.target_state = target_state self.target_state = target_state # Assign self.valid_transitions = valid_transitions or [] self.valid_transitions = valid_transitions or [] # ── Business rules ──────────────────────────────────────────────────── class BusinessRuleViolation(DomainError): # noqa: N818 — DDD term, renaming would break 96 call sites """An operation violates a business invariant.""" # Define function __init__ def __init__(self, message: str) -> None: """Initialise a business-rule violation error. Args: message (str): Human-readable description of the violated rule. Returns: None """ # Call super() super().__init__(message, code="BUSINESS_RULE_VIOLATION") # Define class InvalidOperationError class InvalidOperationError(BusinessRuleViolation): """An operation is invalid in the current context. Kept for backward compatibility; new code should prefer :class:`BusinessRuleViolation` directly. """ # Define function __init__ def __init__(self, message: str) -> None: """Initialise an invalid-operation error. Args: message (str): Human-readable description of why the operation is invalid. Returns: None """ # Call super() super().__init__(message) # Assign self.code = "INVALID_OPERATION" self.code = "INVALID_OPERATION" # ── Authorization ──────────────────────────────────────────────────── class PermissionViolation(DomainError): # noqa: N818 — DDD term, renaming would break 96 call sites """The user lacks permissions for an action.""" # Define function __init__ def __init__(self, message: str = "Insufficient permissions") -> None: """Initialise a permission-violation error. Args: message (str): Human-readable description of the access denial. Returns: None """ # Call super() super().__init__(message, code="FORBIDDEN")