Skip to content
Snippets Groups Projects
Verified Commit bdd5021d authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

remove Tuple import

parent f4df5ebf
Branches
Tags
1 merge request!81Feature/remove typing import
Pipeline #84204 passed
......@@ -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:
-------
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment