Skip to content
Snippets Groups Projects

WS-740: remove last traces of TransitionIF; fix types

Merged Janet Goldstein requested to merge WS-740-excise-transition-IF into main
1 file
+ 10
19
Compare changes
  • Side-by-side
  • Inline
@@ -23,7 +23,9 @@ reacts to certain events by triggering actions and going into another state.
"""
import abc
import json
from typing import Optional
from typing import Dict, List, Optional
from workspaces.capability.schema import State, Transition
class State(abc.ABC):
@@ -34,14 +36,14 @@ class State(abc.ABC):
matching pattern, we perform that transition to another state.
"""
def __init__(self, transitions: list["TransitionIF"]):
def __init__(self, transitions: List[Transition]):
# We have a bit of a chicken-and-egg problem here, in that the State needs Transitions to be initialized but
# the Transition needs States to be initialized. Going from prototype to production here will mean breaking
# this cycle, possibly by introducing a builder of some kind, but for now we can just pretend that they are
# built successfully somehow.
self.transitions = transitions
def on_event(self, event: dict) -> Optional["State"]:
def on_event(self, event: Dict) -> Optional[State]:
# Locate the first matching transition
matching_transition = None
for transition in self.transitions:
@@ -73,7 +75,7 @@ class Action(abc.ABC):
class Pattern(abc.ABC):
@abc.abstractmethod
def matches(self, event: dict) -> bool:
def matches(self, event: Dict) -> bool:
"""
This is most likely going to be implemented as a JSON Path expression, using
the jsonpath-python library: https://pypi.org/project/jsonpath-python/
@@ -83,19 +85,8 @@ class Pattern(abc.ABC):
"""
pass
class TransitionIF(abc.ABC):
"""
A transition between states
"""
def __init__(self, from_state: State, to_state: State, pattern: Pattern, action: Action):
self.from_state, self.to_state = from_state, to_state
self.pattern = pattern
self.action = action
@abc.abstractmethod
def matches(self, event: dict) -> bool:
def matches(self, event: Dict) -> bool:
"""
True if this transition is applicable in the supplied state and matches the supplied event.
:param state: state to check against
@@ -126,7 +117,7 @@ class MealyMachine:
self.transitions = []
self.current_state: State = None
def on_event(self, event: dict):
def on_event(self, event: Dict):
"""
Process an event and possibly take a transition.
@@ -157,7 +148,7 @@ class CapabilityInfoForMachines:
that are active and need to be acted on in response to an event of some kind.
"""
def find_requests_matching_transition(self, event: dict) -> list["CapabilityExecution"]:
def find_requests_matching_transition(self, event: Dict) -> List["CapabilityExecution"]:
"""
The concept here is to let the database do the heavy lifting and actually tell us
whether there are any executions that are both in the right state and matching
@@ -219,7 +210,7 @@ class CapabilityServiceMachineMananger:
def __init__(self):
self.info = CapabilityInfoForMachines()
def on_event(self, event: dict):
def on_event(self, event: Dict):
# we get an event, we just try and find matches for it
for execution in self.info.find_requests_matching_transition(event):
# now we have the execution in hand, we can process the event.
Loading