wfexs_backend.workflow_engines.cwl_engine

Module Contents

Classes

CWLWorkflowEngine

Functions

_normalize_image_id
_normalize_sif_id

API

wfexs_backend.workflow_engines.cwl_engine._normalize_image_id(string: str) → wfexs_backend.common.RelPath

wfexs_backend.workflow_engines.cwl_engine._normalize_sif_id(string: str) → wfexs_backend.common.RelPath

class wfexs_backend.workflow_engines.cwl_engine.CWLWorkflowEngine(container_factory_clazz: Type[ContainerFactory] = NoContainerFactory, cacheDir: pathlib.Path | None = None, engine_config: EngineLocalConfig | None = None, progs_mapping: ProgsMapping | None = None, engineTweaksDir: pathlib.Path | None = None, cacheWorkflowDir: pathlib.Path | None = None, cacheWorkflowInputsDir: pathlib.Path | None = None, workDir: pathlib.Path | None = None, outputsDir: pathlib.Path | None = None, outputMetaDir: pathlib.Path | None = None, intermediateDir: pathlib.Path | None = None, tempDir: pathlib.Path | None = None, stagedContainersDir: pathlib.Path | None = None, secure_exec: bool = False, allowOther: bool = False, config_directory: pathlib.Path | None = None, writable_containers: bool = False)

Bases: wfexs_backend.workflow_engines.WorkflowEngine

CWLTOOL_PYTHON_PACKAGE = 'cwltool'

CWL_UTILS_PYTHON_PACKAGE = 'cwl-utils'

SCHEMA_SALAD_PYTHON_PACKAGE = 'schema-salad'

CWL_REPO = 'https://github.com/common-workflow-language/'

CWLTOOL_REPO = None

DEVEL_CWLTOOL_REPO = None

CWL_UTILS_REPO = None

DEFAULT_CWLTOOL_VERSION = 'cast(...)'

DEVEL_CWLTOOL_PACKAGE = None

DEVEL_CWLTOOL_VERSION = None

PODMAN_CWLTOOL_VERSION = 'cast(...)'

NO_WRAPPER_CWLTOOL_VERSION = 'cast(...)'

CWLTOOL_MAX_PYVER: Sequence[Tuple[int | None, int | None, EngineVersion]] = [(3, None), (3, 6), (None, None)]

INPUT_DECLARATIONS_FILENAME = 'inputdeclarations.yaml'

NODEJS_WRAPPER = 'nodejs_wrapper.bash'

NODEJS_CONTAINER_TAG = 'ContainerTaggedName(...)'

OPERATIONAL_CONTAINER_TAGS = None

ENGINE_NAME = 'cwl'

SUPPORTED_CONTAINER_TYPES = None

SUPPORTED_SECURE_EXEC_CONTAINER_TYPES = None

classmethod MyWorkflowType() → wfexs_backend.workflow_engines.WorkflowType

classmethod HasExplicitOutputs() → bool

classmethod SupportedContainerTypes() → Set[ContainerType]

classmethod SupportedSecureExecContainerTypes() → Set[ContainerType]

property engine_url: wfexs_backend.common.URIType

identifyWorkflow(localWf: wfexs_backend.common.LocalWorkflow, engineVer: EngineVersion | None = None) → Tuple[EngineVersion, LocalWorkflow] | Tuple[None, None]

This method should return the effective engine version needed to run it when this workflow engine recognizes the workflow type

materializeEngineVersion(engineVersion: wfexs_backend.common.EngineVersion) → Tuple[EngineVersion, pathlib.Path, Fingerprint]

Method to ensure the required engine version is materialized It should raise an exception when the exact version is unavailable, and no replacement could be fetched

_materializeEngineVersionLocal(engineVersion: wfexs_backend.common.EngineVersion, search_other: bool = True) → Tuple[EngineVersion, pathlib.Path, Fingerprint]

Method to ensure the required engine version is materialized It should raise an exception when the exact version is unavailable, and no replacement could be fetched

_get_engine_version_str(matWfEng: wfexs_backend.workflow_engines.MaterializedWorkflowEngine) → wfexs_backend.workflow_engines.WorkflowEngineVersionStr

__get_engine_version_str_local(matWfEng: wfexs_backend.workflow_engines.MaterializedWorkflowEngine) → wfexs_backend.workflow_engines.WorkflowEngineVersionStr

_enrichWorkflowDeps(localWf: wfexs_backend.common.LocalWorkflow, engineVer: wfexs_backend.common.EngineVersion) → wfexs_backend.common.LocalWorkflow

materializeWorkflow(matWorkflowEngine: wfexs_backend.workflow_engines.MaterializedWorkflowEngine, consolidatedWorkflowDir: pathlib.Path, offline: bool = False, profiles: Sequence[str] | None = None, context_inputs: Sequence[MaterializedInput] = [], context_environment: Sequence[MaterializedInput] = []) → Tuple[MaterializedWorkflowEngine, Sequence[ContainerTaggedName]]

Method to ensure the workflow has been materialized. In the case of CWL, it returns a new materialized workflow engine, which points to a newly genereated LocalWorkflow instance pointing to the consolidated workflow. Also, it returns the list of containers.

sideContainers() → Sequence[ContainerTaggedName]

Containers needed by the engine to work

simpleContainerFileName(imageUrl: wfexs_backend.common.URIType) → Sequence[RelPath]

This method was borrowed from common-workflow-language/cwltool

static generateDotWorkflow(matWfEng: wfexs_backend.workflow_engines.MaterializedWorkflowEngine, dagFile: pathlib.Path) → None

launchWorkflow(matWfEng: wfexs_backend.workflow_engines.MaterializedWorkflowEngine, matInputs: Sequence[MaterializedInput], matEnvironment: Sequence[MaterializedInput], outputs: Sequence[ExpectedOutput], profiles: Sequence[str] | None = None) → Iterator[StagedExecution]

Method to execute the workflow

createYAMLFile(matInputs: Sequence[MaterializedInput], cwlInputs: Mapping[SymbolicParamName, Any], filename: pathlib.Path) → Mapping[SymbolicParamName, ExecInputVal]

Method to create a YAML file that describes the execution inputs of the workflow needed for their execution. Return parsed inputs.

augmentCWLInputs(matInputs: Sequence[MaterializedInput], cwlInputs: Mapping[SymbolicParamName, Any]) → Sequence[MaterializedInput]

Generate additional MaterializedInput for the implicit params.

executionInputs(matInputs: Sequence[MaterializedInput], cwlInputs: Mapping[SymbolicParamName, Any]) → Mapping[SymbolicParamName, ExecInputVal]

Setting execution inputs needed to execute the workflow