feat(backend): switch to a more formal FSM approach (#12507)

* full fsm implementation

closes https://github.com/inventree/InvenTree/issues/12314

based on https://github.com/matmair/InvenTree/pull/721

* update assertations

* refactor to reduce duplication

* move for cleaner diff

* more moving stuff around

* fix assingment

* remove skip

* merge test classes

* re-enable transition plugin tests

* fix docstrings

* add depreciation warning

* fix type

* nitpicks

* small cleanup

* ensure we alwas pass a str

* add backport

* fix marker position

* full fsm implementation

closes https://github.com/inventree/InvenTree/issues/12314

based on https://github.com/matmair/InvenTree/pull/721

* update assertations

* refactor to reduce duplication

* move for cleaner diff

* more moving stuff around

* fix assingment

* remove skip

* merge test classes

* re-enable transition plugin tests

* fix docstrings

* add depreciation warning

* fix type

* nitpicks

* small cleanup

* ensure we alwas pass a str

* add backport

* fix marker position

* fix ty issue

* ignore this corner case

* ensure invalid transitions can raise a nice validation error

* compact code

* add depreciation mark

* make raise_error default (#754)

* fix merge

* adjust test as this is now not a no-op but a raised error

* fix assertations

* fix test to not take a broken path

* remove unused test statements

* converge

* add test branch for actually working transition

* reduce uneeded code

* ignore depreceated methods

* fix missing coverage

* add prefetch

* reduce diff

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Breaking Changes
 
+- [#12507](https://github.com/inventree/InvenTree/pull/12507) calling an invalid or repeated state transition now raises a ValidationError. Plugins implementing state transitions should evaluate the PR and adapt their usage of transitions to gain the new safeguards.
+
 ### Added
 
 ### Changed

src/backend/InvenTree/build/models.py

@@ -41,7 +41,14 @@ from build.validators import (
 from common.models import ProjectCode
 from common.settings import get_global_setting
 from generic.enums import StringEnum
-from generic.states import StateTransitionMixin, StatusCodeMixin
+from generic.states import (
+    Deprecations,
+    StateTransitionMixin,
+    StatusCodeMixin,
+    can_proceed,
+    deprecated,
+    inventree_transition,
+)
 from InvenTree.helpers_db import bulk_create_and_fetch
 from plugin.events import bulk_trigger_event, trigger_event
 from stock.events import StockEvents
@@ -959,41 +966,21 @@ class Build(
         # which point to this Build Order
         self.allocated_stock.all().delete()
 
-    @transaction.atomic
+    # region fsm
+    @inventree_transition(
+        field=status,
+        source=[BuildStatus.PENDING, BuildStatus.PRODUCTION, BuildStatus.ON_HOLD],
+        target=BuildStatus.COMPLETE,
+    )
     def complete_build(self, user: User, trim_allocated_stock: bool = False):
-        """Mark this build as complete.
+        """Transition this Build to COMPLETE status.
 
         Arguments:
             user: The user who is completing the build
             trim_allocated_stock: If True, trim any allocated stock
         """
-        return self.handle_transition(
-            self.status,
-            BuildStatus.COMPLETE.value,
-            self,
-            self._action_complete,
-            user=user,
-            trim_allocated_stock=trim_allocated_stock,
-        )
-
-    def _action_complete(self, *args, **kwargs):
-        """Action to be taken when a build is completed."""
         import build.tasks
 
-        # Lock this build against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous completion requests
-        # could both offload the completion task (duplicate events and
-        # notifications; the stock-consuming side effects are themselves
-        # guarded against double-processing
-        self.status = Build.objects.select_for_update().get(pk=self.pk).status
-
-        if self.status == BuildStatus.COMPLETE.value:
-            return
-
-        trim_allocated_stock = kwargs.pop('trim_allocated_stock', False)
-        user = kwargs.pop('user', None)
-
-        # Prevent completion if there are open child builds
         if (
             get_global_setting('BUILDORDER_REQUIRE_CLOSED_CHILDS')
             and self.has_open_child_builds
@@ -1016,36 +1003,17 @@ class Build(
             group='build',
         )
 
-    @transaction.atomic
-    def issue_build(self):
-        """Mark the Build as IN PRODUCTION.
-
-        Args:
-            user: The user who is issuing the build
-        """
-        return self.handle_transition(
-            self.status, BuildStatus.PENDING.value, self, self._action_issue
+    @inventree_transition(
+        field=status,
+        source=[BuildStatus.PENDING, BuildStatus.ON_HOLD],
+        target=BuildStatus.PRODUCTION,
+        event=BuildEvents.ISSUED,
     )
+    def issue_build(self):
+        """Transition this Build to PRODUCTION status.
 
-    @property
-    def can_issue(self) -> bool:
-        """Returns True if this BuildOrder can be issued."""
-        return self.status in [BuildStatus.PENDING.value, BuildStatus.ON_HOLD.value]
-
-    def _action_issue(self, *args, **kwargs):
-        """Perform the action to mark this order as PRODUCTION."""
-        # Lock this build against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the issue side effects
-        # (duplicate events and offloaded stock-check tasks).
-        self.status = Build.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_issue:
-            self.status = BuildStatus.PRODUCTION.value
-            self.save()
-
-            trigger_event(BuildEvents.ISSUED, id=self.pk)
-
+        The build must currently be PENDING or ON_HOLD.
+        """
         from build.tasks import check_build_stock
 
         # Run checks on required parts
@@ -1053,65 +1021,43 @@ class Build(
             check_build_stock, self, group='build', force_async=True
         )
 
-    @transaction.atomic
-    def hold_build(self):
-        """Mark the Build as ON HOLD."""
-        return self.handle_transition(
-            self.status, BuildStatus.ON_HOLD.value, self, self._action_hold
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_issue(self) -> bool:
+        """Returns True if this BuildOrder can be issued."""
+        return can_proceed(self.issue_build)
+
+    @inventree_transition(
+        field=status,
+        source=[BuildStatus.PENDING, BuildStatus.PRODUCTION],
+        target=BuildStatus.ON_HOLD,
+        event=BuildEvents.HOLD,
     )
+    def hold_build(self):
+        """Transition this Build to ON_HOLD status.
+
+        The build must currently be PENDING or PRODUCTION.
+        """
 
     @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
     def can_hold(self) -> bool:
         """Returns True if this BuildOrder can be placed on hold."""
-        return self.status in [BuildStatus.PENDING.value, BuildStatus.PRODUCTION.value]
+        return can_proceed(self.hold_build)
 
-    def _action_hold(self, *args, **kwargs):
-        """Action to be taken when a build is placed on hold."""
-        # Lock this build against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the hold side effects
-        # (duplicate events).
-        self.status = Build.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_hold:
-            self.status = BuildStatus.ON_HOLD.value
-            self.save()
-
-            trigger_event(BuildEvents.HOLD, id=self.pk)
-
-    @transaction.atomic
-    def cancel_build(self, user, **kwargs):
-        """Mark the Build as CANCELLED.
-
-        - Delete any pending BuildItem objects (but do not remove items from stock)
-        - Set build status to CANCELLED
-        - Save the Build object
-        """
-        return self.handle_transition(
-            self.status,
-            BuildStatus.CANCELLED.value,
-            self,
-            self._action_cancel,
-            user=user,
-            **kwargs,
+    @inventree_transition(
+        field=status,
+        source=[BuildStatus.PENDING, BuildStatus.PRODUCTION, BuildStatus.ON_HOLD],
+        target=BuildStatus.CANCELLED,
     )
+    def cancel_build(self, user=None, **kwargs):
+        """Transition this Build to CANCELLED status.
 
-    def _action_cancel(self, *args, **kwargs):
-        """Action to be taken when a build is cancelled."""
+        Offloads expensive cleanup operations (de-allocating stock, removing
+        incomplete outputs) to a background task.
+        """
         import build.tasks
 
-        # Lock this build against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous cancellation requests
-        # could both offload the cancellation task (duplicate events and
-        # notifications; the stock-consuming side effects are themselves
-        # guarded against double-processing
-        self.status = Build.objects.select_for_update().get(pk=self.pk).status
-
-        if self.status == BuildStatus.CANCELLED.value:
-            return
-
-        user = kwargs.pop('user', None)
-
         remove_allocated_stock = kwargs.get('remove_allocated_stock', False)
         remove_incomplete_outputs = kwargs.get('remove_incomplete_outputs', False)
 
@@ -1129,8 +1075,7 @@ class Build(
         self.completion_date = InvenTree.helpers.current_date()
         self.completed_by = user
 
-        self.status = BuildStatus.CANCELLED.value
-        self.save()
+    # endregion fsm
 
     @transaction.atomic
     def deallocate_stock(self, build_line=None, output=None):
@@ -2139,7 +2084,7 @@ class Build(
         """Rebuild required quantity field for each BuildLine object."""
         lines_to_update = []
 
-        for line in self.build_lines.all():
+        for line in self.build_lines.select_related('bom_item').all():
             line.quantity = line.bom_item.get_required_quantity(self.quantity)
             lines_to_update.append(line)
 

src/backend/InvenTree/build/test_api.py

@@ -2365,7 +2365,7 @@ class BuildConsumeTest(BuildAPITest):
                 {},
                 expected_code=201,
                 benchmark=True,
-                max_query_count=250,
+                max_query_count=180,
                 max_query_time=1.5,
             )
 

src/backend/InvenTree/build/test_build.py

@@ -619,9 +619,10 @@ class BuildTest(BuildTestBase):
         # cancellation must be skipped based on the database state
         self.assertEqual(build_b.status, status.BuildStatus.PRODUCTION)
 
-        with mock.patch('build.models.trigger_event') as trigger:
+        with self.assertRaises(ValidationError) as err:
             build_b.cancel_build(None)
-            trigger.assert_not_called()
+
+        self.assertIn('Build Order is already Cancelled', str(err.exception))
 
         self.build.refresh_from_db()
         self.assertEqual(self.build.status, status.BuildStatus.CANCELLED)

src/backend/InvenTree/generic/states/__init__.py

@@ -6,15 +6,23 @@ There is a rendered state for each state value. The rendered state is used for d
 States can be extended with custom options for each InvenTree instance - those options are stored in the database and need to link back to state values.
 """
 
+from django_fsm import RETURN_VALUE, can_proceed
+
 from . import fields
+from .deprecations import Deprecations, deprecated
 from .states import ColorEnum, StatusCode, StatusCodeMixin
-from .transition import StateTransitionMixin, TransitionMethod
+from .transition import StateTransitionMixin, TransitionMethod, inventree_transition
 
 __all__ = [
+    'RETURN_VALUE',  # django_fsm import
     'ColorEnum',
+    'Deprecations',
     'StateTransitionMixin',
     'StatusCode',
     'StatusCodeMixin',
     'TransitionMethod',
+    'can_proceed',  # django_fsm import
+    'deprecated',
     'fields',
+    'inventree_transition',
 ]

src/backend/InvenTree/generic/states/deprecations.py

@@ -0,0 +1,27 @@
+"""Helper for deprecating old implementation details."""
+
+from enum import Enum
+from typing import Any, Optional
+
+# py (3, 12) does not ship with depreciation decorator, so we need to import it from typing_extensions
+try:
+    from warnings import deprecated as warn_deprecated  # ty: ignore[unresolved-import]
+except ImportError:
+    from typing_extensions import deprecated as warn_deprecated
+
+
+class Deprecations(Enum):
+    """Deprecations for states."""
+
+    CAN_PROCEED = 'Use can_proceed directly'
+
+
+class deprecated(warn_deprecated):  # noqa: N801
+    """Deprecation decorator for state transition methods."""
+
+    def __init__(
+        self, message: Any | str, version: Optional[str] = None, *args, **kwargs
+    ):
+        """Initialize the decorator with a deprecation reason."""
+        self.version = version
+        super().__init__(str(message), *args, **kwargs)

src/backend/InvenTree/generic/states/fields.py

@@ -9,6 +9,7 @@ from django.db import models
 from django.utils.encoding import force_str
 from django.utils.translation import gettext_lazy as _
 
+from django_fsm import FSMFieldMixin
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
@@ -90,11 +91,19 @@ class ExtraCustomChoiceField(CustomChoiceField):
         return super().to_representation(value) or value
 
 
-class InvenTreeCustomStatusModelField(models.PositiveIntegerField):
+class InvenTreeCustomStatusModelField(FSMFieldMixin, models.PositiveIntegerField):
     """Custom model field for extendable status codes.
 
-    Adds a secondary *_custom_key field to the model which can be used to store additional status information.
-    Models using this model field must also include the InvenTreeCustomStatusSerializerMixin in all serializers that create or update the value.
+    Extends Django's PositiveIntegerField with FSM (Finite State Machine) support
+    via django-fsm-2, enabling use of the @transition decorator directly on model
+    methods that use this field.
+
+    Adds a secondary *_custom_key field to the model which can be used to store
+    additional status information.
+
+    Models using this model field must also include the
+    InvenTreeCustomStatusSerializerMixin in all serializers that create or update
+    the value.
     """
 
     def __init__(self, *args, **kwargs):
@@ -109,12 +118,18 @@ class InvenTreeCustomStatusModelField(models.PositiveIntegerField):
             validators.append(CustomStatusCodeValidator(status_class=self.status_class))
 
         kwargs['validators'] = validators
+        # FSM protection is disabled by default; direct status assignment is still allowed
+        kwargs.setdefault('protected', False)
         super().__init__(*args, **kwargs)
 
     def deconstruct(self):
-        """Deconstruct the field for migrations."""
-        name, path, args, kwargs = super().deconstruct()
+        """Deconstruct the field for migrations.
 
+        FSM-specific kwargs (protected) are excluded to avoid spurious migrations,
+        since they do not affect the database schema.
+        """
+        name, path, args, kwargs = super().deconstruct()
+        kwargs.pop('protected', None)
         return name, path, args, kwargs
 
     def contribute_to_class(self, cls, name):
@@ -182,7 +197,7 @@ class InvenTreeCustomStatusModelField(models.PositiveIntegerField):
 
 
 class ExtraInvenTreeCustomStatusModelField(models.PositiveIntegerField):
-    """Custom field used to detect custom extenteded fields.
+    """Custom field used to detect custom extended fields.
 
     This is not intended to be used directly, if you want to support custom states in your model use InvenTreeCustomStatusModelField.
     """

src/backend/InvenTree/generic/states/states.py

@@ -9,8 +9,8 @@ from typing import Optional
 logger = logging.getLogger('inventree')
 
 
-class BaseEnum(enum.IntEnum):  # noqa: PLW1641
-    """An `Enum` capabile of having its members have docstrings.
+class BaseEnum(enum.IntEnum):
+    """An `Enum` capable of having its members have docstrings.
 
     Based on https://stackoverflow.com/questions/19330460/how-do-i-put-docstrings-on-enums
     """
@@ -42,6 +42,14 @@ class BaseEnum(enum.IntEnum):  # noqa: PLW1641
 
         return super().__eq__(obj)
 
+    def __hash__(self):
+        """Return integer hash so enum members are usable as dict keys and in sets.
+
+        Required because we define ``__eq__``: Python sets ``__hash__ = None``
+        when ``__eq__`` is overridden without a matching ``__hash__``.
+        """
+        return hash(self.value)
+
     def __ne__(self, obj):
         """Override inequality operator to allow comparison with int."""
         if type(self) is type(obj):
@@ -262,7 +270,7 @@ class ColorEnum(Enum):
 class StatusCodeMixin:
     """Mixin class which handles custom 'status' fields.
 
-    - Implements a 'set_stutus' method which can be used to set the status of an object
+    - Implements a 'set_status' method which can be used to set the status of an object
     - Implements a 'get_status' method which can be used to retrieve the status of an object
 
     This mixin assumes that the implementing class has a 'status' field,

src/backend/InvenTree/generic/states/test_transition.py

@@ -2,22 +2,158 @@
 
 from django.core.exceptions import ValidationError
 
+from generic.states import can_proceed
 from InvenTree.unit_test import InvenTreeTestCase
-from order.models import ReturnOrder
-from order.status_codes import ReturnOrderStatus
+from order.models import PurchaseOrder, ReturnOrder, SalesOrder, TransferOrder
+from order.status_codes import (
+    PurchaseOrderStatus,
+    ReturnOrderStatus,
+    SalesOrderStatus,
+    TransferOrderStatus,
+)
 from plugin import registry
+from users.models import Owner
 
 
 class TransitionTests(InvenTreeTestCase):
     """Tests for custom state transition logic."""
 
-    fixtures = ['company', 'return_order', 'part', 'stock', 'location', 'category']
-
     def setUp(self):
         """Set up the test environment."""
         super().setUp()
         self.ensurePluginsLoaded()
 
+    fixtures = [
+        'company',
+        'supplier_part',
+        'category',
+        'part',
+        'location',
+        'stock',
+        'order',
+        'sales_order',
+        'return_order',
+        'transfer_order',
+    ]
+
+    def test_fsm_decorator_applied_to_purchase_order(self):
+        """Verify @inventree_transition metadata is attached to PurchaseOrder methods."""
+        # Methods decorated with @inventree_transition expose _django_fsm metadata
+        self.assertTrue(hasattr(PurchaseOrder.place_order, '_django_fsm'))
+        self.assertTrue(hasattr(PurchaseOrder.complete_order, '_django_fsm'))
+        self.assertTrue(hasattr(PurchaseOrder.hold_order, '_django_fsm'))
+        self.assertTrue(hasattr(PurchaseOrder.cancel_order, '_django_fsm'))
+
+    def test_fsm_can_proceed_purchase_order(self):
+        """Test can_proceed() reflects current state correctly for PurchaseOrder."""
+        po = PurchaseOrder.objects.filter(
+            status=PurchaseOrderStatus.PENDING.value
+        ).first()
+        assert po
+
+        # A PENDING order can be placed or cancelled, but not completed
+        self.assertTrue(po.can_issue)
+        self.assertTrue(po.can_cancel)
+        self.assertFalse(can_proceed(po.complete_order))
+        self.assertTrue(po.is_open)
+
+        # depreceted methods
+        self.assertTrue(po.can_hold)
+
+    def test_fsm_purchase_order_transitions(self):
+        """Test that PurchaseOrder transitions work correctly via @inventree_transition."""
+        po = PurchaseOrder.objects.filter(
+            status=PurchaseOrderStatus.PENDING.value
+        ).first()
+        assert po
+
+        # Place the order (PENDING → PLACED)
+        result = po.place_order()
+        self.assertTrue(result)
+        po.refresh_from_db()
+        self.assertEqual(po.status, PurchaseOrderStatus.PLACED.value)
+
+        # Hold the order (PLACED → ON_HOLD)
+        result = po.hold_order()
+        self.assertTrue(result)
+        po.refresh_from_db()
+        self.assertEqual(po.status, PurchaseOrderStatus.ON_HOLD.value)
+
+        # Place again from ON_HOLD (ON_HOLD → PLACED)
+        result = po.place_order()
+        self.assertTrue(result)
+        po.refresh_from_db()
+        self.assertEqual(po.status, PurchaseOrderStatus.PLACED.value)
+
+        # Cancel the order (PLACED → CANCELLED)
+        result = po.cancel_order()
+        self.assertTrue(result)
+        po.refresh_from_db()
+        self.assertEqual(po.status, PurchaseOrderStatus.CANCELLED.value)
+
+    def test_fsm_sale_order_transitions(self):
+        """Test that SaleOrder transitions work correctly via @inventree_transition."""
+        so = SalesOrder.objects.filter(status=SalesOrderStatus.PENDING.value).first()
+        assert so
+
+        # deprecated methods
+        self.assertTrue(so.can_issue)
+        self.assertTrue(so.can_hold)
+        self.assertTrue(so.can_cancel)
+
+    def test_fsm_invalid_transition_returns_false(self):
+        """Test that an invalid transition returns False (backward-compatible behaviour)."""
+        po = PurchaseOrder.objects.filter(
+            status=PurchaseOrderStatus.CANCELLED.value
+        ).first()
+        assert po
+
+        # Attempting to place a cancelled order must raise
+        with self.assertRaises(ValidationError):
+            po.place_order()
+
+        # The order status must remain CANCELLED
+        po.refresh_from_db()
+        self.assertEqual(po.status, PurchaseOrderStatus.CANCELLED.value)
+
+    def test_fsm_return_order_transitions(self):
+        """Test that ReturnOrder transitions work correctly via @inventree_transition."""
+        ro = ReturnOrder.objects.filter(
+            status=ReturnOrderStatus.IN_PROGRESS.value
+        ).first()
+        assert ro
+
+        # Can complete an IN_PROGRESS return order
+        self.assertTrue(can_proceed(ro.complete_order))
+
+        # Cannot issue an IN_PROGRESS return order (already issued)
+        self.assertFalse(can_proceed(ro.issue_order))
+
+        # deprecated methods
+        self.assertTrue(ro.can_hold)
+        self.assertTrue(ro.can_cancel)
+        self.assertFalse(ro.can_issue)
+
+    def test_fsm_transferorder(self):
+        """Test that TransferOrder transitions work correctly via @inventree_transition."""
+        to = TransferOrder.objects.filter(
+            status=TransferOrderStatus.PENDING.value
+        ).first()
+        assert to
+
+        self.assertTrue(to.can_issue)
+        self.assertTrue(to.can_hold)
+        self.assertTrue(to.can_cancel)
+
+    def test_fsm_can_issue_property(self):
+        """Test the can_issue property delegates to can_proceed."""
+        po = PurchaseOrder.objects.filter(
+            status=PurchaseOrderStatus.PENDING.value
+        ).first()
+        assert po
+
+        self.assertEqual(po.can_issue, can_proceed(po.place_order))
+
     def test_return_order(self):
         """Test transition of a return order."""
         # Ensure plugin is enabled
@@ -36,6 +172,13 @@ class TransitionTests(InvenTreeTestCase):
             str(e.exception),
         )
 
+        ro.responsible = Owner.create(obj=self.user)
+        ro.save()
+        result = ro.complete_order()
+        self.assertEqual(result, '123#abc!')
+        # There should be no change in the status of the return order
+        self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
+
         # Now disable the plugin
         registry.set_plugin_state('sample-transition', False)
 
@@ -61,9 +204,6 @@ class TransitionTests(InvenTreeTestCase):
         ro = ReturnOrder.objects.get(pk=2)
         self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
 
-        # Transition to "ON HOLD" state
-        ro.hold_order()
-
         # Ensure plugin starts in a known state
         plugin = registry.get_plugin('sample-broken-transition')
         plugin.set_setting('BROKEN_GET_METHOD', False)
@@ -76,7 +216,7 @@ class TransitionTests(InvenTreeTestCase):
 
         with self.assertWarnsMessage(UserWarning, msg):
             # No error should occur here
-            ro.complete_order()
+            ro.hold_order()
             self.assertEqual(ro.status, ReturnOrderStatus.ON_HOLD.value)
 
         # No error should be logged
@@ -91,7 +231,7 @@ class TransitionTests(InvenTreeTestCase):
                 ro.issue_order()
                 self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
 
-            # Ensure correct eroror was logged
+            # Ensure correct error was logged
             self.assertIn('Invalid transition handler type: 1', str(cm.output[0]))
 
         # Now, enable the "WRONG_RETURN_TYPE" setting

src/backend/InvenTree/generic/states/transition.py

@@ -1,18 +1,138 @@
 """Classes and functions for plugin controlled object state transitions."""
 
 from collections.abc import Callable
+from enum import Enum
+from functools import wraps
 
+from django.core.exceptions import ValidationError
+from django.db import transaction
 from django.db.models import Model
 
 import structlog
+from django_fsm import TransitionNotAllowed, transition
+
+from plugin.events import trigger_event
+
+from .deprecations import deprecated
 
 logger = structlog.get_logger('inventree')
 
 
-class TransitionMethod:
-    """Base class for all transition classes.
+def inventree_transition(
+    field, source, target, event=None, refresh_field: bool = True, **extra
+):
+    """Decorator that combines ``@django_fsm.transition`` with plugin transitions.
 
-    Must implement a method called `transition` that takes both args and kwargs.
+    Wraps a model method so that:
+    * Source/target state validation is enforced via ``django_fsm.transition``.
+    * Plugin transitions are called automatically before the FSM transition executes.
+    * The model instance is saved after a successful transition.
+    * The whole operation is wrapped in a database transaction.
+
+    Usage::
+
+        class MyOrder(StateTransitionMixin, models.Model):
+            status = InvenTreeCustomStatusModelField(
+                status_class=MyOrderStatus, ...
+            )
+
+            @inventree_transition(
+                field=status,
+                source=[MyOrderStatus.PENDING, MyOrderStatus.ON_HOLD],
+                target=MyOrderStatus.PLACED,
+                event=MyOrderEvents.PLACED,
+                refresh_field=False,
+            )
+            def place_order(self):
+                notify_responsible(...)
+
+    The ``can_<method>`` guard pattern can be preserved using ``can_proceed``::
+
+        @property
+        def can_issue(self) -> bool:
+            return can_proceed(self.place_order)
+
+    Args:
+        field: The ``InvenTreeCustomStatusModelField`` (or its string name) that holds the state.
+        source: Allowed source state(s) for the transition.
+        target: Target state after the transition.
+        event: Optional event (name) to trigger after the transition.  If not provided no event is triggered.
+        refresh_field: (default True) If True, the field is updated from the database before the transition.
+        **extra: Additional keyword arguments forwarded to `django_fsm.transition` (e.g. `conditions`, `on_error`, `permission`).
+    """
+
+    def decorator(func):
+        # Apply the django-fsm @transition decorator first.
+        fsm_func = transition(field=field, source=source, target=target, **extra)(func)
+
+        @transaction.atomic
+        @wraps(func)
+        def wrapper(self, *args, **kwargs):
+            """Ensure that transitions are handled correctly."""
+            if refresh_field:
+                try:
+                    # Update the field from the database to avoid race conditions
+                    current_obj = type(self).objects.select_for_update().get(pk=self.pk)
+                    new_value = getattr(current_obj, field.name)
+                    setattr(self, field.name, new_value)
+                except type(self).DoesNotExist:  # pragma: no cover
+                    raise ValidationError(
+                        f'{self._meta.verbose_name} with pk={self.pk} does not exist in the database'
+                    )
+
+            # Run plugin transition handlers - if no step is taken the decorated method is called
+            if result := _run_plugin_transition_handlers(
+                self,
+                getattr(self, field.name),
+                target,
+                default_action=_noop_default_action,
+            ):
+                return result
+
+            # Execute the FSM-wrapped function.  This validates the source state,
+            # runs the method body, and updates the field to the target state.
+            try:
+                fsm_func(self, *args, **kwargs)
+            except TransitionNotAllowed as exc:
+                # Preserve backward-compatible behaviour: an invalid transition
+                # (wrong source state, failed condition, or explicit raise inside
+                # the method body) returns False rather than raising.
+                if getattr(self, field.name) == target:
+                    target_val = (
+                        target.label
+                        if isinstance(target, Enum) and hasattr(target, 'label')
+                        else target
+                    )
+                    raise ValidationError(
+                        f'{self._meta.verbose_name} is already {target_val}'
+                    ) from exc
+                raise ValidationError(
+                    f'Invalid transition on {self._meta.verbose_name}.{field.name} (source value should be {source}, is {getattr(self, field.name)})'
+                ) from exc
+            # Persist all changes (including the updated status field) to the DB.
+            self.save()
+
+            # trigger event for the transition
+            if event:
+                trigger_event(event, id=self.pk)
+
+            return True
+
+        # Propagate the django-fsm metadata so that can_proceed() works correctly
+        # on the public wrapper method.
+        if hasattr(fsm_func, '_django_fsm'):
+            wrapper._django_fsm = fsm_func._django_fsm
+        return wrapper
+
+    return decorator
+
+
+class TransitionMethod:
+    """Base class for all plugin-defined transition handler classes.
+
+    A plugin transition handler is called for every state transition on a
+    ``StateTransitionMixin`` model.  Subclasses must implement a ``transition``
+    method.
     """
 
     def __init__(self) -> None:
@@ -37,33 +157,36 @@ class TransitionMethod:
     ) -> bool:
         """Perform a state transition.
 
-        Success:
-            - The custom transition logic succeeded
-            - Return True result
-            - No further transitions are attempted
-        Ignore:
-            - The custom transition logic did not apply
-            - Return False result
-            - Further transitions are attempted (if available)
-            - Default action is called if no transition was successful
-        Failure:
-            - The custom transition logic failed
-            - Raise a ValidationError
-            - No further transitions are attempted
-            - Default action is not called
+        When used with ``@inventree_transition``-decorated methods, plugin
+        handlers are invoked before the decorated method body executes.
+        The semantics are:
+
+        * **Ignore** - return ``False``.  Further handlers are attempted;
+          the decorated method body executes as normal.
+        * **Veto** - raise ``ValidationError``.  The transition is cancelled;
+          no further handlers are called; the method body does *not* execute.
+        * **Handle** (deprecated) - return ``True``.  This historically meant
+          "I handled the transition; skip the default action."  Under
+          ``@inventree_transition``, the decorated method body *always* runs, so
+          returning ``True`` now only triggers a ``DeprecationWarning``.  Plugin
+          authors should raise ``ValidationError`` to cancel a transition, or
+          return ``False`` to allow it to proceed.
 
         Arguments:
             current_state: int - Current state of the instance.
             target_state: int - Target state to transition to.
             instance: Model - The object instance to transition.
-            default_action: callable - Default action to be taken if no transition is successful.
+            default_action: callable - No-op when called from the
+                ``inventree_transition`` wrapper; present for backward
+                compatibility.
             **kwargs: Additional keyword arguments for custom logic.
 
         Returns:
-            result: bool - True if the transition method was successful (and no further transitions are attempted), False otherwise.
+            result: bool - ``True`` triggers a deprecation warning (see above).
+                ``False`` allows the transition to proceed normally.
 
         Raises:
-            ValidationError: Alert the user that the transition failed
+            ValidationError: Cancels the transition.
         """
         raise NotImplementedError(
             'TransitionMethod.transition must be implemented'
@@ -71,30 +194,66 @@ class TransitionMethod:
 
 
 class StateTransitionMixin:
-    """Mixin class to enable state transitions.
+    """Mixin class to enable plugin-controlled state transitions.
 
-    This mixin is used to add state transitions handling to a model. With this you can apply custom logic to state transitions via plugins.
-    ```python
-    class MyModel(StateTransitionMixin, models.Model):
-        def some_dummy_function(self, *args, **kwargs):
+    Add this mixin to a Django model to gain:
+
+    * Plugin hook support for all ``@inventree_transition``-decorated methods
+      (via the ``inventree_transition`` decorator).
+    * A legacy ``handle_transition`` method for backward compatibility with
+      code that calls it directly.
+
+    Example::
+
+        class MyOrder(StateTransitionMixin, models.Model):
+            status = InvenTreeCustomStatusModelField(
+                status_class=MyOrderStatus, ...
+            )
+
+            @inventree_transition(
+                field=status,
+                source=[MyOrderStatus.PENDING],
+                target=MyOrderStatus.PLACED,
+            )
+            def place_order(self):
                 pass
-
-        def action(self, *args, **kwargs):
-            self.handle_transition(0, 1, self, self.some_dummy_function)
-    ```
     """
 
+    @deprecated('Use the @inventree_transition decorator instead', version='1.5.0')
     def handle_transition(
         self, current_state, target_state, instance, default_action, **kwargs
-    ):
+    ):  # pragma: no cover
         """Handle a state transition for an object.
 
+        .. deprecated::
+            Use the ``@inventree_transition`` decorator instead.  This method
+            is kept for backward compatibility only and calls the ``default_action``
+            directly without invoking the django-fsm machinery or signals.
+
         Args:
             current_state: Current state of instance
             target_state: Target state of instance
             instance: Object instance
-            default_action: Default action to be taken if none of the transitions returns a boolean true value
+            default_action: Default action to be taken if none of the
+                transitions returns a boolean true value
         """
+        if result := _run_plugin_transition_handlers(
+            instance, current_state, target_state, default_action=default_action
+        ):
+            return result
+        return default_action(current_state, target_state, instance, **kwargs)
+
+
+def _run_plugin_transition_handlers(instance, source, target, default_action):
+    """Run plugin transition handlers before a state transition.
+
+    A plugin handler may veto the transition by raising a ``ValidationError``.
+    """
+    if not isinstance(instance, StateTransitionMixin) or not isinstance(
+        instance, Model
+    ):
+        return  # pragma: no cover
+
     from InvenTree.exceptions import log_error
     from plugin import PluginMixinEnum, registry
 
@@ -116,18 +275,15 @@ class StateTransitionMixin:
 
         for handler in handlers:
             if not isinstance(handler, TransitionMethod):
-                    logger.error(
-                        'INVE-E9: Invalid transition handler type: %s', handler
-                    )
+                logger.error('INVE-E9: Invalid transition handler type: %s', handler)
                 continue
 
-                # Call the transition method
-                result = handler.transition(
-                    current_state, target_state, instance, default_action, **kwargs
-                )
-
-                if result:
+            # Call the transition method.
+            # ValidationError raised here will propagate and cancel the transition.
+            if result := handler.transition(source, target, instance, default_action):
                 return result
 
-        # Default action
-        return default_action(current_state, target_state, instance, **kwargs)
+
+def _noop_default_action(current_state, target_state, instance, **kwargs):
+    """No-op default action for compatibility with transition handlers."""
+    return None  # pragma: no cover

src/backend/InvenTree/order/models.py

@@ -37,7 +37,15 @@ from common.currency import currency_code_default
 from common.notifications import InvenTreeNotificationBodies
 from common.settings import get_global_setting
 from company.models import Address, Company, Contact, SupplierPart
-from generic.states import StateTransitionMixin, StatusCodeMixin
+from generic.states import (
+    RETURN_VALUE,
+    Deprecations,
+    StateTransitionMixin,
+    StatusCodeMixin,
+    can_proceed,
+    deprecated,
+    inventree_transition,
+)
 from generic.states.fields import InvenTreeCustomStatusModelField
 from InvenTree.exceptions import log_error
 from InvenTree.fields import (
@@ -852,26 +860,20 @@ class PurchaseOrder(TotalPriceMixin, Order):
 
         return line
 
-    # region state changes
-    def _action_place(self, *args, **kwargs):
-        """Marks the PurchaseOrder as PLACED.
+    # region fsm
+    @inventree_transition(
+        field=status,
+        source=[PurchaseOrderStatus.PENDING, PurchaseOrderStatus.ON_HOLD],
+        target=PurchaseOrderStatus.PLACED,
+        event=PurchaseOrderEvents.PLACED,
+    )
+    def place_order(self):
+        """Transition this PurchaseOrder to PLACED status.
 
-        Order must be currently PENDING.
+        The order must currently be PENDING or ON_HOLD.
         """
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the transition side
-        # effects (duplicate events and notifications).
-        self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_issue:
-            self.status = PurchaseOrderStatus.PLACED.value
         self.issue_date = InvenTree.helpers.current_date()
-            self.save()
 
-            trigger_event(PurchaseOrderEvents.PLACED, id=self.pk)
-
-            # Notify users that the order has been placed
         notify_responsible(
             self,
             PurchaseOrder,
@@ -880,77 +882,92 @@ class PurchaseOrder(TotalPriceMixin, Order):
             extra_users=self.subscribed_users(),
         )
 
-    def _action_complete(self, *args, **kwargs):
-        """Marks the PurchaseOrder as COMPLETE.
+    @deprecated('Use place_order() instead', version='1.5.0')
+    def issue_order(self):
+        """Equivalent to place_order()."""
+        return self.place_order()
 
-        Order must be currently PLACED.
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_issue(self) -> bool:
+        """Return True if this order can be issued (placed)."""
+        return can_proceed(self.place_order)
+
+    @inventree_transition(
+        field=status,
+        source=PurchaseOrderStatus.PLACED,
+        target=PurchaseOrderStatus.COMPLETE,
+        event=PurchaseOrderEvents.COMPLETED,
+    )
+    def complete_order(self):
+        """Transition this PurchaseOrder to COMPLETE status.
+
+        The order must currently be PLACED.
         """
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe status=PLACED, and each would run the completion side effects
-        # (duplicate events and pricing-update scheduling).
-        self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.status == PurchaseOrderStatus.PLACED:
-            self.status = PurchaseOrderStatus.COMPLETE.value
         self.complete_date = InvenTree.helpers.current_date()
 
-            self.save()
-
         unique_parts = set()
 
-            # Schedule pricing update for any referenced parts
         for line in self.lines.all().prefetch_related('part__part'):
-                # Ensure we only check 'unique' parts
             if line.part and line.part.part:
                 unique_parts.add(line.part.part)
 
         for part in unique_parts:
             part.schedule_pricing_update(create=True, refresh=False)
 
-            trigger_event(PurchaseOrderEvents.COMPLETED, id=self.pk)
+    @inventree_transition(
+        field=status,
+        source=[PurchaseOrderStatus.PENDING, PurchaseOrderStatus.PLACED],
+        target=PurchaseOrderStatus.ON_HOLD,
+        event=PurchaseOrderEvents.HOLD,
+    )
+    def hold_order(self):
+        """Transition this PurchaseOrder to ON_HOLD status.
 
-    @transaction.atomic
-    def issue_order(self):
-        """Equivalent to 'place_order'."""
-        return self.place_order()
+        The order must currently be PENDING or PLACED.
+        """
 
     @property
-    def can_issue(self) -> bool:
-        """Return True if this order can be issued."""
-        return self.status in [
-            PurchaseOrderStatus.PENDING.value,
-            PurchaseOrderStatus.ON_HOLD.value,
-        ]
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_hold(self) -> bool:
+        """Return True if this order can be placed on hold."""
+        return can_proceed(self.hold_order)
 
-    @transaction.atomic
-    def place_order(self):
-        """Attempt to transition to PLACED status."""
-        return self.handle_transition(
-            self.status, PurchaseOrderStatus.PLACED.value, self, self._action_place
+    @inventree_transition(
+        field=status,
+        source=[
+            PurchaseOrderStatus.PENDING,
+            PurchaseOrderStatus.ON_HOLD,
+            PurchaseOrderStatus.PLACED,
+        ],
+        target=PurchaseOrderStatus.CANCELLED,
+        event=PurchaseOrderEvents.CANCELLED,
     )
-
-    @transaction.atomic
-    def complete_order(self):
-        """Attempt to transition to COMPLETE status."""
-        return self.handle_transition(
-            self.status, PurchaseOrderStatus.COMPLETE.value, self, self._action_complete
-        )
-
-    @transaction.atomic
-    def hold_order(self):
-        """Attempt to transition to ON_HOLD status."""
-        return self.handle_transition(
-            self.status, PurchaseOrderStatus.ON_HOLD.value, self, self._action_hold
-        )
-
-    @transaction.atomic
     def cancel_order(self):
-        """Attempt to transition to CANCELLED status."""
-        return self.handle_transition(
-            self.status, PurchaseOrderStatus.CANCELLED.value, self, self._action_cancel
+        """Transition this PurchaseOrder to CANCELLED status.
+
+        The order must currently be open (PENDING, ON_HOLD, or PLACED).
+        """
+        notify_responsible(
+            self,
+            PurchaseOrder,
+            exclude=self.created_by,
+            content=InvenTreeNotificationBodies.OrderCanceled,
+            extra_users=self.subscribed_users(),
         )
 
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_cancel(self) -> bool:
+        """A PurchaseOrder can only be cancelled while it is open.
+
+        - Status is PLACED
+        - Status is PENDING (or ON_HOLD)
+        """
+        return can_proceed(self.cancel_order)
+
+    # endregion fsm
+
     @property
     def is_pending(self) -> bool:
         """Return True if the PurchaseOrder is 'pending'."""
@@ -961,62 +978,6 @@ class PurchaseOrder(TotalPriceMixin, Order):
         """Return True if the PurchaseOrder is 'open'."""
         return self.status in PurchaseOrderStatusGroups.OPEN
 
-    @property
-    def can_cancel(self) -> bool:
-        """A PurchaseOrder can only be cancelled under the following circumstances.
-
-        - Status is PLACED
-        - Status is PENDING (or ON_HOLD)
-        """
-        return self.status in PurchaseOrderStatusGroups.OPEN
-
-    def _action_cancel(self, *args, **kwargs):
-        """Marks the PurchaseOrder as CANCELLED."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the cancellation side
-        # effects (duplicate events and notifications).
-        self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_cancel:
-            self.status = PurchaseOrderStatus.CANCELLED.value
-            self.save()
-
-            trigger_event(PurchaseOrderEvents.CANCELLED, id=self.pk)
-
-            # Notify users that the order has been canceled
-            notify_responsible(
-                self,
-                PurchaseOrder,
-                exclude=self.created_by,
-                content=InvenTreeNotificationBodies.OrderCanceled,
-                extra_users=self.subscribed_users(),
-            )
-
-    @property
-    def can_hold(self) -> bool:
-        """Return True if this order can be placed on hold."""
-        return self.status in [
-            PurchaseOrderStatus.PENDING.value,
-            PurchaseOrderStatus.PLACED.value,
-        ]
-
-    def _action_hold(self, *args, **kwargs):
-        """Mark this purchase order as 'on hold'."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the hold side effects
-        # (duplicate events).
-        self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_hold:
-            self.status = PurchaseOrderStatus.ON_HOLD.value
-            self.save()
-
-            trigger_event(PurchaseOrderEvents.HOLD, id=self.pk)
-
-    # endregion
-
     def pending_line_items(self) -> QuerySet:
         """Return a list of pending line items for this order.
 
@@ -1837,35 +1798,25 @@ class SalesOrder(TotalPriceMixin, Order):
 
         return True
 
-    # region state changes
+    # region fsm
+    @deprecated("Use 'issue_order' instead", version='1.5.0')
     def place_order(self):
         """Deprecated version of 'issue_order'."""
-        self.issue_order()
+        return self.issue_order()
 
-    @property
-    def can_issue(self) -> bool:
-        """Return True if this order can be issued."""
-        return self.status in [
-            SalesOrderStatus.PENDING.value,
-            SalesOrderStatus.ON_HOLD.value,
-        ]
+    @inventree_transition(
+        field=status,
+        source=[SalesOrderStatus.PENDING, SalesOrderStatus.ON_HOLD],
+        target=SalesOrderStatus.IN_PROGRESS,
+        event=SalesOrderEvents.ISSUED,
+    )
+    def issue_order(self):
+        """Transition this SalesOrder to IN_PROGRESS status.
 
-    def _action_place(self, *args, **kwargs):
-        """Change this order from 'PENDING' to 'IN_PROGRESS'."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the transition side
-        # effects (duplicate events and notifications).
-        self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_issue:
-            self.status = SalesOrderStatus.IN_PROGRESS.value
+        The order must currently be PENDING or ON_HOLD.
+        """
         self.issue_date = InvenTree.helpers.current_date()
-            self.save()
 
-            trigger_event(SalesOrderEvents.ISSUED, id=self.pk)
-
-            # Notify users that the order has been placed
         notify_responsible(
             self,
             SalesOrder,
@@ -1875,49 +1826,46 @@ class SalesOrder(TotalPriceMixin, Order):
         )
 
     @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_issue(self) -> bool:
+        """Return True if this order can be issued."""
+        return can_proceed(self.issue_order)
+
+    @inventree_transition(
+        field=status,
+        source=[SalesOrderStatus.PENDING, SalesOrderStatus.IN_PROGRESS],
+        target=SalesOrderStatus.ON_HOLD,
+        event=SalesOrderEvents.HOLD,
+    )
+    def hold_order(self):
+        """Transition this SalesOrder to ON_HOLD status.
+
+        The order must currently be PENDING or IN_PROGRESS.
+        """
+
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
     def can_hold(self) -> bool:
         """Return True if this order can be placed on hold."""
-        return self.status in [
-            SalesOrderStatus.PENDING.value,
-            SalesOrderStatus.IN_PROGRESS.value,
-        ]
+        return can_proceed(self.hold_order)
 
-    def _action_hold(self, *args, **kwargs):
-        """Mark this sales order as 'on hold'."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the hold side effects
-        # (duplicate events).
-        self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
+    def _ship_complete_action(self, user=None, **kwargs) -> SalesOrderStatus:
+        """Shared logic for ship_order and complete_order.
 
-        if self.can_hold:
-            self.status = SalesOrderStatus.ON_HOLD.value
-            self.save()
-
-            trigger_event(SalesOrderEvents.HOLD, id=self.pk)
-
-    @transaction.atomic
-    def _action_complete(self, *args, **kwargs):
-        """Mark this order as "complete."""
-        user = kwargs.pop('user', None)
-
-        # Lock this order against concurrent completion, and re-read the status
-        # from the database. Without this, two simultaneous completion requests
-        # can both observe an "open" status, and each would run the completion
-        # side effects (duplicate events, pricing updates, and shipped quantity
-        # updates against the virtual line items).
-        self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
+        Returns the target state (SHIPPED or COMPLETE) based on global settings
+        and the current order state.
 
+        Raises:
+            TransitionNotAllowed: if business-logic preconditions are not met.
+        """
         if not self.can_complete(**kwargs):
-            return False
+            raise ValidationError('Order cannot be shipped or completed at this time')
 
         bypass_shipped = InvenTree.helpers.str2bool(
             get_global_setting('SALESORDER_SHIP_COMPLETE')
         )
 
-        # Update line items
         for line in self.lines.all():
-            # Mark any "virtual" parts as shipped at this point
             if line.part and line.part.virtual and line.shipped != line.quantity:
                 line.shipped = line.quantity
                 line.save()
@@ -1925,51 +1873,73 @@ class SalesOrder(TotalPriceMixin, Order):
             if line.part:
                 line.part.schedule_pricing_update(create=True)
 
-        if bypass_shipped or self.status == SalesOrderStatus.SHIPPED:
-            self.status = SalesOrderStatus.COMPLETE.value
-        else:
-            self.status = SalesOrderStatus.SHIPPED.value
-
         if self.shipment_date is None:
             self.shipped_by = user
             self.shipment_date = InvenTree.helpers.current_date()
 
-        self.save()
-
         trigger_event(SalesOrderEvents.COMPLETED, id=self.pk)
 
-        return True
+        if bypass_shipped or self.status == SalesOrderStatus.SHIPPED:
+            return SalesOrderStatus.COMPLETE
+        else:
+            return SalesOrderStatus.SHIPPED
 
-    @property
-    def can_cancel(self) -> bool:
-        """Return True if this order can be cancelled."""
-        return self.is_open
+    @inventree_transition(
+        field=status,
+        source=[
+            SalesOrderStatus.PENDING,
+            SalesOrderStatus.IN_PROGRESS,
+            SalesOrderStatus.ON_HOLD,
+            SalesOrderStatus.SHIPPED,
+        ],
+        target=RETURN_VALUE(SalesOrderStatus.SHIPPED, SalesOrderStatus.COMPLETE),
+    )
+    def ship_order(self, user=None, **kwargs):
+        """Attempt to ship or complete this SalesOrder.
 
-    def _action_cancel(self, *args, **kwargs):
-        """Cancel this order (only if it is "open").
-
-        Executes:
-        - Mark the order as 'cancelled'
-        - Delete any StockItems which have been allocated
+        The order must currently be PENDING, IN_PROGRESS, ON_HOLD, or already SHIPPED.
+        Depending on global settings, the order will transition to SHIPPED or COMPLETE.
         """
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an "open" status, and each would run the cancellation side
-        # effects (duplicate events; the allocation deletion itself is
-        # idempotent).
-        self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
+        return self._ship_complete_action(user=user, **kwargs)
 
-        if not self.can_cancel:
-            return False
+    @inventree_transition(
+        field=status,
+        source=[
+            SalesOrderStatus.PENDING,
+            SalesOrderStatus.IN_PROGRESS,
+            SalesOrderStatus.ON_HOLD,
+            SalesOrderStatus.SHIPPED,
+        ],
+        target=RETURN_VALUE(SalesOrderStatus.SHIPPED, SalesOrderStatus.COMPLETE),
+    )
+    def complete_order(self, user=None, **kwargs):
+        """Attempt to complete this SalesOrder.
 
-        self.status = SalesOrderStatus.CANCELLED.value
-        self.save()
+        The order must currently be IN_PROGRESS, ON_HOLD, or SHIPPED.
+        Depending on global settings, the order will transition to SHIPPED or COMPLETE.
+        """
+        return self._ship_complete_action(user=user, **kwargs)
 
-        SalesOrderAllocation.objects.filter(line__order=self).delete()
+    @inventree_transition(
+        field=status,
+        source=[
+            SalesOrderStatus.PENDING,
+            SalesOrderStatus.ON_HOLD,
+            SalesOrderStatus.IN_PROGRESS,
+            SalesOrderStatus.SHIPPED,
+        ],
+        target=SalesOrderStatus.CANCELLED,
+        event=SalesOrderEvents.CANCELLED,
+    )
+    def cancel_order(self):
+        """Transition this SalesOrder to CANCELLED status.
 
-        trigger_event(SalesOrderEvents.CANCELLED, id=self.pk)
+        Deletes all pending stock allocations.
+        """
+        for line in self.lines.all():
+            for allocation in line.allocations.all():
+                allocation.delete()
 
-        # Notify users that the order has been canceled
         notify_responsible(
             self,
             SalesOrder,
@@ -1978,54 +1948,13 @@ class SalesOrder(TotalPriceMixin, Order):
             extra_users=self.subscribed_users(),
         )
 
-        return True
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_cancel(self) -> bool:
+        """Return True if this order can be cancelled."""
+        return can_proceed(self.cancel_order)
 
-    @transaction.atomic
-    def issue_order(self):
-        """Attempt to transition to IN_PROGRESS status."""
-        return self.handle_transition(
-            self.status, SalesOrderStatus.IN_PROGRESS.value, self, self._action_place
-        )
-
-    @transaction.atomic
-    def ship_order(self, user, **kwargs):
-        """Attempt to transition to SHIPPED status."""
-        return self.handle_transition(
-            self.status,
-            SalesOrderStatus.SHIPPED.value,
-            self,
-            self._action_complete,
-            user=user,
-            **kwargs,
-        )
-
-    @transaction.atomic
-    def complete_order(self, user, **kwargs):
-        """Attempt to transition to COMPLETED status."""
-        return self.handle_transition(
-            self.status,
-            SalesOrderStatus.COMPLETE.value,
-            self,
-            self._action_complete,
-            user=user,
-            **kwargs,
-        )
-
-    @transaction.atomic
-    def hold_order(self):
-        """Attempt to transition to ON_HOLD status."""
-        return self.handle_transition(
-            self.status, SalesOrderStatus.ON_HOLD.value, self, self._action_hold
-        )
-
-    @transaction.atomic
-    def cancel_order(self):
-        """Attempt to transition to CANCELLED status."""
-        return self.handle_transition(
-            self.status, SalesOrderStatus.CANCELLED.value, self, self._action_cancel
-        )
-
-    # endregion
+    # endregion fsm
 
     @property
     def line_count(self) -> int:
@@ -3421,48 +3350,38 @@ class ReturnOrder(TotalPriceMixin, Order):
         """Return True if this order is fully received."""
         return not self.lines.filter(received_date=None).exists()
 
+    # region fsm
+
+    @inventree_transition(
+        field=status,
+        source=[ReturnOrderStatus.PENDING, ReturnOrderStatus.IN_PROGRESS],
+        target=ReturnOrderStatus.ON_HOLD,
+        event=ReturnOrderEvents.HOLD,
+    )
+    def hold_order(self):
+        """Transition this ReturnOrder to ON_HOLD status.
+
+        The order must currently be PENDING or IN_PROGRESS.
+        """
+
     @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
     def can_hold(self):
         """Return True if this order can be placed on hold."""
-        return self.status in [
-            ReturnOrderStatus.PENDING.value,
-            ReturnOrderStatus.IN_PROGRESS.value,
-        ]
+        return can_proceed(self.hold_order)
 
-    def _action_hold(self, *args, **kwargs):
-        """Mark this order as 'on hold' (if allowed)."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the hold side effects
-        # (duplicate events).
-        self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_hold:
-            self.status = ReturnOrderStatus.ON_HOLD.value
-            self.save()
-
-            trigger_event(ReturnOrderEvents.HOLD, id=self.pk)
-
-    @property
-    def can_cancel(self):
-        """Return True if this order can be cancelled."""
-        return self.status in ReturnOrderStatusGroups.OPEN
-
-    def _action_cancel(self, *args, **kwargs):
-        """Cancel this ReturnOrder (if not already cancelled)."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the cancellation side
-        # effects (duplicate events and notifications).
-        self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_cancel:
-            self.status = ReturnOrderStatus.CANCELLED.value
-            self.save()
-
-            trigger_event(ReturnOrderEvents.CANCELLED, id=self.pk)
-
-            # Notify users that the order has been canceled
+    @inventree_transition(
+        field=status,
+        source=[
+            ReturnOrderStatus.PENDING,
+            ReturnOrderStatus.ON_HOLD,
+            ReturnOrderStatus.IN_PROGRESS,
+        ],
+        target=ReturnOrderStatus.CANCELLED,
+        event=ReturnOrderEvents.CANCELLED,
+    )
+    def cancel_order(self):
+        """Transition this ReturnOrder to CANCELLED status."""
         notify_responsible(
             self,
             ReturnOrder,
@@ -3471,49 +3390,49 @@ class ReturnOrder(TotalPriceMixin, Order):
             extra_users=self.subscribed_users(),
         )
 
-    def _action_complete(self, *args, **kwargs):
-        """Complete this ReturnOrder (if not already completed)."""
-        # Lock this order against concurrent completion, and re-read the status
-        # from the database. Without this, two simultaneous completion requests
-        # can both observe status=IN_PROGRESS, and each would run the completion
-        # side effects (duplicate events and notifications).
-        self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_cancel(self):
+        """Return True if this order can be cancelled."""
+        return can_proceed(self.cancel_order)
 
-        if self.status == ReturnOrderStatus.IN_PROGRESS.value:
-            self.status = ReturnOrderStatus.COMPLETE.value
+    @inventree_transition(
+        field=status,
+        source=ReturnOrderStatus.IN_PROGRESS,
+        target=ReturnOrderStatus.COMPLETE,
+        event=ReturnOrderEvents.COMPLETED,
+    )
+    def complete_order(self):
+        """Transition this ReturnOrder to COMPLETE status.
+
+        The order must currently be IN_PROGRESS.
+        """
         self.complete_date = InvenTree.helpers.current_date()
-            self.save()
-
-            trigger_event(ReturnOrderEvents.COMPLETED, id=self.pk)
 
+    @deprecated('Use issue_order directly', version='1.5.0')
     def place_order(self):
-        """Deprecated version of 'issue_order."""
-        self.issue_order()
+        """Deprecated version of 'issue_order'."""
+        return self.issue_order()
 
     @property
-    def can_issue(self):
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_issue(self) -> bool:
         """Return True if this order can be issued."""
-        return self.status in [
-            ReturnOrderStatus.PENDING.value,
-            ReturnOrderStatus.ON_HOLD.value,
-        ]
+        return can_proceed(self.issue_order)
 
-    def _action_place(self, *args, **kwargs):
-        """Issue this ReturnOrder (if currently pending)."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the issue side effects
-        # (duplicate events and notifications).
-        self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
+    @inventree_transition(
+        field=status,
+        source=[ReturnOrderStatus.PENDING, ReturnOrderStatus.ON_HOLD],
+        target=ReturnOrderStatus.IN_PROGRESS,
+        event=ReturnOrderEvents.ISSUED,
+    )
+    def issue_order(self):
+        """Transition this ReturnOrder to IN_PROGRESS status.
 
-        if self.can_issue:
-            self.status = ReturnOrderStatus.IN_PROGRESS.value
+        The order must currently be PENDING or ON_HOLD.
+        """
         self.issue_date = InvenTree.helpers.current_date()
-            self.save()
 
-            trigger_event(ReturnOrderEvents.ISSUED, id=self.pk)
-
-            # Notify users that the order has been placed
         notify_responsible(
             self,
             ReturnOrder,
@@ -3522,34 +3441,7 @@ class ReturnOrder(TotalPriceMixin, Order):
             extra_users=self.subscribed_users(),
         )
 
-    @transaction.atomic
-    def hold_order(self):
-        """Attempt to transition to ON_HOLD status."""
-        return self.handle_transition(
-            self.status, ReturnOrderStatus.ON_HOLD.value, self, self._action_hold
-        )
-
-    @transaction.atomic
-    def issue_order(self):
-        """Attempt to transition to IN_PROGRESS status."""
-        return self.handle_transition(
-            self.status, ReturnOrderStatus.IN_PROGRESS.value, self, self._action_place
-        )
-
-    @transaction.atomic
-    def complete_order(self):
-        """Attempt to transition to COMPLETE status."""
-        return self.handle_transition(
-            self.status, ReturnOrderStatus.COMPLETE.value, self, self._action_complete
-        )
-
-    @transaction.atomic
-    def cancel_order(self):
-        """Attempt to transition to CANCELLED status."""
-        return self.handle_transition(
-            self.status, ReturnOrderStatus.CANCELLED.value, self, self._action_cancel
-        )
-
+    # endregion fsm
     # endregion
 
     @transaction.atomic
@@ -3928,6 +3820,7 @@ class TransferOrder(Order):
         """Check if this order is "transferred" (all line items transferred)."""
         return all(line.is_completed() for line in self.lines.all())
 
+    # region fsm
     def can_complete(
         self, raise_error: bool = False, allow_incomplete_lines: bool = False
     ) -> bool:
@@ -3957,40 +3850,24 @@ class TransferOrder(Order):
         return True
 
     @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
     def can_issue(self) -> bool:
         """Return True if this order can be issued."""
-        return self.status in [
-            TransferOrderStatus.PENDING.value,
-            TransferOrderStatus.ON_HOLD.value,
-        ]
+        return can_proceed(self.issue_order)
 
-    @transaction.atomic
-    def issue_order(self):
-        """Attempt to transition to PLACED status."""
-        return self.handle_transition(
-            self.status, TransferOrderStatus.ISSUED.value, self, self._action_issue
+    @inventree_transition(
+        field=status,
+        source=[TransferOrderStatus.PENDING, TransferOrderStatus.ON_HOLD],
+        target=TransferOrderStatus.ISSUED,
+        event=TransferOrderEvents.ISSUED,
     )
+    def issue_order(self):
+        """Transition this TransferOrder to ISSUED status.
 
-    # region state changes
-    def _action_issue(self, *args, **kwargs):
-        """Marks the TransferOrder as ISSUED.
-
-        Order must be currently PENDING.
+        The order must currently be PENDING or ON_HOLD.
         """
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the issue side effects
-        # (duplicate events and notifications).
-        self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if self.can_issue:
-            self.status = TransferOrderStatus.ISSUED.value
         self.issue_date = InvenTree.helpers.current_date()
-            self.save()
 
-            trigger_event(TransferOrderEvents.ISSUED, id=self.pk)
-
-            # Notify users that the order has been issued
         notify_responsible(
             self,
             TransferOrder,
@@ -4000,121 +3877,60 @@ class TransferOrder(Order):
         )
 
     @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
     def can_hold(self) -> bool:
         """Return True if this order can be placed on hold."""
-        return self.status in [
-            TransferOrderStatus.PENDING.value,
-            TransferOrderStatus.ISSUED.value,
-        ]
+        return can_proceed(self.hold_order)
 
-    def _action_hold(self, *args, **kwargs):
-        """Mark this transfer order as 'on hold'."""
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an eligible status, and each would run the hold side effects
-        # (duplicate events).
-        self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
+    @inventree_transition(
+        field=status,
+        source=[TransferOrderStatus.PENDING, TransferOrderStatus.ISSUED],
+        target=TransferOrderStatus.ON_HOLD,
+        event=TransferOrderEvents.HOLD,
+    )
+    def hold_order(self):
+        """Transition this TransferOrder to ON_HOLD status."""
 
-        if self.can_hold:
-            self.status = TransferOrderStatus.ON_HOLD.value
-            self.save()
+    @inventree_transition(
+        field=status,
+        source=TransferOrderStatus.ISSUED,
+        target=TransferOrderStatus.COMPLETE,
+        event=TransferOrderEvents.COMPLETED,
+    )
+    def complete_order(self, user=None, **kwargs):
+        """Transition this TransferOrder to COMPLETE status.
 
-            trigger_event(TransferOrderEvents.HOLD, id=self.pk)
-
-    @transaction.atomic
-    def _action_complete(self, *args, **kwargs):
-        """Marks the TransferOrder as COMPLETE.
-
-        Order must be currently ISSUED.
+        The order must currently be ISSUED and meet all completion requirements.
         """
+        if not user:
             user = kwargs.pop('user', None)
 
-        # Lock this order against concurrent completion, and re-read the status
-        # from the database. Without this, two simultaneous completion requests
-        # can both observe status=ISSUED, and each would process every allocation
-        # (duplicating all associated stock operations).
-        self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
+        self.can_complete(raise_error=True, **kwargs)
 
-        if not self.can_complete(raise_error=True, **kwargs):
-            return False
-
-        if self.status == TransferOrderStatus.ISSUED:
         for allocation in self.allocations():
-                # execute each transfer
             allocation.complete_allocation(user)
 
-            self.status = TransferOrderStatus.COMPLETE.value
         self.complete_date = InvenTree.helpers.current_date()
 
-            self.save()
-
-            trigger_event(TransferOrderEvents.COMPLETED, id=self.pk)
-
-            return True
-
-    @transaction.atomic
-    def complete_order(self, user, **kwargs):
-        """Attempt to transition to COMPLETE status."""
-        return self.handle_transition(
-            self.status,
-            TransferOrderStatus.COMPLETE.value,
-            self,
-            self._action_complete,
-            user=user,
-            **kwargs,
+    @inventree_transition(
+        field=status,
+        source=[
+            TransferOrderStatus.PENDING,
+            TransferOrderStatus.ON_HOLD,
+            TransferOrderStatus.ISSUED,
+        ],
+        target=TransferOrderStatus.CANCELLED,
+        event=TransferOrderEvents.CANCELLED,
     )
-
-    @transaction.atomic
-    def hold_order(self):
-        """Attempt to transition to ON_HOLD status."""
-        return self.handle_transition(
-            self.status, TransferOrderStatus.ON_HOLD.value, self, self._action_hold
-        )
-
-    @transaction.atomic
     def cancel_order(self):
-        """Attempt to transition to CANCELLED status."""
-        return self.handle_transition(
-            self.status, TransferOrderStatus.CANCELLED.value, self, self._action_cancel
-        )
+        """Transition this TransferOrder to CANCELLED status.
 
-    @property
-    def can_cancel(self) -> bool:
-        """A TransferOrder can only be cancelled under the following circumstances.
-
-        - Status is ISSUED
-        - Status is PENDING (or ON_HOLD)
+        Deletes all pending stock allocations.
         """
-        return self.status in TransferOrderStatusGroups.OPEN
-
-    def _action_cancel(self, *args, **kwargs):
-        """Cancel this TransferOrder (only if we're allowed to).
-
-        Executes:
-        - Mark the order as 'cancelled'
-        - Delete any StockItems which have been allocated
-        """
-        # Lock this order against concurrent transitions, and re-read the status
-        # from the database. Without this, two simultaneous requests can both
-        # observe an "open" status, and each would run the cancellation side
-        # effects (duplicate events; the allocation deletion itself is
-        # idempotent).
-        self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
-
-        if not self.can_cancel:
-            return False
-
-        self.status = TransferOrderStatus.CANCELLED.value
-        self.save()
-
-        # delete allocations
         for line in self.lines.all():
             for allocation in line.allocations.all():
                 allocation.delete()
 
-        trigger_event(TransferOrderEvents.CANCELLED, id=self.pk)
-
-        # Notify users that the order has been canceled
         notify_responsible(
             self,
             TransferOrder,
@@ -4123,7 +3939,17 @@ class TransferOrder(Order):
             extra_users=self.subscribed_users(),
         )
 
-    # endregion
+    @property
+    @deprecated(Deprecations.CAN_PROCEED, version='1.5.0')
+    def can_cancel(self) -> bool:
+        """A TransferOrder can only be cancelled while it is open.
+
+        - Status is ISSUED
+        - Status is PENDING (or ON_HOLD)
+        """
+        return can_proceed(self.cancel_order)
+
+    # endregion fsm
 
     @property
     def line_count(self) -> int:

src/backend/InvenTree/order/test_api.py

@@ -711,9 +711,10 @@ class PurchaseOrderTest(OrderTest):
         # completion must be skipped based on the database state
         self.assertEqual(po_b.status, PurchaseOrderStatus.PLACED)
 
-        with mock.patch('order.models.trigger_event') as trigger:
+        with self.assertRaises(ValidationError) as err:
             po_b.complete_order()
-            trigger.assert_not_called()
+
+        self.assertIn('Purchase Order is already Complete', str(err.exception))
 
         po.refresh_from_db()
         self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)
@@ -3546,9 +3547,10 @@ class ReturnOrderTests(InvenTreeAPITestCase):
         # completion must be skipped based on the database state
         self.assertEqual(order_b.status, ReturnOrderStatus.IN_PROGRESS.value)
 
-        with mock.patch('order.models.trigger_event') as trigger:
+        with self.assertRaises(ValidationError) as err:
             order_b.complete_order()
-            trigger.assert_not_called()
+
+        self.assertIn('Return Order is already Complete', str(err.exception))
 
         rma.refresh_from_db()
         self.assertEqual(rma.status, ReturnOrderStatus.COMPLETE.value)
@@ -4017,6 +4019,7 @@ class TransferOrderTest(OrderTest):
         self.assertEqual(instance_b.status, TransferOrderStatus.PENDING)
 
         with mock.patch('order.models.trigger_event') as trigger:
+            with self.assertRaises(ValidationError):
                 instance_b.cancel_order()
             trigger.assert_not_called()
 
@@ -4450,12 +4453,22 @@ class TransferOrderTest(OrderTest):
         with self.assertRaises(ValidationError) as err:
             instance_b.complete_order(None)
 
-        self.assertIn('Order is already complete', str(err.exception))
+        self.assertIn('Transfer Order is already Complete', str(err.exception))
 
         # The transferred quantity has not been double-counted
         line.refresh_from_db()
         self.assertEqual(line.transferred, 10)
 
+        # check that the wrong starting point also triggers an error
+        instance_b.status = TransferOrderStatus.CANCELLED.value
+        instance_b.save()
+        with self.assertRaises(ValidationError) as err:
+            instance_b.complete_order(None)
+        self.assertIn(
+            'Invalid transition on Transfer Order.status (source value should be 20, is 40)',
+            str(err.exception),
+        )
+
     def test_output_options(self):
         """Test the output options for the TransferOrder detail endpoint."""
         self.run_output_test(

src/backend/InvenTree/order/test_sales_order.py

@@ -305,8 +305,8 @@ class SalesOrderTest(InvenTreeAPITestCase):
             self.order.can_complete(raise_error=True)
 
         # Now try to ship it - should fail
-        result = self.order.ship_order(None)
-        self.assertFalse(result)
+        with self.assertRaises(ValidationError):
+            self.order.ship_order(None)
 
     def test_order_cancel_stale_instance_is_noop(self):
         """A second cancellation attempt with a stale order instance must be a no-op.
@@ -334,9 +334,8 @@ class SalesOrderTest(InvenTreeAPITestCase):
         # cancellation must be skipped based on the database state
         self.assertEqual(order_b.status, status.SalesOrderStatus.PENDING)
 
-        with mock.patch('order.models.trigger_event') as trigger:
+        with self.assertRaises(ValidationError):
             order_b.cancel_order()
-            trigger.assert_not_called()
 
         self.order.refresh_from_db()
         self.assertEqual(self.order.status, status.SalesOrderStatus.CANCELLED)
@@ -353,9 +352,8 @@ class SalesOrderTest(InvenTreeAPITestCase):
         self.assertEqual(SalesOrderAllocation.objects.count(), 2)
 
         # Attempt to ship the order (but shipments are not completed!)
-        result = self.order.ship_order(None)
-
-        self.assertFalse(result)
+        with self.assertRaises(ValidationError):
+            self.order.ship_order(None)
 
         self.assertIsNone(self.shipment.shipment_date)
         self.assertFalse(self.shipment.is_complete())
@@ -461,9 +459,8 @@ class SalesOrderTest(InvenTreeAPITestCase):
         # completion must be skipped based on the database state
         self.assertEqual(order_b.status, status.SalesOrderStatus.SHIPPED)
 
-        with mock.patch('order.models.trigger_event') as trigger:
+        with self.assertRaises(ValidationError):
             self.assertFalse(order_b.complete_order(None))
-            trigger.assert_not_called()
 
         self.order.refresh_from_db()
         self.assertEqual(self.order.status, status.SalesOrderStatus.COMPLETE)

src/backend/InvenTree/plugin/samples/integration/transition.py

@@ -37,7 +37,7 @@ class SampleTransitionPlugin(TransitionMixin, InvenTreePlugin):
                 msg = 'Return order without responsible owner can not be completed!'
 
                 # Trigger whoever created the return order
-                instance.created_by
+
                 trigger_notification(
                     instance,
                     'sampel_123_456',
@@ -46,7 +46,12 @@ class SampleTransitionPlugin(TransitionMixin, InvenTreePlugin):
                 )
 
                 raise ValidationError(msg)
-
+            if (
+                instance.responsible
+                and instance.responsible.owner.username == 'testuser'
+            ):
+                # If the responsible user is "testuser", we will stop the transition
+                return '123#abc!'
             return False  # Do not act
 
     TRANSITION_HANDLERS = [ReturnChangeHandler()]

src/backend/requirements-3.14.txt

@@ -510,6 +510,7 @@ django==5.2.16 \
     #   django-error-report-2
     #   django-filter
     #   django-flags
+    #   django-fsm-2
     #   django-ical
     #   django-js-asset
     #   django-maintenance-mode
@@ -527,6 +528,7 @@ django==5.2.16 \
     #   django-stdimage
     #   django-storages
     #   django-structlog
+    #   django-stubs-ext
     #   django-taggit
     #   django-xforwardedfor-middleware
     #   djangorestframework
@@ -580,6 +582,12 @@ django-flags==5.2.0 \
     # via
     #   -c src/backend/requirements.txt
     #   -r src/backend/requirements.in
+django-fsm-2==4.2.4 \
+    --hash=sha256:0072be359520075eee4baead3873e5eb8f1700b4f91e41208058cbea0ceae9b9 \
+    --hash=sha256:0e395afd03ff504afd476e8568ca213aafc3324bd4e5427e5b3287f7b1e7d979
+    # via
+    #   -c src/backend/requirements.txt
+    #   -r src/backend/requirements.in
 django-ical==1.9.2 \
     --hash=sha256:44c9b6fa90d09f25e9ebaa91ed9eb007f079afbc23d6aac909cfc18188a8e90c \
     --hash=sha256:74a16bca05735f91a00120cad7250f3c3aa292a9f698a6cfdc544a922c11de70
@@ -703,6 +711,12 @@ django-structlog==10.1.0 \
     # via
     #   -c src/backend/requirements.txt
     #   -r src/backend/requirements.in
+django-stubs-ext==6.0.6 \
+    --hash=sha256:5470c970f61a3ccf5aae7633feef1a097f944f417b955da200c9ac26ecd9134b \
+    --hash=sha256:e6f09884e48d7c5b250a373dfa22aa3e83bb91b8babbd8d05187f6b92247f232
+    # via
+    #   -c src/backend/requirements.txt
+    #   django-fsm-2
 django-taggit==6.1.0 \
     --hash=sha256:ab776264bbc76cb3d7e49e1bf9054962457831bd21c3a42db9138b41956e4cf0 \
     --hash=sha256:c4d1199e6df34125dd36db5eb0efe545b254dec3980ce5dd80e6bab3e78757c3
@@ -2127,6 +2141,8 @@ typing-extensions==4.16.0 \
     --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5
     # via
     #   -c src/backend/requirements.txt
+    #   -r src/backend/requirements.in
+    #   django-stubs-ext
     #   flexcache
     #   flexparser
     #   grpcio

src/backend/requirements-dev-3.14.txt

@@ -414,11 +414,13 @@ django-stubs==6.0.7 \
     # via
     #   -c src/backend/requirements-dev.txt
     #   -r src/backend/requirements-dev.in
-django-stubs-ext==6.0.7 \
-    --hash=sha256:53a9c7c5a7c7e718cc6308cfce1e7470f2cac0b9d38dbcd60fbfa82704f1d592 \
-    --hash=sha256:c3172c5126614fd2a44d0196b313b44c21f717cb09477ba52b447d41f4ce613e
+django-stubs-ext==6.0.6 \
+    --hash=sha256:5470c970f61a3ccf5aae7633feef1a097f944f417b955da200c9ac26ecd9134b \
+    --hash=sha256:e6f09884e48d7c5b250a373dfa22aa3e83bb91b8babbd8d05187f6b92247f232
     # via
+    #   -c src/backend/requirements-3.14.txt
     #   -c src/backend/requirements-dev.txt
+    #   -c src/backend/requirements.txt
     #   django-stubs
 django-test-migrations==1.5.0 \
     --hash=sha256:1cbff04b1e82c5564a6f635284907b381cc11a2ff883adff46776d9126824f07 \

src/backend/requirements-dev.txt

@@ -388,10 +388,12 @@ django-stubs==6.0.7 \
     --hash=sha256:7ed9a14c438e589272ca04e966dee82a4d1ff7ca5c2171bc986c50a0d03ec35b \
     --hash=sha256:bc55431c0af745a64e39cf33a8d36c87dccbedeae2fe26fab47dd355270e8538
     # via -r src/backend/requirements-dev.in
-django-stubs-ext==6.0.7 \
-    --hash=sha256:53a9c7c5a7c7e718cc6308cfce1e7470f2cac0b9d38dbcd60fbfa82704f1d592 \
-    --hash=sha256:c3172c5126614fd2a44d0196b313b44c21f717cb09477ba52b447d41f4ce613e
-    # via django-stubs
+django-stubs-ext==6.0.6 \
+    --hash=sha256:5470c970f61a3ccf5aae7633feef1a097f944f417b955da200c9ac26ecd9134b \
+    --hash=sha256:e6f09884e48d7c5b250a373dfa22aa3e83bb91b8babbd8d05187f6b92247f232
+    # via
+    #   -c src/backend/requirements.txt
+    #   django-stubs
 django-test-migrations==1.5.0 \
     --hash=sha256:1cbff04b1e82c5564a6f635284907b381cc11a2ff883adff46776d9126824f07 \
     --hash=sha256:96a08f085fc8bfaa53d44618341d82a2d22fd194c821cd81b147b66f0bec0da8

src/backend/requirements.in

@@ -9,6 +9,7 @@ django-cors-headers                     # CORS headers extension for DRF
 django-dbbackup                         # Backup / restore of database and media files
 django-error-report-2                   # Error report viewer for the admin interface
 django-filter                           # Extended filtering options
+django-fsm-2                            # Finite state machine for Django models
 django-flags                            # Feature flags
 django-ical                             # iCal export for calendar views
 django-maintenance-mode                 # Shut down application while reloading etc.
@@ -53,6 +54,7 @@ sentry-sdk                              # Error reporting (optional)
 setuptools                              # Standard dependency
 tablib[xls,xlsx,yaml]                   # Support for XLS and XLSX formats
 tqdm                                    # Progress bars for CLI
+typing_extensions                       # new features - for 3, 12 support
 weasyprint                              # PDF generation
 whitenoise                              # Enhanced static file serving
 

src/backend/requirements.txt

@@ -486,6 +486,7 @@ django==5.2.16 \
     #   django-error-report-2
     #   django-filter
     #   django-flags
+    #   django-fsm-2
     #   django-ical
     #   django-js-asset
     #   django-maintenance-mode
@@ -503,6 +504,7 @@ django==5.2.16 \
     #   django-stdimage
     #   django-storages
     #   django-structlog
+    #   django-stubs-ext
     #   django-taggit
     #   django-xforwardedfor-middleware
     #   djangorestframework
@@ -540,6 +542,9 @@ django-flags==5.2.0 \
     --hash=sha256:a3aee5e0d11e24e35f5ae4a423d52cad721f639181c1b1251e0148d8530a3f81 \
     --hash=sha256:aa22543410f68a4082c576f3be469ea67a51c933f8a43eb8b10f6236f158e17e
     # via -r src/backend/requirements.in
+django-fsm-2==4.2.4 \
+    --hash=sha256:0072be359520075eee4baead3873e5eb8f1700b4f91e41208058cbea0ceae9b9
+    # via -r src/backend/requirements.in
 django-ical==1.9.2 \
     --hash=sha256:44c9b6fa90d09f25e9ebaa91ed9eb007f079afbc23d6aac909cfc18188a8e90c \
     --hash=sha256:74a16bca05735f91a00120cad7250f3c3aa292a9f698a6cfdc544a922c11de70
@@ -621,6 +626,10 @@ django-structlog==10.1.0 \
     --hash=sha256:450842bfb03887fc14469570769e515d22ea67d753848a25cdbd05405382c090 \
     --hash=sha256:812f61dea873e559511f42c0f3a7cdc36d4472f6cc15460565a97e5cbe06a76d
     # via -r src/backend/requirements.in
+django-stubs-ext==6.0.6 \
+    --hash=sha256:5470c970f61a3ccf5aae7633feef1a097f944f417b955da200c9ac26ecd9134b \
+    --hash=sha256:e6f09884e48d7c5b250a373dfa22aa3e83bb91b8babbd8d05187f6b92247f232
+    # via django-fsm-2
 django-taggit==6.1.0 \
     --hash=sha256:ab776264bbc76cb3d7e49e1bf9054962457831bd21c3a42db9138b41956e4cf0 \
     --hash=sha256:c4d1199e6df34125dd36db5eb0efe545b254dec3980ce5dd80e6bab3e78757c3
@@ -1883,7 +1892,9 @@ typing-extensions==4.16.0 \
     --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 \
     --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5
     # via
+    #   -r src/backend/requirements.in
     #   django-redis
+    #   django-stubs-ext
     #   flexcache
     #   flexparser
     #   grpcio