custom exceptions

Author: pfouque

django_fsm/__init__.py

@@ -17,6 +17,12 @@
 from django.db.models.query_utils import DeferredAttribute
 from django.db.models.signals import class_prepared
 
+from .exceptions import ConcurrentTransition
+from .exceptions import InvalidResultState
+from .exceptions import InvalidTransition
+from .exceptions import NoTransition
+from .exceptions import TransitionConditionsUnmet
+from .exceptions import TransitionNotAllowed
 from .signals import post_transition
 from .signals import pre_transition
 
@@ -64,6 +70,9 @@
     "FSMIntegerField",
     "FSMKeyField",
     "FSMModelMixin",
+    "InvalidTransition",
+    "NoTransition",
+    "TransitionConditionsUnmet",
     "TransitionNotAllowed",
     "can_proceed",
     "has_transition_perm",
@@ -75,29 +84,6 @@
 ANY_OTHER_STATE = "+"
 
 
-class TransitionNotAllowed(Exception):  # noqa: N818
-    """Raised when a transition is not allowed"""
-
-    @override
-    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
-        self.object = kwargs.pop("object", None)
-        self.method = kwargs.pop("method", None)
-        self.failed_condition = kwargs.pop("failed_condition", None)
-        super().__init__(*args, **kwargs)
-
-
-class InvalidResultState(Exception):  # noqa: N818
-    """Raised when we got invalid result state"""
-
-
-class ConcurrentTransition(Exception):  # noqa: N818
-    """
-    Raised when the transition cannot be executed because the
-    object has become stale (state has been changed since it
-    was fetched from the database).
-    """
-
-
 class Transition:
     def __init__(
         self,
@@ -148,7 +134,8 @@ def __eq__(self, other: object) -> bool:
 
 
 def get_available_FIELD_transitions(  # noqa: N802
-    instance: _FSMModel, field: FSMFieldMixin
+    instance: _FSMModel,
+    field: FSMFieldMixin,
 ) -> typing.Generator[Transition]:
     """
     List of transitions available in current model state
@@ -164,7 +151,8 @@ def get_available_FIELD_transitions(  # noqa: N802
 
 
 def get_all_FIELD_transitions(  # noqa: N802
-    instance: _FSMModel, field: FSMFieldMixin
+    instance: _FSMModel,
+    field: FSMFieldMixin,
 ) -> typing.Generator[Transition]:
     """
     List of all transitions available in current model state
@@ -173,7 +161,9 @@ def get_all_FIELD_transitions(  # noqa: N802
 
 
 def get_available_user_FIELD_transitions(  # noqa: N802
-    instance: _FSMModel, user: UserWithPermissions, field: FSMFieldMixin
+    instance: _FSMModel,
+    user: UserWithPermissions,
+    field: FSMFieldMixin,
 ) -> typing.Generator[Transition]:
     """
     List of transitions available in current model state
@@ -256,7 +246,7 @@ def conditions_met(self, instance: _FSMModel, state: _StateValue) -> bool:
 
         return all(condition(instance) for condition in transition.conditions)
 
-    def _get_first_unmet_condition(
+    def get_first_unmet_condition(
         self, instance: _FSMModel, state: _StateValue
     ) -> _Condition | None:
         """
@@ -290,15 +280,15 @@ def next_state(self, current_state: _StateValue) -> _StateValue:
         transition = self.get_transition(current_state)
 
         if transition is None:
-            raise TransitionNotAllowed(f"No transition from {current_state}")
+            raise NoTransition(f"No transition from {current_state}")
 
         return transition.target
 
     def exception_state(self, current_state: _StateValue) -> _StateValue | None:
         transition = self.get_transition(current_state)
 
         if transition is None:
-            raise TransitionNotAllowed(f"No transition from {current_state}")
+            raise NoTransition(f"No transition from {current_state}")
 
         return transition.on_error
 
@@ -390,14 +380,14 @@ def change_state(
         current_state = self.get_state(instance)
 
         if not meta.has_transition(current_state):
-            raise TransitionNotAllowed(
+            raise InvalidTransition(
                 f"Can't switch from state '{current_state}' using method '{method_name}'",
                 object=instance,
                 method=method,
             )
-        unmet = meta._get_first_unmet_condition(instance, current_state)
+        unmet = meta.get_first_unmet_condition(instance, current_state)
         if unmet is not None:
-            raise TransitionNotAllowed(
+            raise TransitionConditionsUnmet(
                 f"Transition conditions have not been met for method '{method_name}': "
                 f"{getattr(unmet, '__name__', repr(unmet))}",
                 object=instance,

django_fsm/exceptions.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import typing
+
+try:
+    from typing import override
+except ImportError:  # pragma: no cover
+    # Py<3.12
+    from typing_extensions import override
+
+if typing.TYPE_CHECKING:  # pragma: no cover
+    from . import _Condition
+    from . import _FSMModel
+    from . import _TransitionFunc
+
+
+class FSMException(Exception):  # noqa: N818
+    ...
+
+
+class TransitionNotAllowed(FSMException):
+    """Raised when a transition is not allowed"""
+
+    object: _FSMModel
+    method: _TransitionFunc
+
+    @override
+    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.object = kwargs.pop("object", None)
+        self.method = kwargs.pop("method", None)
+        super().__init__(*args, **kwargs)
+
+
+class NoTransition(TransitionNotAllowed):
+    """Raised when no transition exists for the current state"""
+
+
+class InvalidTransition(TransitionNotAllowed):
+    """Raised when a transition method is not valid for the current state"""
+
+
+class TransitionConditionsUnmet(TransitionNotAllowed):
+    """Raised when a transition condition fails"""
+
+    failed_condition: _Condition
+
+    @override
+    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.failed_condition = kwargs.pop("failed_condition", None)
+        super().__init__(*args, **kwargs)
+
+
+class InvalidResultState(FSMException):
+    """Raised when we got invalid result state"""
+
+
+class ConcurrentTransition(FSMException):
+    """
+    Raised when the transition cannot be executed because the
+    object has become stale (state has been changed since it
+    was fetched from the database).
+    """

tests/testapp/tests/test_basic_transitions.py

@@ -77,7 +77,7 @@ def test_known_transitions_succeed(self):
 
     def test_unavailable_transition_fails(self):
         assert not fsm.can_proceed(self.model.hide)
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.hide()
 
     def test_state_unchanged_when_transition_raises(self):
@@ -94,7 +94,7 @@ def test_available_transition_with_empty_target_keeps_state(self):
         assert self.model.state == ApplicationState.PUBLISHED
 
     def test_unavailable_transition_with_empty_target_keeps_state(self):
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.notify_all()
 
         assert self.model.state == ApplicationState.NEW
@@ -134,7 +134,7 @@ def test_any_other_state_transition_fails_when_same_source(self):
         self.model.block()
 
         assert not fsm.can_proceed(self.model.block)
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.block()
 
     def test_empty_string_target_sets_blank_state(self):
@@ -170,7 +170,7 @@ def test_signals_fire_on_valid_transition(self):
         assert self.post_transition_called
 
     def test_signals_do_not_fire_on_invalid_transition(self):
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.hide()
 
         assert not self.pre_transition_called

tests/testapp/tests/test_conditions.py

@@ -62,27 +62,26 @@ def test_unmet_condition(self):
 
         assert not fsm.can_proceed(self.model.remove)
 
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.TransitionConditionsUnmet):
             self.model.remove()
 
         assert fsm.can_proceed(self.model.remove, check_conditions=False)
 
     def test_failed_condition_reported_on_exception(self):
         self.model.publish()
-        with pytest.raises(fsm.TransitionNotAllowed) as exc_info:
+        with pytest.raises(fsm.TransitionConditionsUnmet) as exc_info:
             self.model.remove()
         assert exc_info.value.failed_condition is BlogPostWithConditions.unmet_condition
 
     def test_failed_condition_named_in_message(self):
         self.model.publish()
-        with pytest.raises(fsm.TransitionNotAllowed, match="unmet_condition"):
+        with pytest.raises(fsm.TransitionConditionsUnmet, match="unmet_condition"):
             self.model.remove()
 
     def test_failed_condition_is_none_when_no_condition_failure(self):
-        """TransitionNotAllowed for a missing transition has no failed_condition."""
-        with pytest.raises(fsm.TransitionNotAllowed) as exc_info:
+        """InvalidTransition for a missing transition has no failed_condition."""
+        with pytest.raises(fsm.InvalidTransition):
             self.model.remove()  # state is "new", destroy only works from "published"
-        assert exc_info.value.failed_condition is None
 
 
 def _eval_tracking_condition(instance: models.Model) -> bool:
@@ -114,7 +113,7 @@ class ShortCircuitTest(TestCase):
     def test_only_first_failing_condition_evaluated(self):
         obj = BlogPostShortCircuit()
         obj._eval_log = []
-        with pytest.raises(fsm.TransitionNotAllowed) as exc_info:
+        with pytest.raises(fsm.TransitionConditionsUnmet) as exc_info:
             obj.publish()
         assert exc_info.value.failed_condition is _eval_tracking_condition
         assert obj._eval_log == ["first"]

tests/testapp/tests/test_integer_field.py

@@ -35,5 +35,5 @@ def test_known_transition_should_succeed(self):
         assert self.model.state == BlogPostState.HIDDEN
 
     def test_unknown_transition_fails(self):
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.hide()

tests/testapp/tests/test_key_field.py

@@ -84,7 +84,7 @@ def test_known_transition_should_succeed(self):
     def test_unknown_transition_fails(self):
         assert not fsm.can_proceed(self.model.hide)
 
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.hide()
 
     def test_state_non_changed_after_fail(self):
@@ -104,7 +104,7 @@ def test_allowed_null_transition_should_succeed(self):
         assert self.model.state == "_PUBLISHED_"
 
     def test_unknown_null_transition_should_fail(self):
-        with pytest.raises(fsm.TransitionNotAllowed):
+        with pytest.raises(fsm.InvalidTransition):
             self.model.notify_all()
 
         assert self.model.state == "_NEW_"

tests/testapp/tests/test_permissions.py

@@ -11,39 +11,39 @@
 class PermissionFSMFieldTest(TestCase):
     def setUp(self):
         self.model = BlogPost()
-        self.unprivileged = User.objects.create(username="unprivileged")
-        self.privileged = User.objects.create(username="privileged")
-        self.staff = User.objects.create(username="staff", is_staff=True)
+        self.unprivileged_user = User.objects.create(username="unprivileged")
+        self.privileged_user = User.objects.create(username="privileged")
+        self.staff_user = User.objects.create(username="staff", is_staff=True)
 
-        self.privileged.user_permissions.add(
+        self.privileged_user.user_permissions.add(
             Permission.objects.get_by_natural_key("can_publish_post", "testapp", "blogpost")
         )
-        self.privileged.user_permissions.add(
+        self.privileged_user.user_permissions.add(
             Permission.objects.get_by_natural_key("can_remove_post", "testapp", "blogpost")
         )
 
     def test_privileged_access_succeed(self):
-        assert fsm.has_transition_perm(self.model.publish, self.privileged)
-        assert fsm.has_transition_perm(self.model.remove, self.privileged)
+        assert fsm.has_transition_perm(self.model.publish, self.privileged_user)
+        assert fsm.has_transition_perm(self.model.remove, self.privileged_user)
 
         available_transitions = self.model.get_available_user_state_transitions(  # type: ignore[attr-defined]
-            self.privileged
+            self.privileged_user
         )
-        transition_names = {transition.name for transition in available_transitions}
-
-        assert {"publish", "remove", "moderate"} == transition_names
+        assert {transition.name for transition in available_transitions} == {
+            "publish",
+            "remove",
+            "moderate",
+        }
 
     def test_unprivileged_access_prohibited(self):
-        assert not fsm.has_transition_perm(self.model.publish, self.unprivileged)
-        assert not fsm.has_transition_perm(self.model.remove, self.unprivileged)
+        assert not fsm.has_transition_perm(self.model.publish, self.unprivileged_user)
+        assert not fsm.has_transition_perm(self.model.remove, self.unprivileged_user)
 
         available_transitions = self.model.get_available_user_state_transitions(  # type: ignore[attr-defined]
-            self.unprivileged
+            self.unprivileged_user
         )
-        transition_names = {transition.name for transition in available_transitions}
-
-        assert {"moderate"} == transition_names
+        assert {transition.name for transition in available_transitions} == {"moderate"}
 
     def test_permission_instance_method(self):
-        assert not fsm.has_transition_perm(self.model.restore, self.unprivileged)
-        assert fsm.has_transition_perm(self.model.restore, self.staff)
+        assert not fsm.has_transition_perm(self.model.restore, self.unprivileged_user)
+        assert fsm.has_transition_perm(self.model.restore, self.staff_user)