Skip to content
Snippets Groups Projects

Feature/remove typing import

Merged Karel van Klink requested to merge feature/remove-typing-import into develop
1 file
+ 8
8
Compare changes
  • Side-by-side
  • Inline
@@ -2,7 +2,7 @@ import difflib
import pprint
from copy import deepcopy
from itertools import chain, repeat
from typing import Callable, Tuple, cast
from typing import Callable, cast
from uuid import uuid4
import structlog
@@ -143,7 +143,7 @@ class WorkflowInstanceForTests(LazyWorkflowInstance):
return f"WorkflowInstanceForTests('{self.workflow}','{self.name}')"
def _store_step(step_log: list[Tuple[Step, WFProcess]]) -> Callable[[ProcessStat, Step, WFProcess], WFProcess]:
def _store_step(step_log: list[tuple[Step, WFProcess]]) -> Callable[[ProcessStat, Step, WFProcess], WFProcess]:
def __store_step(pstat: ProcessStat, step: Step, state: WFProcess) -> WFProcess:
try:
state = state.map(lambda s: json_loads(json_dumps(s)))
@@ -164,14 +164,14 @@ def _sanitize_input(input_data: State | list[State]) -> list[State]:
return cast(list[State], json_loads(json_dumps(input_data)))
def run_workflow(workflow_key: str, input_data: State | list[State]) -> Tuple[WFProcess, ProcessStat, list]:
def run_workflow(workflow_key: str, input_data: State | list[State]) -> tuple[WFProcess, ProcessStat, list]:
# ATTENTION!! This code needs to be as similar as possible to `server.services.processes.start_process`
# The main differences are: we use a different step log function and we don't run in
# a sepperate thread
user_data = _sanitize_input(input_data)
user = "john.doe"
step_log: list[Tuple[Step, WFProcess]] = []
step_log: list[tuple[Step, WFProcess]] = []
process_id = uuid4()
workflow = get_workflow(workflow_key)
@@ -201,8 +201,8 @@ def run_workflow(workflow_key: str, input_data: State | list[State]) -> Tuple[WF
def resume_workflow(
process: ProcessStat, step_log: list[Tuple[Step, WFProcess]], input_data: State
) -> Tuple[WFProcess, list]:
process: ProcessStat, step_log: list[tuple[Step, WFProcess]], input_data: State
) -> tuple[WFProcess, list]:
# ATTENTION!! This code needs to be as similar as possible to `server.services.processes.resume_process`
# The main differences are: we use a different step log function, and we don't run in a separate thread
user_data = _sanitize_input(input_data)
@@ -229,7 +229,7 @@ def resume_workflow(
def run_form_generator(
form_generator: FormGenerator, extra_inputs: list[State] | None = None
) -> Tuple[list[dict], State]:
) -> tuple[list[dict], State]:
"""Run a form generator to get the resulting forms and result.
Warning! This does not run the actual pydantic validation on purpose. However, you should
@@ -245,7 +245,7 @@ def run_form_generator(
Returns:
-------
Tuple[list[dict], State]: A list of generated forms and the result state for the whole generator.
tuple[list[dict], State]: A list of generated forms and the result state for the whole generator.
Example:
-------
Loading