From 7778135aa72a2730518424f5f8306efb82194666 Mon Sep 17 00:00:00 2001 From: David Perl Date: Fri, 24 Jul 2026 16:01:58 +0200 Subject: [PATCH] fix: tidy exception classes --- src/aarecommon/errors/exception_handler.py | 247 ++++----------------- tests/test_aare_exception.py | 6 +- tests/test_error_codes.py | 4 +- tests/test_exception_handler.py | 6 +- 4 files changed, 53 insertions(+), 210 deletions(-) diff --git a/src/aarecommon/errors/exception_handler.py b/src/aarecommon/errors/exception_handler.py index 27190e6..0ef617f 100644 --- a/src/aarecommon/errors/exception_handler.py +++ b/src/aarecommon/errors/exception_handler.py @@ -1,6 +1,5 @@ from __future__ import annotations -import time from typing import ClassVar from aarecommon.errors.codes import AuthErrorCode @@ -10,19 +9,18 @@ logger = setup_logger("aareDAQ") class AareException(Exception): - """Root of the aare exception hierarchy. + """Root of the aare exception hierarchy.""" - Class-level ``critical`` is the default; pass ``critical=...`` to the - constructor of an "optionally critical" subclass to override on a single - raise site. - """ + DEFAULT_CRITICALITY: ClassVar[bool] = False + DEFAULT_MESSAGE: ClassVar[str] = "Base Aare Exception" - critical: ClassVar[bool] = False + def __init__(self, message: str | None = None, *, critical: bool | None = None, **kwargs): + super().__init__(message, **kwargs) + self.critical = critical if critical is not None else self.__class__.DEFAULT_CRITICALITY + self.message = message if message is not None else self.__class__.DEFAULT_MESSAGE - def __init__(self, *args, critical: bool | None = None, **kwargs): - super().__init__(*args, **kwargs) - if critical is not None: - self.critical = critical + def __str__(self) -> str: + return self.message class AutomationError(AareException): @@ -76,62 +74,31 @@ class BeamlineStateException(AutomationError): class TransformationInvalidException(AutomationError): - critical: ClassVar[bool] = True - - def __init__( - self, message: str = "Transformation is not implemented", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Transformation is not implemented" class StateTransitionFailed(BeamlineStateException): - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True - def __init__( - self, message: str = "Beamline state transition failed", *, critical: bool | None = None - ): + def __init__(self, message: str = "Beamline state transition failed", *, critical: bool = True): super().__init__(message, critical=critical) self.message = message - def __str__(self) -> str: - return self.message - class MaintenanceStateException(BeamlineStateException): - critical: ClassVar[bool] = True - - def __init__( - self, message: str = "Beamline is in Maintenance state", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Beamline is in Maintenance state" class DataCollectionException(AutomationError): """Group parent for data-collection-time exceptions; also raisable directly.""" - def __init__(self, message: str = "Data collection failed", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Data collection failed" class RasterScanException(DataCollectionException): - def __init__(self, message: str = "Data collection failed", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Data collection failed" class AutoRasterSampleSkipped(AutomationError): @@ -139,65 +106,32 @@ class AutoRasterSampleSkipped(AutomationError): class LoopCenteringFailed(AutomationError): - def __init__( - self, - message: str = "Loop Centering did not detect a sample", - *, - critical: bool | None = None, - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Loop Centering did not detect a sample" class UnmountingFailed(TellException): """Sample failed to unmount. Optionally critical via instance flag.""" - def __init__( - self, message: str = "A sample was not unmounted", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "A sample was not unmounted" class MountingFailed(TellException): """Sample failed to mount. Optionally critical via instance flag.""" - def __init__(self, message: str = "A sample was not mounted", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "A sample was not mounted" class ManualMountException(AareUserError): """Custom exception for manual mounting""" - def __init__(self, message: str = "Manual mounting failed", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Manual mounting failed" class SmartMagnetFaultException(AutomationError): """Custom exception for smart magnet fault""" - critical: ClassVar[bool] = True - - def __init__(self, message: str = "Smart magnet fault", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Smart magnet fault" class DoorSafetyError(AutomationError): @@ -209,98 +143,44 @@ class DoorSafetyError(AutomationError): critical so automation halts and the GUI shows a pop-up. """ - critical: ClassVar[bool] = True - - def __init__( - self, message: str = "Door safety could not be activated", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Door safety could not be activated" class TellCommandWhileBusyException(TellException): """Custom exception for trying to move Tell when it is busy""" - def __init__(self, message: str = "Tell is busy", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Tell is busy" class TellConnectionException(TellException): """Custom exception for connection problems""" - critical: ClassVar[bool] = True - - def __init__(self, message: str = "Lost connection to Tell", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Lost connection to Tell" class WarningTellException(TellException): - def __init__(self, message: str = "Warning error in TELL", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Warning error in TELL" class CriticalTellException(TellException): - critical: ClassVar[bool] = True - - def __init__(self, message: str = "Critical error in TELL", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Critical error in TELL" class AXCFailed(AutomationError): - def __init__( - self, message: str = "Auto X-ray centering failed", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Auto X-ray centering failed" class BeamlineBusyException(BeamlineStateException): - critical: ClassVar[bool] = True - - def __init__(self, message: str = "Beamline is in busy state", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Beamline is in busy state" class BeamlineBusyTimeoutException(BeamlineStateException): - critical: ClassVar[bool] = True - - def __init__( - self, - message: str = "Beamline busy state expired during operation", - *, - critical: bool | None = None, - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Beamline busy state expired during operation" class SampleException(AareUserError): @@ -310,12 +190,7 @@ class SampleException(AareUserError): AareUserError pending confirmation -- "sample not found" reads as bad input rather than a runtime failure.""" - def __init__(self, message: str = "Sample not found", *, critical: bool | None = None): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message + DEFAULT_MESSAGE = "Sample not found" # --------------------------------------------------------------------------- @@ -339,9 +214,6 @@ class AuthenticationException(AareAuthError): self.headers = headers self.code = code - def __str__(self) -> str: - return self.message - class UserRightsException(AareAuthError): _last_log_ts_by_message: dict[str, float] = {} @@ -362,9 +234,6 @@ class UserRightsException(AareAuthError): self.headers = headers self.code = code - def __str__(self) -> str: - return self.message - # --------------------------------------------------------------------------- # Device communication exceptions @@ -377,7 +246,7 @@ class SmargonCommunicationError(SmargonException): Keep the original exception in `__cause__` by using `raise ... from e`. """ - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True def __init__( self, @@ -396,9 +265,6 @@ class SmargonCommunicationError(SmargonException): self.operation = operation self.status_code = status_code - def __str__(self) -> str: - return self.message - class TellCommunicationError(TellException): """ @@ -406,7 +272,7 @@ class TellCommunicationError(TellException): Intended to be caught centrally by FastAPI exception handlers. """ - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True def __init__( self, @@ -423,9 +289,6 @@ class TellCommunicationError(TellException): self.base_url = base_url self.operation = operation - def __str__(self) -> str: - return self.message - class JFJochCommunicationError(JFJochException): """ @@ -433,7 +296,7 @@ class JFJochCommunicationError(JFJochException): Intended for scan-time fallbacks and GUI-visible alerts. """ - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True def __init__( self, @@ -452,9 +315,6 @@ class JFJochCommunicationError(JFJochException): self.base_url = base_url self.status_code = status_code - def __str__(self) -> str: - return self.message - class AareDBCommunicationError(AutomationError): """Raised when AareDB HTTPS communication fails. @@ -479,9 +339,6 @@ class AareDBCommunicationError(AutomationError): self.base_url = base_url self.status_code = status_code - def __str__(self) -> str: - return self.message - class AerotechCommunicationError(AerotechException): """ @@ -489,7 +346,7 @@ class AerotechCommunicationError(AerotechException): Keep the original exception in `__cause__` by using `raise ... from e`. """ - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True def __init__( self, @@ -508,25 +365,14 @@ class AerotechCommunicationError(AerotechException): self.operation = operation self.status_code = status_code - def __str__(self) -> str: - return self.message - -class MagnetPositionSensorErorr(AutomationError): - critical: ClassVar[bool] = True - - def __init__( - self, message: str = "Magnet position sensor error", *, critical: bool | None = None - ): - super().__init__(message, critical=critical) - self.message = message - - def __str__(self) -> str: - return self.message +class MagnetPositionSensorError(AutomationError): + DEFAULT_CRITICALITY = True + DEFAULT_MESSAGE = "Magnet position sensor error" class BECCommunicationError(BECException): - critical: ClassVar[bool] = True + DEFAULT_CRITICALITY = True def __init__( self, @@ -544,6 +390,3 @@ class BECCommunicationError(BECException): self.exception = exception self.endpoint = endpoint self.base_url = base_url - - def __str__(self) -> str: - return self.message diff --git a/tests/test_aare_exception.py b/tests/test_aare_exception.py index c218a00..4b775eb 100644 --- a/tests/test_aare_exception.py +++ b/tests/test_aare_exception.py @@ -31,7 +31,7 @@ from aarecommon.errors.exception_handler import ( JFJochCommunicationError, JFJochException, LoopCenteringFailed, - MagnetPositionSensorErorr, + MagnetPositionSensorError, MaintenanceStateException, ManualMountException, MountingFailed, @@ -73,7 +73,7 @@ def test_phase3_classes_have_critical_default_true(): MaintenanceStateException, BeamlineBusyException, BeamlineBusyTimeoutException, - MagnetPositionSensorErorr, + MagnetPositionSensorError, SmartMagnetFaultException, TransformationInvalidException, ): @@ -149,7 +149,7 @@ def test_automation_errors_are_automation_error(): AareDBCommunicationError, BeamlineBusyException, RasterScanException, - MagnetPositionSensorErorr, + MagnetPositionSensorError, SmartMagnetFaultException, TransformationInvalidException, AutoRasterSampleSkipped, diff --git a/tests/test_error_codes.py b/tests/test_error_codes.py index 1c7213a..17b76ae 100644 --- a/tests/test_error_codes.py +++ b/tests/test_error_codes.py @@ -52,7 +52,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum(): DataCollectionException, JFJochCommunicationError, LoopCenteringFailed, - MagnetPositionSensorErorr, + MagnetPositionSensorError, MaintenanceStateException, ManualMountException, MountingFailed, @@ -94,7 +94,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum(): AXCFailed, AutoRasterSampleSkipped, TransformationInvalidException, - MagnetPositionSensorErorr, + MagnetPositionSensorError, SmartMagnetFaultException, ManualMountException, SampleException, diff --git a/tests/test_exception_handler.py b/tests/test_exception_handler.py index eda074d..266af8d 100644 --- a/tests/test_exception_handler.py +++ b/tests/test_exception_handler.py @@ -10,7 +10,7 @@ from aarecommon.errors.exception_handler import ( DataCollectionException, JFJochCommunicationError, LoopCenteringFailed, - MagnetPositionSensorErorr, + MagnetPositionSensorError, ManualMountException, MountingFailed, RasterScanException, @@ -152,7 +152,7 @@ def test_aerotech_communication_error(): def test_magnet_position_sensor_error(): - exc = MagnetPositionSensorErorr() + exc = MagnetPositionSensorError() assert "Magnet position sensor error" in str(exc) @@ -178,7 +178,7 @@ def test_magnet_position_sensor_error(): lambda: JFJochCommunicationError("jce", operation="POST"), lambda: AareDBCommunicationError("dbce"), lambda: AerotechCommunicationError("ace"), - lambda: MagnetPositionSensorErorr(), + lambda: MagnetPositionSensorError(), ], ) def test_exception_construction_emits_no_logs(factory, caplog): -- 2.54.0