diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 7400b278..bda01a66 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -16,6 +16,39 @@ Changed Contributed by @nzlosh * Added support for python3.10 to 3.12. Contributed by @nzlosh +* Readability-only cleanup pass across the conductor, graph, spec, state-machine + and event layers. These changes make the code easier to follow without + altering any design decision, runtime behavior, or the serialized wire format + that StackStorm persists to Mongo. Specifically: + + * Documented the previously implicit data shapes as type aliases and + ``TypedDict``/``NamedTuple`` definitions in ``statetypes.py`` (``RouteId`` vs + ``RouteDetails``/``RoutesRegistry``, ``TaskTransition``, ``TaskId``, + transition-id conventions), and added module/class/method docstrings and + annotations to ``events.py`` and ``machines.py`` explaining the two state + machines and their ``current status -> {event name -> new status}`` tables. + * Replaced magic positional indexes on workflow-graph edges with a + ``TaskTransition`` named tuple, so the conductor and graph read + ``t.source`` / ``t.destination`` / ``t.key`` / ``t.data`` instead of + ``t[0..3]``. ``TaskTransition`` is a ``tuple`` subclass, so positional + access, unpacking, sorting and set-dedup are unchanged. Likewise + ``extract_vars`` now returns a ``ContextVar`` named tuple. + * Named the recurring literal ``0`` used for the root context/route as + ``constants.ROOT_CONTEXT_INDEX`` / ``ROOT_ROUTE_ID``, and the implicit retry + attempt count as ``DEFAULT_RETRY_COUNT``. + * Added a ``dictionary.first_item`` helper for the single-key-dict spec case, + and switched ``(errors, ctx)`` result handling to named unpacking. + * No copies added or removed: every ``json_util.deepcopy`` call site is + unchanged, and all shared payloads (context dicts, action dicts, edge + ``data`` dicts) remain the same references they were before — preserving the + deliberate no-copy coupling with StackStorm. The changes are allocation + neutral: the spec-rendering paths allocate slightly less (``first_item`` + avoids materializing ``list(dict.items())``; a redundant one-element slice + was removed), offsetting the one extra wrapper tuple per graph edge, whose + inner ``data`` dict is never copied. All 865 unit tests pass unchanged. + * Removed the standalone ``mock`` test dependency in favor of the stdlib + ``unittest.mock``. + * Contributed by guzzijones12@gmail.com Fixed ~~~~~ diff --git a/orquesta/composers/native.py b/orquesta/composers/native.py index 139d25ad..81505a91 100644 --- a/orquesta/composers/native.py +++ b/orquesta/composers/native.py @@ -23,6 +23,10 @@ LOG = logging.getLogger(__name__) +# Default number of attempts for the implicit "retry" task transition (the +# `next: - do: retry` shorthand), used when the spec does not specify a count. +DEFAULT_RETRY_COUNT = 3 + class WorkflowComposer(comp_base.WorkflowComposer): wf_spec_type = native_specs.WorkflowSpec @@ -84,7 +88,10 @@ def _compose_wf_graph(cls, wf_spec): for next_task_name, condition, task_transition_item_idx in next_tasks: if next_task_name == "retry": - retry_spec = {"when": condition or "<% completed() %>", "count": 3} + retry_spec = { + "when": condition or "<% completed() %>", + "count": DEFAULT_RETRY_COUNT, + } wf_graph.update_task(task_name, retry=retry_spec) continue @@ -115,7 +122,7 @@ def _compose_wf_graph(cls, wf_spec): wf_graph.update_transition( task_name, next_task_name, - key=seqs[0][2], + key=seqs[0].key, criteria=crta, ref=task_transition_item_idx, ) diff --git a/orquesta/conducting.py b/orquesta/conducting.py index 6fd97ec8..9ca4043e 100644 --- a/orquesta/conducting.py +++ b/orquesta/conducting.py @@ -13,8 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import logging import queue +import typing from orquesta import constants from orquesta import events @@ -24,6 +27,7 @@ from orquesta import machines from orquesta.specs import base as spec_base from orquesta.specs import loader as spec_loader +from orquesta import statetypes from orquesta import statuses from orquesta.utils import context as ctx_util from orquesta.utils import dictionary as dict_util @@ -37,13 +41,13 @@ class WorkflowState(object): def __init__(self, conductor=None): self.conductor = conductor - self.contexts = list() - self.routes = list() - self.sequence = list() - self.staged = list() - self.status = statuses.UNSET - self.tasks = dict() - self.reruns = list() + self.contexts: list[dict] = list() + self.routes: statetypes.RoutesRegistry = list() + self.sequence: list[statetypes.TaskStateEntry] = list() + self.staged: list[statetypes.StagedTask] = list() + self.status: str = statuses.UNSET + self.tasks: statetypes.TaskIndex = dict() + self.reruns: list[list[int]] = list() def serialize(self): data = { @@ -61,7 +65,7 @@ def serialize(self): return data @classmethod - def deserialize(cls, data): + def deserialize(cls, data: statetypes.SerializedWorkflowState) -> "WorkflowState": instance = cls() instance.contexts = data.get("contexts", list()) instance.routes = data.get("routes", list()) @@ -92,7 +96,7 @@ def get_tasks(self, task_id=None, route=None, last_occurrence=True): result = list(enumerate(self.sequence)) if last_occurrence: - result = [s for s in result if s[0] in self.tasks.values()] + result = [(i, t) for i, t in result if i in self.tasks.values()] return result @@ -102,7 +106,7 @@ def get_tasks_by_status(self, statuses, last_occurrence=True): ] if last_occurrence: - result = [s for s in result if s[0] in self.tasks.values()] + result = [(i, t) for i, t in result if i in self.tasks.values()] return result @@ -178,7 +182,7 @@ def get_unreachable_barriers(self): return unreachable_barriers - def get_staged_tasks(self, filtered=True): + def get_staged_tasks(self, filtered=True) -> list[statetypes.StagedTask]: if not filtered: return self.staged @@ -188,11 +192,13 @@ def get_staged_tasks(self, filtered=True): def has_staged_tasks(self): return len(self.get_staged_tasks()) > 0 - def add_staged_task(self, task_id, route, ctxs=None, prev=None, ready=True, retry=False): + def add_staged_task( + self, task_id, route, ctxs=None, prev=None, ready=True, retry=False + ) -> statetypes.StagedTask: if not ctxs: - ctxs = [0] + ctxs = [constants.ROOT_CONTEXT_INDEX] - entry = { + entry: statetypes.StagedTask = { "id": task_id, "ctxs": {"in": ctxs}, "route": route, @@ -207,7 +213,7 @@ def add_staged_task(self, task_id, route, ctxs=None, prev=None, ready=True, retr return entry - def get_staged_task(self, task_id, route): + def get_staged_task(self, task_id, route) -> typing.Optional[statetypes.StagedTask]: staged_tasks = [x for x in self.staged if x["id"] == task_id and x["route"] == route] return staged_tasks[0] if staged_tasks else None @@ -338,12 +344,13 @@ def workflow_state(self): # Set the initial workflow context. self._workflow_state.contexts.append(init_ctx) - # Set the initial execution route. + # Register the initial (main) route as route id 0. Its + # fingerprint is empty because no splits have been taken yet. self._workflow_state.routes.append([]) # Identify the starting tasks and set the pointer to the initial context entry. for task_node in self.graph.roots: - ctxs, route = [0], 0 + ctxs, route = [constants.ROOT_CONTEXT_INDEX], constants.ROOT_ROUTE_ID self._workflow_state.add_staged_task( task_node["id"], route, ctxs=ctxs, ready=True ) @@ -458,7 +465,7 @@ def request_workflow_status(self, status): raise exc.InvalidWorkflowStatusTransition(current_status, wf_ex_event.name) def get_workflow_initial_context(self): - return self.workflow_state.contexts[0] + return self.workflow_state.contexts[constants.ROOT_CONTEXT_INDEX] def get_workflow_terminal_context(self): if self.get_workflow_status() not in statuses.COMPLETED_STATUSES: @@ -471,7 +478,7 @@ def get_workflow_terminal_context(self): if not term_tasks: return wf_term_ctx - _, first_term_task = term_tasks[0:1][0] + _, first_term_task = term_tasks[0] other_term_tasks = term_tasks[1:] wf_term_ctx = self.get_task_context(first_term_task["ctxs"]["in"]) @@ -518,23 +525,27 @@ def get_inbound_criteria_status(self, task_id, route): # Get the list of inbound task transitions for the barrier task. inbound_transitions = self.graph.get_prev_transitions(task_id) - # Setup the result for the evaluation of the criteria for inbound task transitions. - inbound_evaluation = {i: None for i in list(set(t[0] for t in inbound_transitions))} + # Setup the result for the evaluation of the criteria for inbound task + # transitions, keyed by source task id. + inbound_evaluation = {i: None for i in list(set(t.source for t in inbound_transitions))} - # Identify the join requirement. + # Identify the join requirement: how many inbound branches must satisfy their criteria. + # A barrier of "*" (or an unset barrier, which defaults to 1) means "all inbound branches". barrier = self.graph.get_barrier(task_id) or 1 requirement = len(inbound_evaluation.keys()) if barrier == "*" else barrier # Evaluate the criteria for each inbound task transitions. for prev_transition in inbound_transitions: - prev_task_state_entry = self.get_task_state_entry(prev_transition[0], route) + prev_task_state_entry = self.get_task_state_entry(prev_transition.source, route) if not prev_task_state_entry: continue + # The transition id recorded on the source task's "next" is built from + # the destination (this barrier task) and the transition key. prev_task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % ( - prev_transition[1], - str(prev_transition[2]), + prev_transition.destination, + str(prev_transition.key), ) satisfied = ( @@ -542,8 +553,8 @@ def get_inbound_criteria_status(self, task_id, route): and prev_task_state_entry["next"][prev_task_transition_id] ) - if not bool(inbound_evaluation[prev_transition[0]]): - inbound_evaluation[prev_transition[0]] = satisfied + if not bool(inbound_evaluation[prev_transition.source]): + inbound_evaluation[prev_transition.source] = satisfied # If the count of inbound task(s) where the criteria is True >= requirements, # then the join requirement is satisified. @@ -562,7 +573,7 @@ def get_inbound_criteria_status(self, task_id, route): # If reached here, then the requirement is not satisified. return constants.INBOUND_CRITERIA_NOT_SATISFIED - def get_task(self, task_id, route): + def get_task(self, task_id, route) -> statetypes.RuntimeTask: try: task_ctx = self.get_task_initial_context(task_id, route) except ValueError: @@ -575,7 +586,7 @@ def get_task(self, task_id, route): task_spec = self.spec.tasks.get_task(task_id).copy() task_spec, action_specs = task_spec.render(task_ctx) - task = { + task: statetypes.RuntimeTask = { "id": task_id, "route": route, "ctx": task_ctx, @@ -604,7 +615,7 @@ def get_task(self, task_id, route): return task - def _evaluate_task_actions(self, task): + def _evaluate_task_actions(self, task: statetypes.RuntimeTask) -> statetypes.RuntimeTask: task_id = task["id"] task_route = task["route"] @@ -619,21 +630,27 @@ def _evaluate_task_actions(self, task): if "items" not in staged_task or not staged_task["items"]: staged_task["items"] = [{"status": statuses.UNSET}] * task["items_count"] - # Trim the list of actions in the task per concurrency policy. - all_items = list(zip(task["actions"], staged_task["items"])) - notrun_items = list(filter(lambda x: x[1]["status"] == statuses.UNSET, all_items)) - active_items = list(filter(lambda x: x[1]["status"] in statuses.ACTIVE_STATUSES, all_items)) + # Trim the list of actions in the task per concurrency policy. Each entry pairs an + # action with its item-execution state so items can be filtered by status. + action_item_pairs = list(zip(task["actions"], staged_task["items"])) + notrun_items = [ + (action, item) for action, item in action_item_pairs if item["status"] == statuses.UNSET + ] + active_items = [ + (action, item) + for action, item in action_item_pairs + if item["status"] in statuses.ACTIVE_STATUSES + ] if task["concurrency"] is not None: # Concurrency below 1 prevents scheduling of tasks. if task["concurrency"] <= 0: task["concurrency"] = 1 availability = task["concurrency"] - len(active_items) - candidates = list(zip(*notrun_items[:availability])) - task["actions"] = list(candidates[0]) if candidates and availability > 0 else [] + schedulable = notrun_items[:availability] if availability > 0 else [] + task["actions"] = [action for action, item in schedulable] else: - candidates = list(zip(*notrun_items)) - task["actions"] = list(candidates[0]) if candidates else [] + task["actions"] = [action for action, item in notrun_items] return task @@ -648,12 +665,12 @@ def _has_next(self, task_id, route=None, eval_join_ready=True): outbounds = self.graph.get_next_transitions(task_id) - for next_seq in outbounds: - next_task_id, seq_key = next_seq[1], next_seq[2] + for transition in outbounds: + next_task_id = transition.destination task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % ( next_task_id, - str(seq_key), + str(transition.key), ) # Ignore if the next task is the engine command to "continue". @@ -737,7 +754,7 @@ def _get_task_state_idx(self, task_id, route): constants.TASK_STATE_ROUTE_FORMAT % (task_id, str(route)) ) - def get_task_state_entry(self, task_id, route): + def get_task_state_entry(self, task_id, route) -> typing.Optional[statetypes.TaskStateEntry]: task_state_seq_idx = self._get_task_state_idx(task_id, route) if task_state_seq_idx is None: @@ -807,14 +824,16 @@ def setup_retry_in_task_state(self, task_state_entry, in_ctx_idxs): task_state_entry["retry"]["count"] = count_value - def add_task_state(self, task_id, route, in_ctx_idxs=None, prev=None): + def add_task_state( + self, task_id, route, in_ctx_idxs=None, prev=None + ) -> statetypes.TaskStateEntry: if not self.graph.has_task(task_id): raise exc.InvalidTask(task_id) if not in_ctx_idxs: - in_ctx_idxs = [0] + in_ctx_idxs = [constants.ROOT_CONTEXT_INDEX] - task_state_entry = { + task_state_entry: statetypes.TaskStateEntry = { "id": task_id, "route": route, "ctxs": {"in": in_ctx_idxs}, @@ -964,15 +983,16 @@ def update_task_state(self, task_id, route, event): # Iterate thru each outbound task transitions. for task_transition in task_transitions: + # A TaskTransition (source, destination, key, data); see statetypes. task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % ( - task_transition[1], - str(task_transition[2]), + task_transition.destination, + str(task_transition.key), ) # Evaluate the criteria for task transition. If there is a failure while # evaluating expression(s), fail the workflow. try: - criteria = task_transition[3].get("criteria") or [] + criteria = task_transition.data.get("criteria") or [] evaluated_criteria = [expr_base.evaluate(c, current_ctx) for c in criteria] task_state_entry["next"][task_transition_id] = all(evaluated_criteria) except Exception as e: @@ -982,7 +1002,7 @@ def update_task_state(self, task_id, route, event): # If criteria met, then mark the next task staged and calculate outgoing context. if task_state_entry["next"][task_transition_id]: - next_task_node = self.graph.get_task(task_transition[1]) + next_task_node = self.graph.get_task(task_transition.destination) next_task_id = next_task_node["id"] new_ctx_idx = None @@ -1018,9 +1038,11 @@ def update_task_state(self, task_id, route, event): next_task_id, next_task_route ) + # The backref id embeds the *source* (current) task and the + # transition key, and is stored on the next task's "prev". backref = constants.TASK_STATE_TRANSITION_FORMAT % ( task_id, - str(task_transition[2]), + str(task_transition.key), ) # If the next task is already staged. @@ -1097,29 +1119,47 @@ def update_task_state(self, task_id, route, event): return task_state_entry - def _evaluate_route(self, task_transition, prev_route): - task_id = task_transition[1] - + def _evaluate_route( + self, task_transition: statetypes.TaskTransition, prev_route: statetypes.RouteId + ) -> statetypes.RouteId: + """Determine which route the task on the far side of a transition belongs to. + + Given an outbound ``task_transition`` and the ``prev_route`` (route id) + the current task ran on, return the route id for the next task. See the + "Routing" section in :mod:`orquesta.statetypes` for the route model. + + The next task stays on ``prev_route`` unless the transition forks a new + branch, i.e. the next task is a split task and is not inside a cycle. + When it does fork, the next task's route "fingerprint" is the previous + route's fingerprint plus this transition id. If that fingerprint is new, + a new route is registered and its id returned; if an identical + fingerprint already applies, the existing ``prev_route`` is reused. + """ + # The next (destination) task is what may fork onto a new route. prev_task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % ( - task_transition[0], - str(task_transition[2]), + task_transition.source, + str(task_transition.key), ) - is_split_task = self.spec.tasks.is_split_task(task_id) - is_in_cycle = self.graph.in_cycle(task_id) + is_split_task = self.spec.tasks.is_split_task(task_transition.destination) + is_in_cycle = self.graph.in_cycle(task_transition.destination) + # Not a fork: the next task continues on the same route. if not is_split_task or is_in_cycle: return prev_route + # Build the candidate fingerprint for the forked branch. old_route_details = self.workflow_state.routes[prev_route] new_route_details = json_util.deepcopy(old_route_details) if prev_task_transition_id not in old_route_details: new_route_details.append(prev_task_transition_id) + # Fingerprint unchanged: this split was already accounted for, reuse it. if old_route_details == new_route_details: return prev_route + # New branch: register it and return its route id (its index). self.workflow_state.routes.append(new_route_details) return len(self.workflow_state.routes) - 1 @@ -1172,14 +1212,19 @@ def get_task_transition_contexts(self, task_id, route): if not task_state_entry: raise exc.InvalidTaskStateEntry(task_id) - for t in self.graph.get_next_transitions(task_id): - task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % (t[1], str(t[2])) + for transition in self.graph.get_next_transitions(task_id): + task_transition_id = constants.TASK_STATE_TRANSITION_FORMAT % ( + transition.destination, + str(transition.key), + ) if ( task_transition_id in task_state_entry["next"] and task_state_entry["next"][task_transition_id] ): - contexts[task_transition_id] = self.get_task_initial_context(t[1], route) + contexts[task_transition_id] = self.get_task_initial_context( + transition.destination, route + ) return contexts @@ -1223,7 +1268,7 @@ def _collapse_task_rerun_requests(self, tasks=None): # The method get_task_sequence returns the index and the dictionary for the task entry. # Only the index is required for further evaluation below. result = { - k: [i[0] for i in self.workflow_state.get_task_sequence(t.task_id, t.route)] + k: [idx for idx, entry in self.workflow_state.get_task_sequence(t.task_id, t.route)] for k, t in tasks.items() } diff --git a/orquesta/constants.py b/orquesta/constants.py index 79153c89..d6833313 100644 --- a/orquesta/constants.py +++ b/orquesta/constants.py @@ -15,6 +15,13 @@ TASK_STATE_ROUTE_FORMAT = "%s__r%s" TASK_STATE_TRANSITION_FORMAT = "%s__t%s" +# The initial (main) workflow context and route. When a workflow starts, the +# root context is registered at this index in WorkflowState.contexts and the +# main route at this id in WorkflowState.routes. The literal 0 is referenced +# from many places in the conductor; these names make that intent explicit. +ROOT_CONTEXT_INDEX = 0 +ROOT_ROUTE_ID = 0 + INBOUND_CRITERIA_WIP = "inbound_criteria_wip" INBOUND_CRITERIA_SATISFIED = "inbound_criteria_satisfied" INBOUND_CRITERIA_NOT_SATISFIED = "inbound_criteria_not_satisfied" diff --git a/orquesta/events.py b/orquesta/events.py index 5a1c09f1..a78ffa8e 100644 --- a/orquesta/events.py +++ b/orquesta/events.py @@ -12,9 +12,35 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Execution events and the event-name vocabulary that drive the state machines. + +Orquesta is event driven: the conductor turns things that happen (an action +finished, a pause was requested, ...) into :class:`ExecutionEvent` instances and +feeds them to the state machines in :mod:`orquesta.machines`, which look up the +event ``name`` in their transition tables to decide the next status. + +There are three families of event *names* (the module-level string constants), +plus engine-operation events: + +* ``WORKFLOW_*`` / ``WORKFLOW_EXECUTION_EVENTS`` -- workflow-level status events. +* ``TASK_*`` / ``TASK_EXECUTION_EVENTS`` -- task-level status events. The subset + in ``TASK_CONDITIONAL_EVENTS`` needs extra workflow context to be processed. +* ``ACTION_*`` / ``ACTION_EXECUTION_EVENTS`` -- action-execution status events, + including the with-items ``..._TASK_[ACTIVE|DORMANT]_ITEMS_*`` variants. +* ``ENGINE_OPERATION_EVENTS`` -- synthetic events for the engine commands + (continue/fail/noop/retry); see ``ENGINE_EVENT_MAP``. + +The classes near the bottom of this module are the runtime carriers of those +names plus their payload (status/result/context/etc.). +""" + +from __future__ import annotations + import logging +import typing from orquesta import exceptions as exc +from orquesta import statetypes from orquesta import statuses @@ -318,29 +344,46 @@ class ExecutionEvent(object): - def __init__(self, name, status, result=None, context=None): + """Base class for events fed to the workflow and task state machines. + + Attributes: + name: the event name; matched against the state-machine transition + tables (one of the ``*_EVENTS`` constants above, possibly with a + context suffix added by the machine). + status: the status being reported by this event. + result: the action result, when the event carries one. + context: extra workflow context attached to the event, when relevant. + """ + + def __init__(self, name: str, status: str, result: typing.Any = None, context: dict = None): if not statuses.is_valid(status): raise exc.InvalidStatus(status) - self.name = name - self.status = status - self.result = result - self.context = context + self.name: str = name + self.status: str = status + self.result: typing.Any = result + self.context: typing.Optional[dict] = context class WorkflowExecutionEvent(ExecutionEvent): + """A workflow-level status event, e.g. request to pause/cancel/resume.""" + def __init__(self, status): super(WorkflowExecutionEvent, self).__init__("workflow_%s" % status, status) class TaskExecutionEvent(ExecutionEvent): - def __init__(self, task_id, route, status): + """A task-level status event, identifying the task by id and route.""" + + def __init__(self, task_id: statetypes.TaskId, route: statetypes.RouteId, status: str): super(TaskExecutionEvent, self).__init__("task_%s" % status, status) - self.task_id = task_id - self.route = route + self.task_id: statetypes.TaskId = task_id + self.route: statetypes.RouteId = route class ActionExecutionEvent(ExecutionEvent): + """An action-execution status event (the result of running one action).""" + def __init__(self, status, result=None, context=None): super(ActionExecutionEvent, self).__init__( "action_%s" % status, status, result=result, context=context @@ -348,13 +391,27 @@ def __init__(self, status, result=None, context=None): class TaskItemActionExecutionEvent(ActionExecutionEvent): - def __init__(self, item_id, status, result=None, accumulated_result=None): + """An action-execution event for a single item of a with-items task. + + Attributes: + item_id: index of the item within the with-items task. + accumulated_result: results accumulated across items so far. + """ + + def __init__(self, item_id: int, status, result=None, accumulated_result=None): super(TaskItemActionExecutionEvent, self).__init__(status, result=result) - self.item_id = item_id + self.item_id: int = item_id self.accumulated_result = accumulated_result class EngineOperationEvent(ExecutionEvent): + """Base class for synthetic events produced by engine commands. + + Unlike the other events, these are not reported by an external actor; the + conductor emits them when a task transition targets an engine command (see + ``ENGINE_EVENT_MAP``). Each subclass hardcodes the name/status it carries. + """ + pass diff --git a/orquesta/expressions/base.py b/orquesta/expressions/base.py index 055a90e9..765734ed 100644 --- a/orquesta/expressions/base.py +++ b/orquesta/expressions/base.py @@ -20,6 +20,7 @@ import logging import re import threading +import typing from stevedore import extension @@ -34,6 +35,19 @@ _EXP_EVALUATOR_NAMESPACE = "orquesta.expressions.evaluators" +class ContextVar(typing.NamedTuple): + """A context variable reference extracted from an expression. + + Returned by :func:`extract_vars`. It is a ``tuple`` subclass, so existing + positional access, unpacking, sorting, and set-dedup all behave as before; + the named fields simply make ``[0]``/``[1]``/``[2]`` self-documenting. + """ + + type: str # the evaluator type that produced it, e.g. "jinja" / "yaql" + expression: str # the full expression string the variable was found in + name: str # the referenced variable name + + class Evaluator(metaclass=abc.ABCMeta): _type = "unspecified" _delimiter = None @@ -163,11 +177,11 @@ def extract_vars(statement): for regex_var_extract in evaluator.get_var_extraction_regexes(): result = re.search(regex_var_extract, var_ref) var = result.group(1) if result else "" - variables.append((evaluator.get_type(), statement, var)) + variables.append(ContextVar(evaluator.get_type(), statement, var)) - variables = [v for v in variables if v[2] != ""] + variables = [v for v in variables if v.name != ""] - return sorted(list(set(variables)), key=lambda var: var[2]) + return sorted(list(set(variables)), key=lambda var: var.name) def func_has_ctx_arg(func): diff --git a/orquesta/graphing.py b/orquesta/graphing.py index 4d39f6e4..9d70d933 100644 --- a/orquesta/graphing.py +++ b/orquesta/graphing.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import abc import logging @@ -19,6 +21,7 @@ from networkx.readwrite import json_graph from orquesta import exceptions as exc +from orquesta import statetypes from orquesta.utils import dictionary as dict_util from orquesta.utils import jsonify as json_util @@ -101,7 +104,7 @@ def update_task(self, task_id, **kwargs): for key, value in kwargs.items(): self._graph.nodes[task_id][key] = value - def has_transition(self, source, destination, **kwargs): + def has_transition(self, source, destination, **kwargs) -> list[statetypes.TaskTransition]: edges = filter( lambda e: e[0] == source and e[1] == destination, self._graph.edges(data=True, keys=True), # pylint: disable=unexpected-keyword-arg @@ -110,9 +113,9 @@ def has_transition(self, source, destination, **kwargs): for attr, value in kwargs.items(): edges = filter(lambda e: e[3].get(attr, None) == value, list(edges)) - return list(edges) + return [statetypes.TaskTransition(*e) for e in edges] - def get_transition(self, source, destination, key=None, **kwargs): + def get_transition(self, source, destination, key=None, **kwargs) -> statetypes.TaskTransition: if key is not None: edges = filter( lambda e: e[0] == source and e[1] == destination and e[2] == key, @@ -135,7 +138,7 @@ def get_transition(self, source, destination, key=None, **kwargs): if len(edges) > 1: raise exc.AmbiguousTaskTransition(source, destination) - return edges[0] + return statetypes.TaskTransition(*edges[0]) def get_transition_attributes(self, attribute): return nx.get_edge_attributes(self._graph, attribute) @@ -160,16 +163,24 @@ def update_transition(self, source, destination, key, **kwargs): seq = self.get_transition(source, destination, key=key) for attr, value in kwargs.items(): - self._graph[source][destination][seq[2]][attr] = value + self._graph[source][destination][seq.key][attr] = value - def get_next_transitions(self, task_id): + def get_next_transitions(self, task_id) -> list[statetypes.TaskTransition]: return sorted( - [e for e in self._graph.out_edges([task_id], data=True, keys=True)], key=lambda x: x[1] + [ + statetypes.TaskTransition(*e) + for e in self._graph.out_edges([task_id], data=True, keys=True) + ], + key=lambda x: x.destination, ) - def get_prev_transitions(self, task_id): + def get_prev_transitions(self, task_id) -> list[statetypes.TaskTransition]: return sorted( - [e for e in self._graph.in_edges([task_id], data=True, keys=True)], key=lambda x: x[1] + [ + statetypes.TaskTransition(*e) + for e in self._graph.in_edges([task_id], data=True, keys=True) + ], + key=lambda x: x.destination, ) def get_barriers(self): @@ -202,7 +213,7 @@ def is_cycle_closed(self, cycle): # transition to any task that is not a member of the cycle. for task_id in cycle["tasks"]: for transition in self.get_next_transitions(task_id): - if transition[1] not in cycle["tasks"]: + if transition.destination not in cycle["tasks"]: return False return True diff --git a/orquesta/machines.py b/orquesta/machines.py index 8657eec6..77b642c7 100644 --- a/orquesta/machines.py +++ b/orquesta/machines.py @@ -13,17 +13,52 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""The two state machines that decide status transitions from events. + +The conductor feeds :class:`orquesta.events.ExecutionEvent` instances to two +state machines defined here: + +* :class:`WorkflowStateMachine` -- owns the *workflow's* status. It reacts to + workflow-level events and to task-level events (a task finishing can complete + or fail the whole workflow). +* :class:`TaskStateMachine` -- owns a single *task's* status. It reacts to + action-execution events, with-items events, and workflow-level events. + +Both are driven by a lookup table (``*_STATE_MACHINE_DATA`` below). A table maps +``current status -> {event name -> new status}``: given where you are and what +happened, it tells you where to go. If the current status has no entry for the +event, the status is left unchanged. + +The ``add_context_to_*`` methods are the reason the event *name* fed to the +table is not always the raw ``event.name``: they append suffixes such as +``_workflow_active`` / ``_task_dormant_items_completed`` derived from the rest +of the workflow state, so a single raw event can resolve to different +transitions depending on context. Those expanded names are exactly the +``events.*_EXECUTION_EVENTS`` constants used as table keys. +""" + +from __future__ import annotations + import logging +import typing from orquesta import events from orquesta import exceptions as exc +from orquesta import statetypes from orquesta import statuses +if typing.TYPE_CHECKING: + from orquesta import conducting + LOG = logging.getLogger(__name__) -WORKFLOW_STATE_MACHINE_DATA = { +# A state-machine transition table: current status -> {event name -> new status}. +StateTransitionTable = typing.Dict[str, typing.Dict[str, str]] + + +WORKFLOW_STATE_MACHINE_DATA: StateTransitionTable = { statuses.UNSET: { events.WORKFLOW_REQUESTED: statuses.REQUESTED, events.WORKFLOW_SCHEDULED: statuses.SCHEDULED, @@ -225,7 +260,7 @@ } -TASK_STATE_MACHINE_DATA = { +TASK_STATE_MACHINE_DATA: StateTransitionTable = { statuses.UNSET: { events.ACTION_REQUESTED: statuses.REQUESTED, events.ACTION_SCHEDULED: statuses.SCHEDULED, @@ -450,8 +485,17 @@ class TaskStateMachine(object): + """Decides a single task's status from the events it receives. + + All methods are class methods; the machine is stateless. It mutates the + ``task_state`` entry (a :class:`statetypes.TaskStateEntry`) in place by + writing its ``status`` key, using ``TASK_STATE_MACHINE_DATA`` as the table. + """ + @classmethod - def is_transition_valid(cls, old_status, new_status): + def is_transition_valid( + cls, old_status: typing.Optional[str], new_status: typing.Optional[str] + ) -> bool: if old_status is None: old_status = "null" @@ -473,11 +517,22 @@ def is_transition_valid(cls, old_status, new_status): return False @classmethod - def add_context_to_action_event(cls, workflow_state, task_id, task_route, ac_ex_event): + def add_context_to_action_event( + cls, + workflow_state: "conducting.WorkflowState", + task_id: statetypes.TaskId, + task_route: statetypes.RouteId, + ac_ex_event: events.ActionExecutionEvent, + ) -> str: return ac_ex_event.name @classmethod - def process_action_event(cls, workflow_state, task_state, ac_ex_event): + def process_action_event( + cls, + workflow_state: "conducting.WorkflowState", + task_state: statetypes.TaskStateEntry, + ac_ex_event: events.ActionExecutionEvent, + ) -> None: # Check if event is valid. if ac_ex_event.name not in events.ACTION_EXECUTION_EVENTS + events.ENGINE_OPERATION_EVENTS: raise exc.InvalidEvent(ac_ex_event.name) @@ -509,7 +564,13 @@ def process_action_event(cls, workflow_state, task_state, ac_ex_event): task_state["status"] = new_task_status @classmethod - def add_context_to_task_item_event(cls, workflow_state, task_id, task_route, ac_ex_event): + def add_context_to_task_item_event( + cls, + workflow_state: "conducting.WorkflowState", + task_id: statetypes.TaskId, + task_route: statetypes.RouteId, + ac_ex_event: events.TaskItemActionExecutionEvent, + ) -> str: action_event = ac_ex_event.name requirements = [ @@ -563,7 +624,12 @@ def add_context_to_task_item_event(cls, workflow_state, task_id, task_route, ac_ return action_event @classmethod - def process_task_item_event(cls, workflow_state, task_state, ac_ex_event): + def process_task_item_event( + cls, + workflow_state: "conducting.WorkflowState", + task_state: statetypes.TaskStateEntry, + ac_ex_event: events.TaskItemActionExecutionEvent, + ) -> None: # Check if event is valid. if ac_ex_event.name not in events.ACTION_EXECUTION_EVENTS + events.ENGINE_OPERATION_EVENTS: raise exc.InvalidEvent(ac_ex_event.name) @@ -595,7 +661,13 @@ def process_task_item_event(cls, workflow_state, task_state, ac_ex_event): task_state["status"] = new_task_status @classmethod - def add_context_to_workflow_event(cls, workflow_state, task_id, task_route, wf_ex_event): + def add_context_to_workflow_event( + cls, + workflow_state: "conducting.WorkflowState", + task_id: statetypes.TaskId, + task_route: statetypes.RouteId, + wf_ex_event: events.WorkflowExecutionEvent, + ) -> str: workflow_event = wf_ex_event.name requirements = statuses.PAUSE_STATUSES + statuses.CANCEL_STATUSES staged_task = workflow_state.get_staged_task(task_id, task_route) @@ -610,7 +682,12 @@ def add_context_to_workflow_event(cls, workflow_state, task_id, task_route, wf_e return workflow_event @classmethod - def process_workflow_event(cls, workflow_state, task_state, wf_ex_event): + def process_workflow_event( + cls, + workflow_state: "conducting.WorkflowState", + task_state: statetypes.TaskStateEntry, + wf_ex_event: events.WorkflowExecutionEvent, + ) -> None: # Check if event is valid. if wf_ex_event.name not in events.WORKFLOW_EXECUTION_EVENTS: raise exc.InvalidEvent(wf_ex_event.name) @@ -642,7 +719,12 @@ def process_workflow_event(cls, workflow_state, task_state, wf_ex_event): task_state["status"] = new_task_status @classmethod - def process_event(cls, workflow_state, task_state, event): + def process_event( + cls, + workflow_state: "conducting.WorkflowState", + task_state: statetypes.TaskStateEntry, + event: events.ExecutionEvent, + ) -> None: if isinstance(event, events.WorkflowExecutionEvent): cls.process_workflow_event(workflow_state, task_state, event) return @@ -663,8 +745,19 @@ def process_event(cls, workflow_state, task_state, event): class WorkflowStateMachine(object): + """Decides the workflow's status from the events it receives. + + All methods are class methods; the machine is stateless. It mutates + ``workflow_state.status`` in place, using ``WORKFLOW_STATE_MACHINE_DATA`` as + the table. Unlike :class:`TaskStateMachine`, it also reacts to task-level + events, since a task reaching a terminal status can complete or fail the + whole workflow. + """ + @classmethod - def is_transition_valid(cls, old_status, new_status): + def is_transition_valid( + cls, old_status: typing.Optional[str], new_status: typing.Optional[str] + ) -> bool: if old_status is None: old_status = "null" @@ -689,7 +782,11 @@ def is_transition_valid(cls, old_status, new_status): return False @classmethod - def add_context_to_task_event(cls, workflow_state, tk_ex_event): + def add_context_to_task_event( + cls, + workflow_state: "conducting.WorkflowState", + tk_ex_event: events.TaskExecutionEvent, + ) -> str: # Identify current workflow status. task_event = tk_ex_event.name task_id = getattr(tk_ex_event, "task_id", None) @@ -731,7 +828,11 @@ def add_context_to_task_event(cls, workflow_state, tk_ex_event): return task_event + "_completed" @classmethod - def process_task_event(cls, workflow_state, tk_ex_event): + def process_task_event( + cls, + workflow_state: "conducting.WorkflowState", + tk_ex_event: events.TaskExecutionEvent, + ) -> None: # Append additional workflow context to the event. event_name = cls.add_context_to_task_event(workflow_state, tk_ex_event) @@ -775,7 +876,11 @@ def process_task_event(cls, workflow_state, tk_ex_event): workflow_state.conductor.log_error(e, task_id=entry["id"], route=entry["route"]) @classmethod - def add_context_to_workflow_event(cls, workflow_state, wf_ex_event): + def add_context_to_workflow_event( + cls, + workflow_state: "conducting.WorkflowState", + wf_ex_event: events.WorkflowExecutionEvent, + ) -> str: # Identify current workflow status. workflow_event = wf_ex_event.name has_active_tasks = workflow_state.has_active_tasks @@ -800,7 +905,11 @@ def add_context_to_workflow_event(cls, workflow_state, wf_ex_event): return workflow_event @classmethod - def process_workflow_event(cls, workflow_state, wf_ex_event): + def process_workflow_event( + cls, + workflow_state: "conducting.WorkflowState", + wf_ex_event: events.WorkflowExecutionEvent, + ) -> None: # Append additional workflow context to the event. event_name = cls.add_context_to_workflow_event(workflow_state, wf_ex_event) @@ -828,7 +937,11 @@ def process_workflow_event(cls, workflow_state, wf_ex_event): workflow_state.status = new_workflow_status @classmethod - def process_event(cls, workflow_state, event): + def process_event( + cls, + workflow_state: "conducting.WorkflowState", + event: events.ExecutionEvent, + ) -> None: if isinstance(event, events.WorkflowExecutionEvent): cls.process_workflow_event(workflow_state, event) return diff --git a/orquesta/specs/base.py b/orquesta/specs/base.py index f1a31e96..803fb319 100644 --- a/orquesta/specs/base.py +++ b/orquesta/specs/base.py @@ -443,9 +443,9 @@ def inspect_context(self, parent=None): def decorate_ctx_var(variable, spec_path, schema_path): return { - "type": variable[0], - "expression": variable[1], - "name": variable[2], + "type": variable.type, + "expression": variable.expression, + "name": variable.name, "spec_path": spec_path, "schema_path": schema_path, } @@ -520,9 +520,9 @@ def inspect_ctx(prop_name, prop_value, spec_path, schema_path, rolling_ctx, erro "schema_path": schema_path, } - result = prop_value.inspect_context(parent=item_parent) - errors.extend(result[0]) - rolling_ctx = list(set(rolling_ctx + result[1])) + result_errors, result_ctx = prop_value.inspect_context(parent=item_parent) + errors.extend(result_errors) + rolling_ctx = list(set(rolling_ctx + result_ctx)) continue diff --git a/orquesta/specs/native/v1/models.py b/orquesta/specs/native/v1/models.py index 6dff4bd1..cecaf57a 100644 --- a/orquesta/specs/native/v1/models.py +++ b/orquesta/specs/native/v1/models.py @@ -210,13 +210,12 @@ def finalize_context(self, next_task_name, task_transition_meta, in_ctx): errors = [] task_transition_specs = getattr(self, "next") or [] - task_transition_spec = task_transition_specs[task_transition_meta[3]["ref"]] + task_transition_spec = task_transition_specs[task_transition_meta.data["ref"]] next_task_names = getattr(task_transition_spec, "do") or [] if next_task_name in next_task_names: for task_publish_spec in getattr(task_transition_spec, "publish") or {}: - var_name = list(task_publish_spec.items())[0][0] - default_var_value = list(task_publish_spec.items())[0][1] + var_name, default_var_value = dict_util.first_item(task_publish_spec) try: rendered_var_value = expr_base.evaluate(default_var_value, rolling_ctx) @@ -552,9 +551,9 @@ def inspect_context(self, parent=None): schema_path = parent.get("schema_path") + ".patternProperties.^\\w+$" task_parent = {"ctx": task_ctx, "spec_path": spec_path, "schema_path": schema_path} - result = task_spec.inspect_context(parent=task_parent) - errors.extend(result[0]) - task_ctx = list(set(task_ctx + result[1])) + result_errors, result_ctx = task_spec.inspect_context(parent=task_parent) + errors.extend(result_errors) + task_ctx = list(set(task_ctx + result_ctx)) rolling_ctx = list(set(rolling_ctx + task_ctx)) # Identify the next set of tasks and related transition specs. @@ -577,9 +576,7 @@ def inspect_context(self, parent=None): transitions.append(entry) for entry in transitions: - next_task_name = entry[0] - task_transition_spec = entry[1] - seq_num = entry[2] + next_task_name, task_transition_spec, seq_num = entry parent_ctx = { "ctx": task_ctx, @@ -587,9 +584,9 @@ def inspect_context(self, parent=None): "schema_path": schema_path + ".properties.next.items", } - result = task_transition_spec.inspect_context(parent_ctx) - errors.extend(result[0]) - branch_ctx = list(set(task_ctx + result[1])) + result_errors, result_ctx = task_transition_spec.inspect_context(parent_ctx) + errors.extend(result_errors) + branch_ctx = list(set(task_ctx + result_ctx)) if ( not next_task_name @@ -656,8 +653,7 @@ def render_input(self, runtime_inputs, in_ctx=None): for input_spec in getattr(self, "input") or []: if isinstance(input_spec, dict): - input_name = list(input_spec.items())[0][0] - default_input_value = list(input_spec.items())[0][1] + input_name, default_input_value = dict_util.first_item(input_spec) else: input_name = input_spec default_input_value = None @@ -679,8 +675,7 @@ def render_vars(self, in_ctx): errors = [] for var_spec in getattr(self, "vars") or []: - var_name = list(var_spec.items())[0][0] - default_var_value = list(var_spec.items())[0][1] + var_name, default_var_value = dict_util.first_item(var_spec) try: rendered_var_value = expr_base.evaluate(default_var_value, rolling_ctx) @@ -698,8 +693,7 @@ def render_output(self, in_ctx): errors = [] for output_spec in output_specs: - output_name = list(output_spec.items())[0][0] - default_output_value = list(output_spec.items())[0][1] + output_name, default_output_value = dict_util.first_item(output_spec) try: rendered_output_value = expr_base.evaluate(default_output_value, rolling_ctx) diff --git a/orquesta/statetypes.py b/orquesta/statetypes.py new file mode 100644 index 00000000..f81f7e29 --- /dev/null +++ b/orquesta/statetypes.py @@ -0,0 +1,283 @@ +# Copyright 2025 The StackStorm Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Static type definitions for the runtime state passed around the conductor. + +The workflow conductor (:mod:`orquesta.conducting`) and the state machines +(:mod:`orquesta.machines`) exchange a handful of plain ``dict`` structures that +are serialized to JSON and persisted (e.g. in StackStorm's database). Because +they are plain dicts, their shape is invisible to readers, IDEs, and type +checkers -- you have to grep the code to learn what keys exist. + +These ``TypedDict`` definitions document those shapes *without changing any +runtime behavior*: a ``TypedDict`` is an ordinary ``dict`` at run time, so +``serialize``/``deserialize`` and the on-disk/wire format are completely +unaffected. They exist purely so method signatures can say what they accept and +return, and so ``mypy`` / editors can check key access. + +Compatibility note: orquesta targets Python 3.10+, where ``typing.NotRequired`` +is not yet available (3.11+). Optional keys are therefore expressed with the +"required base class + ``total=False`` subclass" idiom, and the functional +``TypedDict(...)`` form is used where a key name is a Python keyword (``in``). +""" + +from __future__ import annotations + +from typing import Any +from typing import Dict +from typing import List +from typing import NamedTuple +from typing import TypedDict + + +# --------------------------------------------------------------------------- +# Shared leaf structures +# --------------------------------------------------------------------------- + +# A context "pointer" bundle. "in" is a list of indexes into +# ``WorkflowState.contexts``; "out" maps a task-transition id to the index of +# the context produced for that transition. Note "in" is a Python keyword, so +# the functional TypedDict syntax is required here. +_TaskContextsIn = TypedDict("_TaskContextsIn", {"in": List[int]}) +_TaskContextsOut = TypedDict("_TaskContextsOut", {"out": Dict[str, int]}, total=False) + + +# pylint on Python 3.10 does not recognize functional-form TypedDicts as classes, +class TaskContexts( # pylint: disable=inherit-non-class,duplicate-bases + _TaskContextsIn, _TaskContextsOut +): + """``{"in": [, ...], "out"?: {: }}``. + + ``in`` is always present; ``out`` is only added once a task completes and + produces a new context for one of its outbound transitions. + """ + + +class ItemState(TypedDict): + """Per-item execution status for a with-items task (``staged["items"][i]``).""" + + status: str + + +class _RetryStateRequired(TypedDict): + # Comes from the task's retry spec plus a runtime ``tally`` counter. + when: Any # spec expression string, or None when unconditional + count: int # max number of retries + tally: int # number of retries performed so far + + +class RetryState(_RetryStateRequired, total=False): + """Runtime retry bookkeeping stored under a task entry's ``retry`` key.""" + + delay: int # resolved delay (seconds) between retries + + +# --------------------------------------------------------------------------- +# Task state entry (an element of ``WorkflowState.sequence``) +# --------------------------------------------------------------------------- + + +class _TaskStateEntryRequired(TypedDict): + id: str + route: int + ctxs: TaskContexts + # backref transition id -> index of the predecessor entry in ``sequence``. + prev: Dict[str, int] + # outbound transition id -> whether its criteria evaluated to True. + next: Dict[str, bool] + + +class TaskStateEntry(_TaskStateEntryRequired, total=False): + """A task execution record in ``WorkflowState.sequence``. + + ``status`` is absent until the task state machine processes its first + event; the remaining keys are set only in specific situations (retry + configured, task is terminal, task marked to be ignored on rerun). + """ + + status: str + retry: RetryState + term: bool + ignore: bool + + +# --------------------------------------------------------------------------- +# Staged task (an element of ``WorkflowState.staged``) +# --------------------------------------------------------------------------- + + +class _StagedTaskRequired(TypedDict): + id: str + route: int + ctxs: TaskContexts + prev: Dict[str, int] + ready: bool + + +class StagedTask(_StagedTaskRequired, total=False): + """A task queued to (potentially) run next, in ``WorkflowState.staged``. + + ``items``/``completed`` appear for with-items tasks, ``retry`` when the + task is re-staged for a retry, and ``run_on_fail`` when the task should + still run after the workflow has failed (remediation). + """ + + items: List[ItemState] + completed: bool + retry: RetryState + run_on_fail: bool + + +# --------------------------------------------------------------------------- +# Serialized workflow state (WorkflowState.serialize / deserialize) +# --------------------------------------------------------------------------- + +# "task__rN" -> index into ``sequence`` of that task/route's latest entry. +TaskIndex = Dict[str, int] + +# --- Routing ------------------------------------------------------------- +# +# A "route" identifies one branch of execution through the workflow. Whenever a +# workflow forks (a "split" task), each parallel branch needs its own copy of +# the context so sibling branches don't clobber each other's variables. Each +# such branch is a distinct route. +# +# Routes live in ``WorkflowState.routes``, which is effectively a registry +# indexed by route id. Note the deliberate two-part vocabulary -- the code +# passes both around and they are easy to confuse: +# +# * ``RouteId`` -- an *integer index* into that registry. This is the value +# passed around everywhere as ``route`` / ``prev_route`` / +# ``next_task_route``. Route id 0 is the initial (main) route. +# * ``RouteDetails`` -- the value stored *at* that index: the ordered list of +# split task-transition ids taken to reach the branch. It is the branch's +# "fingerprint" -- the root route is ``[]`` (no splits taken) and each +# further split appends the transition id that caused the fork. +# +# Two branches share a route id if and only if they were reached by the exact +# same sequence of splits (see ``WorkflowConductor._evaluate_route``). +RouteId = int +RouteDetails = List[str] +RoutesRegistry = List[RouteDetails] + +# --- Task transitions ---------------------------------------------------- +# +# A task transition is a directed edge in the workflow graph: "when task A +# completes, and criteria C holds, go to task B". The graph is a NetworkX +# MultiDiGraph, whose edges are 4-tuples ``(source, destination, key, data)``. +# ``graph.get_next_transitions`` / ``get_prev_transitions`` return those edges +# wrapped as ``TaskTransition`` named tuples, so both attribute access +# (``t.destination``) and the historical positional access (``t[1]``, +# unpacking, sorting) work. The fields are: +# +# * ``source`` -- id of the task the transition leaves from. +# * ``destination`` -- id of the task the transition leads to (the "next" +# task). +# * ``key`` -- the MultiDiGraph edge key; the ordinal that +# distinguishes multiple transitions between the same +# pair of tasks. Also called the transition/seq key. +# * ``data`` -- the edge attributes, e.g. +# ``{"criteria": [, ...], "ref": }``. +# +# ``TaskTransitionId`` is the string ``"__t"`` (see +# ``constants.TASK_STATE_TRANSITION_FORMAT``). Watch out for its two uses, +# which differ in *which* task id is embedded: +# +# * Forward (in ``TaskStateEntry["next"]`` and ``TaskContexts["out"]``): +# built from the *destination* id -- "did this task's outbound transition +# to B fire?". +# * Backref (in ``TaskStateEntry["prev"]`` / ``StagedTask["prev"]``): built +# from the *source* id -- "which predecessor transition reached me?". +TaskId = str +TransitionKey = int +TransitionData = Dict[str, Any] +TaskTransitionId = str + + +class TaskTransition(NamedTuple): + """A workflow-graph edge (``source`` -> ``destination``) with its metadata. + + Backward compatible with the raw NetworkX edge tuple it replaces: it is a + ``tuple`` subclass, so positional access (``t[0]``), unpacking, and sorting + all behave exactly as before. + """ + + source: TaskId + destination: TaskId + key: TransitionKey + data: TransitionData + + +class _SerializedWorkflowStateRequired(TypedDict): + contexts: List[Dict[str, Any]] + routes: List[RouteDetails] + sequence: List[TaskStateEntry] + staged: List[StagedTask] + status: str + tasks: TaskIndex + + +class SerializedWorkflowState(_SerializedWorkflowStateRequired, total=False): + """The JSON-serializable form produced by ``WorkflowState.serialize``. + + ``reruns`` is only included when at least one rerun has been requested. + """ + + # Each entry is the list of ``sequence`` indexes rerun in one rerun request. + reruns: List[List[int]] + + +# --------------------------------------------------------------------------- +# Transient (non-serialized) structures +# --------------------------------------------------------------------------- + + +class _RuntimeTaskRequired(TypedDict): + id: str + route: int + ctx: Dict[str, Any] + spec: Any # a rendered TaskSpec instance (orquesta.specs.*) + actions: List[Any] # list of rendered action specs + + +class RuntimeTask(_RuntimeTaskRequired, total=False): + """The in-memory "next task" dict built by ``WorkflowConductor.get_task``. + + This is *not* serialized -- it is handed to the caller (e.g. the workflow + engine) to describe a task to schedule. ``delay`` is added when the task + defines one; ``items_count``/``concurrency`` are added for with-items tasks. + """ + + delay: int + items_count: int + concurrency: Any # resolved concurrency (int) or None + + +class _LogEntryRequired(TypedDict): + type: str # "info" | "warn" | "error" + message: str + + +class LogEntry(_LogEntryRequired, total=False): + """An entry appended to the conductor's ``log`` or ``errors`` lists. + + The optional keys are only present when the corresponding argument was + provided to ``WorkflowConductor.log_entry`` (nulls are not inserted). + """ + + task_id: str + route: int + task_transition_id: str + result: Any + data: Any diff --git a/orquesta/tests/unit/commands/test_rehearse_command.py b/orquesta/tests/unit/commands/test_rehearse_command.py index 86cd7a7a..d5db89f7 100644 --- a/orquesta/tests/unit/commands/test_rehearse_command.py +++ b/orquesta/tests/unit/commands/test_rehearse_command.py @@ -13,9 +13,10 @@ # limitations under the License. import argparse -import mock import unittest +from unittest import mock + from orquesta.commands import rehearsal from orquesta import exceptions as exc from orquesta.tests.fixtures import loader as fixture_loader diff --git a/orquesta/utils/date.py b/orquesta/utils/date.py index 06a7b128..43138fc5 100644 --- a/orquesta/utils/date.py +++ b/orquesta/utils/date.py @@ -49,6 +49,8 @@ def format(dt, usec=True, offset=True): if offset: ost = dt.strftime("%z") + # strftime gives the offset as "+HHMM"; insert a colon after the sign+hours + # (index 3) to make it ISO-8601 "+HH:MM". Empty (naive datetime) defaults to UTC. ost = (ost[:3] + ":" + ost[3:]) if ost else "+00:00" else: tz = dt.tzinfo.tzname(dt) if dt.tzinfo else "UTC" diff --git a/orquesta/utils/dictionary.py b/orquesta/utils/dictionary.py index 0d2ec965..c86f6738 100644 --- a/orquesta/utils/dictionary.py +++ b/orquesta/utils/dictionary.py @@ -35,6 +35,16 @@ def merge_dicts(left, right, overwrite=True): return left +def first_item(obj): + """Return the first ``(key, value)`` pair of a dict as a tuple. + + Convenience for the common single-key-dict case (e.g. a ``{name: value}`` + input/var/output/publish spec) where the caller wants both the sole key and + its value. Reads clearer than ``list(obj.items())[0][0]`` / ``[0][1]``. + """ + return next(iter(obj.items())) + + def get_dict_value(obj, path, raise_key_error=False): item = obj traversed = "" diff --git a/requirements-test.txt b/requirements-test.txt index 5cf56026..7ac3cb86 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -2,7 +2,6 @@ coverage==7.2.7 black==22.3.0 flake8==7.3.0 -mock==4.0.3 pytest==6.2.5 pytest-cov==4.1.0 pep8==1.7.1