Antoni Nuñez Romeu
78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
from datetime import datetime, timezone
|
|
|
|
|
|
class PIPException(Exception):
|
|
def __init__(self, message: str, code: str = "PIP_ERROR", status_code: int = 500, details: dict | None = None):
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.code = code
|
|
self.status_code = status_code
|
|
self.details = details or {}
|
|
self.timestamp = datetime.now(timezone.utc).isoformat()
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"error": {
|
|
"code": self.code,
|
|
"message": self.message,
|
|
"details": self.details,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
}
|
|
|
|
|
|
class NotFoundException(PIPException):
|
|
def __init__(self, resource: str, resource_id: str | None = None):
|
|
msg = f"{resource} not found"
|
|
if resource_id:
|
|
msg = f"{resource} '{resource_id}' not found"
|
|
super().__init__(message=msg, code="NOT_FOUND", status_code=404, details={"resource": resource, "resource_id": resource_id})
|
|
|
|
|
|
class ConflictException(PIPException):
|
|
def __init__(self, message: str, details: dict | None = None):
|
|
super().__init__(message=message, code="CONFLICT", status_code=409, details=details)
|
|
|
|
|
|
class ValidationException(PIPException):
|
|
def __init__(self, message: str, details: dict | None = None):
|
|
super().__init__(message=message, code="VALIDATION_ERROR", status_code=422, details=details)
|
|
|
|
|
|
class UnauthorizedException(PIPException):
|
|
def __init__(self, message: str = "Authentication required"):
|
|
super().__init__(message=message, code="UNAUTHORIZED", status_code=401)
|
|
|
|
|
|
class ForbiddenException(PIPException):
|
|
def __init__(self, message: str = "Insufficient permissions", required_role: str | None = None):
|
|
details = {}
|
|
if required_role:
|
|
details["required_role"] = required_role
|
|
super().__init__(message=message, code="FORBIDDEN", status_code=403, details=details)
|
|
|
|
|
|
class RateLimitException(PIPException):
|
|
def __init__(self, message: str = "Rate limit exceeded"):
|
|
super().__init__(message=message, code="RATE_LIMIT_EXCEEDED", status_code=429)
|
|
|
|
|
|
class ServiceUnavailableException(PIPException):
|
|
def __init__(self, service: str, message: str | None = None):
|
|
msg = message or f"Service '{service}' is unavailable"
|
|
super().__init__(message=msg, code="SERVICE_UNAVAILABLE", status_code=503, details={"service": service})
|
|
|
|
|
|
class ConnectorException(PIPException):
|
|
def __init__(self, connector_type: str, message: str, details: dict | None = None):
|
|
d = details or {}
|
|
d["connector_type"] = connector_type
|
|
super().__init__(message=message, code="CONNECTOR_ERROR", status_code=502, details=d)
|
|
|
|
|
|
class SyncException(PIPException):
|
|
def __init__(self, pharmacy_id: str, message: str, details: dict | None = None):
|
|
d = details or {}
|
|
d["pharmacy_id"] = pharmacy_id
|
|
super().__init__(message=message, code="SYNC_ERROR", status_code=502, details=d)
|