File size: 5,152 Bytes
e391a84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """
domain/exceptions/domain_exceptions.py
───────────────────────────────────────
Domain-specific exception hierarchy.
All exceptions here represent business-rule violations — they are completely
framework-agnostic (no HTTP codes, no FastAPI imports).
Hierarchy:
DomainException (base)
├── InvalidSignalError — PPG data fails domain validation → 400
├── PredictionOutOfRangeError — Predicted BP is physiologically impossible → 422
├── EntityNotFoundError — Entity lookup returned nothing → 404
├── ConflictError — Duplicate/unique constraint violation → 409
└── DatabaseError — DB unavailable or unrecoverable query err → 503
"""
from __future__ import annotations
class DomainException(Exception):
"""
Base class for all domain-layer exceptions.
Catch this if you want to handle any business-rule violation without
knowing the specific type.
"""
def __init__(self, message: str, context: dict | None = None) -> None:
super().__init__(message)
self.message = message
self.context: dict = context or {}
def __repr__(self) -> str:
return f"{self.__class__.__name__}(message={self.message!r}, context={self.context})"
class InvalidSignalError(DomainException):
"""
Raised when a PPG signal fails domain validation.
Examples:
- Sampling rate is outside the accepted range
- Signal is too short / too long
- ``ppg_values`` list is empty
- ``duration_seconds`` does not match the number of samples
HTTP Status: 400 Bad Request
"""
def __init__(self, field: str, value: object, reason: str) -> None:
message = f"Invalid PPG signal — field '{field}': {reason} (got {value!r})"
super().__init__(message, context={"field": field, "value": value, "reason": reason})
self.field = field
self.value = value
self.reason = reason
class PredictionOutOfRangeError(DomainException):
"""
Raised when a model's prediction is physiologically implausible.
Examples:
- Predicted SBP < 60 mmHg or > 260 mmHg
- Predicted DBP < 30 mmHg or > 160 mmHg
- SBP < DBP (pulse pressure is negative)
HTTP Status: 422 Unprocessable Entity
"""
def __init__(self, predicted_sbp: float, predicted_dbp: float, reason: str) -> None:
message = (
f"Prediction out of physiological range — "
f"SBP={predicted_sbp} mmHg, DBP={predicted_dbp} mmHg: {reason}"
)
super().__init__(
message,
context={
"predicted_sbp": predicted_sbp,
"predicted_dbp": predicted_dbp,
"reason": reason,
},
)
self.predicted_sbp = predicted_sbp
self.predicted_dbp = predicted_dbp
self.reason = reason
class EntityNotFoundError(DomainException):
"""
Raised when a repository lookup fails to find the requested entity.
HTTP Status: 404 Not Found
Args:
entity_type: Class name of the entity (e.g. ``"PPGSignal"``).
entity_id: The ID that was looked up.
"""
def __init__(self, entity_type: str, entity_id: str) -> None:
message = f"{entity_type} with id='{entity_id}' not found."
super().__init__(message, context={"entity_type": entity_type, "entity_id": entity_id})
self.entity_type = entity_type
self.entity_id = entity_id
class ConflictError(DomainException):
"""
Raised when an insert/update violates a unique constraint in the database.
Typical causes:
- Duplicate PPG signal ID (re-submission of the same record)
- Unique index violation on (user_id, timestamp) composite key
HTTP Status: 409 Conflict
Args:
entity_type: Class name of the conflicting entity (e.g. ``"PPGSignal"``).
detail: Human-readable description of the conflict.
"""
def __init__(self, entity_type: str, detail: str) -> None:
message = f"Conflict on {entity_type}: {detail}"
super().__init__(message, context={"entity_type": entity_type, "detail": detail})
self.entity_type = entity_type
self.detail = detail
class DatabaseError(DomainException):
"""
Raised when the database is unreachable or an unrecoverable query error occurs.
Wraps SQLAlchemy/asyncpg low-level errors so that domain and application
layers never import SQLAlchemy directly.
HTTP Status: 503 Service Unavailable
Args:
operation: The repository operation that failed (e.g. ``"add"``, ``"get_by_id"``).
reason: Human-readable description of what went wrong.
"""
def __init__(self, operation: str, reason: str) -> None:
message = f"Database error during '{operation}': {reason}"
super().__init__(message, context={"operation": operation, "reason": reason})
self.operation = operation
self.reason = reason
|