Improve typing of transition configuration options (#683)

Changelog.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.9.3 ()
+
+- Bug #683: Typing wrongly suggested that `Transition` instances can be passed to `Machine.__init__` and/or `Machine.add_transition(s)` (thanks @antonio-antuan)
+- Typing should be more precise now
+  - `Machine.add_transitions` and `Machine.__init__` expect a `Sequence` of configurations for transitions now
+  - Added 'async' callbacks to types in `asyncio` extension
+
 ## 0.9.2 (August 2024)
 
 Release 0.9.2 is a minor release and contains a new mermaid diagram backend, a new model decoration mode for easier development with types and some more features and bugfixes.

README.md

@@ -1,11 +1,11 @@
 # <a name="transitions-module"></a> transitions
 
-[![Version](https://img.shields.io/badge/version-v0.9.2-orange.svg)](https://github.com/pytransitions/transitions)
+[![Version](https://img.shields.io/badge/version-v0.9.3-orange.svg)](https://github.com/pytransitions/transitions)
 [![Build Status](https://github.com/pytransitions/transitions/actions/workflows/pytest.yml/badge.svg)](https://github.com/pytransitions/transitions/actions?query=workflow%3Apytest)
 [![Coverage Status](https://coveralls.io/repos/github/pytransitions/transitions/badge.svg?branch=master)](https://coveralls.io/github/pytransitions/transitions?branch=master)
 [![PyPi](https://img.shields.io/pypi/v/transitions.svg)](https://pypi.org/project/transitions)
 [![Copr](https://img.shields.io/badge/dynamic/json?color=blue&label=copr&query=builds.latest.source_package.version&url=https%3A%2F%2Fcopr.fedorainfracloud.org%2Fapi_3%2Fpackage%3Fownername%3Daleneum%26projectname%3Dpython-transitions%26packagename%3Dpython-transitions%26with_latest_build%3DTrue)](https://copr.fedorainfracloud.org/coprs/aleneum/python-transitions/)
-[![GitHub commits](https://img.shields.io/github/commits-since/pytransitions/transitions/0.9.1.svg)](https://github.com/pytransitions/transitions/compare/0.9.1...master)
+[![GitHub commits](https://img.shields.io/github/commits-since/pytransitions/transitions/0.9.2.svg)](https://github.com/pytransitions/transitions/compare/0.9.2...master)
 [![License](https://img.shields.io/github/license/pytransitions/transitions.svg)](LICENSE)
 [![Binder](https://mybinder.org/badge_logo.svg)](https://mybinder.org/v2/gh/pytransitions/transitions/master?filepath=examples%2FPlayground.ipynb)
 

tests/test_async.py

@@ -1,12 +1,12 @@
-from transitions.extensions.asyncio import AsyncMachine, HierarchicalAsyncMachine
 from transitions.extensions.factory import AsyncGraphMachine, HierarchicalAsyncGraphMachine
 
 try:
     import asyncio
+    from transitions.extensions.asyncio import AsyncMachine, HierarchicalAsyncMachine, AsyncEventData
+
 except (ImportError, SyntaxError):
     asyncio = None  # type: ignore
 
-
 from unittest.mock import MagicMock
 from unittest import skipIf
 from functools import partial
@@ -17,7 +17,8 @@
 from .test_pygraphviz import pgv
 
 if TYPE_CHECKING:
-    from typing import Type
+    from typing import Type, Sequence
+    from transitions.extensions.asyncio import AsyncTransitionConfig
 
 
 @skipIf(asyncio is None, "AsyncMachine requires asyncio and contextvars suppport")
@@ -218,6 +219,7 @@ def test_queued(self):
         # Define with list of dictionaries
 
         async def change_state(machine):
+            # type: (AsyncMachine) -> None
             self.assertEqual(machine.state, 'A')
             if machine.has_queue:
                 await machine.run(machine=machine)
@@ -227,20 +229,22 @@ async def change_state(machine):
                     await machine.run(machine=machine)
 
         async def raise_machine_error(event_data):
+            # type: (AsyncEventData) -> None
             self.assertTrue(event_data.machine.has_queue)
             await event_data.model.to_A()
             event_data.machine._queued = False
             await event_data.model.to_C()
 
         async def raise_exception(event_data):
+            # type: (AsyncEventData) -> None
             await event_data.model.to_C()
             raise ValueError("Clears queue")
 
         transitions = [
             {'trigger': 'walk', 'source': 'A', 'dest': 'B', 'before': change_state},
             {'trigger': 'run', 'source': 'B', 'dest': 'C'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
-        ]
+        ]  # type: Sequence[AsyncTransitionConfig]
 
         m = self.machine_cls(states=states, transitions=transitions, initial='A')
         asyncio.run(m.walk(machine=m))
@@ -269,10 +273,12 @@ def check_mock():
         m2 = DummyModel()
 
         async def run():
-            transitions = [{'trigger': 'mock', 'source': ['A', 'B'], 'dest': 'B', 'after': mock},
-                           {'trigger': 'delayed', 'source': 'A', 'dest': 'B', 'before': partial(asyncio.sleep, 0.1)},
-                           {'trigger': 'check', 'source': 'B', 'dest': 'A', 'after': check_mock},
-                           {'trigger': 'error', 'source': 'B', 'dest': 'C', 'before': self.raise_value_error}]
+            transitions = [
+                {'trigger': 'mock', 'source': ['A', 'B'], 'dest': 'B', 'after': mock},
+                {'trigger': 'delayed', 'source': 'A', 'dest': 'B', 'before': partial(asyncio.sleep, 0.1)},
+                {'trigger': 'check', 'source': 'B', 'dest': 'A', 'after': check_mock},
+                {'trigger': 'error', 'source': 'B', 'dest': 'C', 'before': self.raise_value_error}
+            ]  # type: Sequence[AsyncTransitionConfig]
             m = self.machine_cls(model=[m1, m2], states=['A', 'B', 'C'], transitions=transitions, initial='A',
                                  queued='model')
             # call m1.delayed and m2.mock should be called immediately
@@ -307,7 +313,7 @@ def check_queue(expect, event_data):
              'before': partial(check_queue, 4), 'after': remove_model},
             {'trigger': 'remove_queue', 'source': 'B', 'dest': None, 'prepare': ['to_A', 'to_C'],
              'before': partial(check_queue, 3), 'after': remove_model}
-        ]
+        ]  # type: Sequence[AsyncTransitionConfig]
 
         async def run():
             m1 = DummyModel()
@@ -599,13 +605,15 @@ def test_nested_async(self):
         mock = MagicMock()
 
         async def sleep_mock():
+            # type: () -> None
             await asyncio.sleep(0.1)
             mock()
 
         states = ['A', 'B', {'name': 'C', 'children': ['1', {'name': '2', 'children': ['a', 'b'], 'initial': 'a'},
                                                        '3'], 'initial': '2'}]
-        transitions = [{'trigger': 'go', 'source': 'A', 'dest': 'C',
-                        'after': [sleep_mock] * 100}]
+        transitions = [
+            {'trigger': 'go', 'source': 'A', 'dest': 'C', 'after': [sleep_mock] * 100}
+        ]  # type: Sequence[AsyncTransitionConfig]
         machine = self.machine_cls(states=states, transitions=transitions, initial='A')
         asyncio.run(machine.go())
         self.assertEqual('C{0}2{0}a'.format(machine.state_cls.separator), machine.state)

tests/test_core.py

@@ -22,7 +22,8 @@
     from mock import MagicMock  # type: ignore
 
 if TYPE_CHECKING:
-    from typing import List, Union, Dict, Callable
+    from typing import Sequence
+    from transitions.core import TransitionConfig, StateConfig
 
 
 def on_exit_A(event):
@@ -104,7 +105,7 @@ def test_transition_definitions(self):
             {'trigger': 'walk', 'source': 'A', 'dest': 'B'},
             {'trigger': 'run', 'source': 'B', 'dest': 'C'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
-        ]  # type: List[Union[List[str], Dict[str, str]]]
+        ]  # type: Sequence[TransitionConfig]
         m = Machine(states=states, transitions=transitions, initial='A')
         m.walk()
         self.assertEqual(m.state, 'B')
@@ -351,7 +352,7 @@ def test_send_event_data_conditions(self):
         self.assertEqual(s.state, 'B')
 
     def test_auto_transitions(self):
-        states = ['A', {'name': 'B'}, State(name='C')]  # type: List[Union[str, Dict[str, str], State]]
+        states = ['A', {'name': 'B'}, State(name='C')]  # type: Sequence[StateConfig]
         m = Machine(states=states, initial='A', auto_transitions=True)
         m.to_B()
         self.assertEqual(m.state, 'B')
@@ -609,7 +610,7 @@ def change_state(machine):
             {'trigger': 'walk', 'source': 'A', 'dest': 'B', 'before': change_state},
             {'trigger': 'run', 'source': 'B', 'dest': 'C'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
-        ]
+        ]  # type: Sequence[TransitionConfig]
 
         m = Machine(states=states, transitions=transitions, initial='A')
         m.walk(machine=m)
@@ -628,8 +629,11 @@ def after_change(machine):
             machine.to_C(machine)
 
         states = ['A', 'B', 'C']
-        transitions = [{'trigger': 'do', 'source': '*', 'dest': 'C',
-                        'before': partial(self.stuff.this_raises, ValueError)}]
+        transitions = [{
+            'trigger': 'do', 'source': '*', 'dest': 'C',
+            'before': partial(self.stuff.this_raises, ValueError)
+        }]  # type: Sequence[TransitionConfig]
+
         m = Machine(states=states, transitions=transitions, queued=True,
                     before_state_change=before_change, after_state_change=after_change)
         with self.assertRaises(MachineError):
@@ -972,7 +976,7 @@ def always_fails():
             {'trigger': 'go', 'source': 'A', 'dest': 'B', 'conditions': always_fails, 'prepare': local_callback},
             {'trigger': 'go', 'source': 'A', 'dest': 'B', 'prepare': local_callback},
 
-        ]
+        ]  # type: Sequence[TransitionConfig]
         m = Machine(states=['A', 'B'], transitions=transitions,
                     prepare_event=global_callback, initial='A')
 

tests/test_enum.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from unittest import TestCase, skipIf
 
 from transitions.core import Machine
@@ -15,7 +16,8 @@
 from .test_graphviz import pgv as gv
 
 if TYPE_CHECKING:
-    from typing import Type, List, Union, Dict
+    from typing import Type, List, Union, Dict, Sequence
+    from transitions.core import TransitionConfig
 
 
 @skipIf(enum is None, "enum is not available")
@@ -73,7 +75,7 @@ def test_property_initial(self):
         transitions = [
             {'trigger': 'switch_to_yellow', 'source': self.States.RED, 'dest': self.States.YELLOW},
             {'trigger': 'switch_to_green', 'source': 'YELLOW', 'dest': 'GREEN'},
-        ]
+        ]  # type: Sequence[TransitionConfig]
 
         m = self.machine_cls(states=self.States, initial=self.States.RED, transitions=transitions)
         m.switch_to_yellow()
@@ -137,7 +139,7 @@ class State(IntEnum):
         transitions = [
             ['foo', State.FOO, State.BAR],
             ['bar', State.BAR, State.FOO]
-        ]
+        ]  # type: Sequence[List[Union[str, Enum]]]
 
         m = self.machine_cls(states=State, initial=State.FOO, transitions=transitions)
         m.foo()
@@ -418,4 +420,5 @@ class Invalid(enum.Enum):
             INVALID = 1
 
         with self.assertRaises(ValueError):
-            self.machine_cls(states=States, transitions=[['go', '*', Invalid.INVALID]], initial=States.ONE)
+            transitions = [['go', '*', Invalid.INVALID]]  # type: Sequence[List[Union[str, Enum]]]
+            self.machine_cls(states=States, transitions=transitions, initial=States.ONE)

tests/test_experimental.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Sequence, Union, List
 from unittest import TestCase
 from types import ModuleType
 from unittest.mock import MagicMock
@@ -11,7 +11,7 @@
 from .utils import Stuff
 
 if TYPE_CHECKING:
-    from transitions.core import MachineConfig
+    from transitions.core import MachineConfig, TransitionConfig
     from typing import Type
 
 
@@ -137,8 +137,10 @@ class Model:
             def is_B(self) -> bool:
                 return False
 
+            transition_config = [["A", "B"], "C"]  # type: TransitionConfig
+
             @add_transitions(transition(source="A", dest="B"))
-            @add_transitions([["A", "B"], "C"])
+            @add_transitions(transition_config)
             def go(self) -> bool:
                 raise RuntimeError("Should be overridden!")
 
@@ -194,7 +196,8 @@ class Model:
             def is_B(self) -> bool:
                 return False
 
-            go = event(transition(source="A", dest="B"), [["A", "B"], "C"])
+            transition_config = [["A", "B"], "C"]  # type: TransitionConfig
+            go = event(transition(source="A", dest="B"), transition_config)
 
         model = Model()
         machine = self.trigger_machine(model, states=["A", "B", "C"], initial="A")

tests/test_graphviz.py

@@ -19,7 +19,8 @@
     pgv = None
 
 if TYPE_CHECKING:
-    from typing import Type, List, Collection, Union, Literal
+    from typing import Type, List, Collection, Union, Literal, Sequence, Dict, Optional
+    from transitions.core import TransitionConfig
 
 
 class TestDiagramsImport(TestCase):
@@ -74,7 +75,7 @@ def setUp(self):
             {'trigger': 'run', 'source': 'B', 'dest': 'C'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D', 'conditions': 'is_fast'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'B'}
-        ]
+        ]  # type: Sequence[Dict[str, str]]
 
     def test_diagram(self):
         m = self.machine_cls(states=self.states, transitions=self.transitions, initial='A', auto_transitions=False,
@@ -325,7 +326,8 @@ def setUp(self):
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D',    # + 1 edge
              'conditions': 'is_fast'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'B'},   # + 1 edge
-            {'trigger': 'reset', 'source': '*', 'dest': 'A'}]    # + 4 edges (from base state) = 8
+            {'trigger': 'reset', 'source': '*', 'dest': 'A'}     # + 4 edges (from base state) = 8
+        ]  # type: Sequence[Dict[str, str]]
 
     def test_diagram(self):
         m = self.machine_cls(states=self.states, transitions=self.transitions, initial='A', auto_transitions=False,
@@ -419,9 +421,11 @@ def test_roi_parallel_deeper(self):
 
     def test_internal(self):
         states = ['A', 'B']
-        transitions = [['go', 'A', 'B'],
-                       dict(trigger='fail', source='A', dest=None, conditions=['failed']),
-                       dict(trigger='fail', source='A', dest='B', unless=['failed'])]
+        transitions = [
+            ['go', 'A', 'B'],
+            dict(trigger='fail', source='A', dest=None, conditions=['failed']),
+            dict(trigger='fail', source='A', dest='B', unless=['failed'])
+        ]  # type: Sequence[TransitionConfig]
         m = self.machine_cls(states=states, transitions=transitions, initial='A', show_conditions=True,
                              graph_engine=self.graph_engine)
 
@@ -442,7 +446,7 @@ def test_internal_wildcards(self):
             {"trigger": "polled", "source": "ready", "dest": "running", "conditions": "door_closed"},
             ["done", "running", "ready"],
             ["polled", "*", None]
-        ]
+        ]  # type: Sequence[TransitionConfig]
         m = self.machine_cls(states=states, transitions=transitions, show_conditions=True,
                              graph_engine=self.graph_engine, initial='initial')
         _, nodes, edges = self.parse_dot(m.get_graph())

tests/test_markup.py

@@ -28,7 +28,8 @@ class Enum:  # type: ignore
         pass
 
 if TYPE_CHECKING:
-    from typing import List, Dict, Type, Union
+    from typing import List, Dict, Sequence, Union, Type
+    from transitions.core import TransitionConfig, StateConfig
 
 
 class SimpleModel(object):
@@ -96,12 +97,12 @@ class TestMarkupMachine(TestCase):
 
     def setUp(self):
         self.machine_cls = MarkupMachine
-        self.states = ['A', 'B', 'C', 'D']  # type: Union[List[Union[str, Dict]], Type[Enum]]
+        self.states = ['A', 'B', 'C', 'D']  # type: Union[Sequence[StateConfig], Type[Enum]]
         self.transitions = [
             {'trigger': 'walk', 'source': 'A', 'dest': 'B'},
             {'trigger': 'run', 'source': 'B', 'dest': 'C'},
             {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
-        ]  # type: List[Union[str, Dict[str, Union[str, Enum]]]]
+        ]  # type: Sequence[TransitionConfig]
         self.num_trans = len(self.transitions)
         self.num_auto = len(self.states) ** 2
 

tests/test_nesting.py

@@ -1,5 +1,4 @@
 # -*- coding: utf-8 -*-
-
 try:
     from builtins import object
 except ImportError:
@@ -31,8 +30,8 @@
 
 
 if TYPE_CHECKING:
-    from typing import List, Dict, Union, Type
-
+    from typing import List, Dict, Union, Type, Sequence
+    from transitions.core import TransitionConfig
 
 default_separator = NestedState.separator
 
@@ -661,7 +660,7 @@ def test_get_nested_triggers(self):
             ['goA', '*', 'A'],
             ['goF1', ['C{0}1'.format(self.machine_cls.separator), 'C{0}2'.format(self.machine_cls.separator)], 'F'],
             ['goF2', 'C', 'F']
-        ]
+        ]  # type: Sequence[TransitionConfig]
         m = self.machine_cls(states=test_states, transitions=transitions, auto_transitions=False, initial='A')
         self.assertEqual(1, len(m.get_nested_triggers(['C', '1'])))
         with m('C'):