diff --git a/cwltool/factory.py b/cwltool/factory.py index eaf98e3cf..7bc320462 100644 --- a/cwltool/factory.py +++ b/cwltool/factory.py @@ -1,12 +1,21 @@ +"""Wrap a CWL document as a callable Python object.""" + +import argparse +import functools import os +import sys from typing import Any, Optional, Union from . import load_tool -from .context import LoadingContext, RuntimeContext +from .argparser import arg_parser +from .context import LoadingContext, RuntimeContext, getdefault from .errors import WorkflowException from .executors import JobExecutor, SingleJobExecutor +from .main import find_default_container from .process import Process -from .utils import CWLObjectType +from .resolver import tool_resolver +from .secrets import SecretStore +from .utils import DEFAULT_TMP_PREFIX, CWLObjectType class WorkflowStatus(Exception): @@ -25,11 +34,15 @@ def __init__(self, t: Process, factory: "Factory") -> None: self.t = t self.factory = factory - def __call__(self, **kwargs): - # type: (**Any) -> Union[str, Optional[CWLObjectType]] - runtime_context = self.factory.runtime_context.copy() - runtime_context.basedir = os.getcwd() - out, status = self.factory.executor(self.t, kwargs, runtime_context) + def __call__(self, **kwargs: Any) -> Union[str, Optional[CWLObjectType]]: + """ + Execute the process. + + :raise WorkflowStatus: If the result is not a success. + """ + if not self.factory.runtime_context.basedir: + self.factory.runtime_context.basedir = os.getcwd() + out, status = self.factory.executor(self.t, kwargs, self.factory.runtime_context) if status != "success": raise WorkflowStatus(out, status) else: @@ -47,18 +60,24 @@ def __init__( executor: Optional[JobExecutor] = None, loading_context: Optional[LoadingContext] = None, runtime_context: Optional[RuntimeContext] = None, + argsl: Optional[list[str]] = None, + args: Optional[argparse.Namespace] = None, ) -> None: + """Create a CWL Process factory from a CWL document.""" + if argsl is not None: + args = arg_parser().parse_args(argsl) if executor is None: - executor = SingleJobExecutor() - self.executor = executor + self.executor: JobExecutor = SingleJobExecutor() + else: + self.executor = executor if runtime_context is None: - self.runtime_context = RuntimeContext() + self.runtime_context = RuntimeContext(vars(args) if args else {}) + self._fix_runtime_context() else: self.runtime_context = runtime_context if loading_context is None: - self.loading_context = LoadingContext() - self.loading_context.singularity = self.runtime_context.singularity - self.loading_context.podman = self.runtime_context.podman + self.loading_context = LoadingContext(vars(args) if args else {}) + self._fix_loading_context(self.runtime_context) else: self.loading_context = loading_context @@ -68,3 +87,45 @@ def make(self, cwl: Union[str, dict[str, Any]]) -> Callable: if isinstance(load, int): raise WorkflowException("Error loading tool") return Callable(load, self) + + def _fix_loading_context(self, runtime_context: RuntimeContext) -> None: + self.loading_context.resolver = getdefault(self.loading_context.resolver, tool_resolver) + self.loading_context.singularity = runtime_context.singularity + self.loading_context.podman = runtime_context.podman + + def _fix_runtime_context(self) -> None: + self.runtime_context.basedir = os.getcwd() + self.runtime_context.find_default_container = functools.partial( + find_default_container, default_container=None, use_biocontainers=None + ) + + if sys.platform == "darwin": + default_mac_path = "/private/tmp/docker_tmp" + if self.runtime_context.tmp_outdir_prefix == DEFAULT_TMP_PREFIX: + self.runtime_context.tmp_outdir_prefix = default_mac_path + + for dirprefix in ("tmpdir_prefix", "tmp_outdir_prefix", "cachedir"): + if ( + getattr(self.runtime_context, dirprefix) + and getattr(self.runtime_context, dirprefix) != DEFAULT_TMP_PREFIX + ): + sl = ( + "/" + if getattr(self.runtime_context, dirprefix).endswith("/") + or dirprefix == "cachedir" + else "" + ) + setattr( + self.runtime_context, + dirprefix, + os.path.abspath(getattr(self.runtime_context, dirprefix)) + sl, + ) + if not os.path.exists(os.path.dirname(getattr(self.runtime_context, dirprefix))): + try: + os.makedirs(os.path.dirname(getattr(self.runtime_context, dirprefix))) + except Exception as e: + print("Failed to create directory: %s", e) + + self.runtime_context.secret_store = getdefault( + self.runtime_context.secret_store, SecretStore() + )