Skip to content
Snippets Groups Projects
Commit 64bad766 authored by Janet Goldstein's avatar Janet Goldstein
Browse files

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

parent 0ecc2aa8
No related branches found
No related tags found
1 merge request!602WS-740: remove last traces of TransitionIF; fix types
Pipeline #3407 passed
......@@ -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.
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment