This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for cwltool.context

"""Shared context objects that replace use of kwargs."""
import copy
import os
import shutil
import tempfile
import threading
from typing import (

from ruamel.yaml.comments import CommentedMap
from schema_salad.avro.schema import Names
from schema_salad.ref_resolver import Loader
from schema_salad.utils import FetcherCallableType

from .mpi import MpiConfig
from .pathmapper import PathMapper
from .stdfsaccess import StdFsAccess
from .utils import DEFAULT_TMP_PREFIX, CWLObjectType, HasReqsHints, ResolverType

    from cwl_utils.parser.cwl_v1_2 import LoadingOptions

    from .builder import Builder
    from .cwlprov.provenance_profile import ProvenanceProfile
    from .cwlprov.ro import ResearchObject
    from .mutation import MutationManager
    from .process import Process
    from .secrets import SecretStore
    from .software_requirements import DependenciesConfiguration

class ContextBase:
    """Shared kwargs based initializer for :py:class:`RuntimeContext` and :py:class:`LoadingContext`."""

    def __init__(self, kwargs: Optional[Dict[str, Any]] = None) -> None:
        if kwargs:
            for k, v in kwargs.items():
                if hasattr(self, k):
                    setattr(self, k, v)

def make_tool_notimpl(toolpath_object: CommentedMap, loadingContext: "LoadingContext") -> "Process":
    """Fake implementation of the make tool function."""
    raise NotImplementedError()

default_make_tool = make_tool_notimpl

def log_handler(
    outdir: str,
    base_path_logs: str,
    stdout_path: Optional[str],
    stderr_path: Optional[str],
) -> None:
    """Move logs from log location to final output."""
    if outdir != base_path_logs:
        if stdout_path:
            new_stdout_path = stdout_path.replace(base_path_logs, outdir)
            shutil.copy2(stdout_path, new_stdout_path)
        if stderr_path:
            new_stderr_path = stderr_path.replace(base_path_logs, outdir)
            shutil.copy2(stderr_path, new_stderr_path)

def set_log_dir(outdir: str, log_dir: str, subdir_name: str) -> str:
    """Set the log directory."""
    if log_dir == "":
        return outdir
        return log_dir + "/" + subdir_name

[docs]class LoadingContext(ContextBase):
[docs] def __init__(self, kwargs: Optional[Dict[str, Any]] = None) -> None: """Initialize the LoadingContext from the kwargs.""" self.debug: bool = False self.metadata: CWLObjectType = {} self.requirements: Optional[List[CWLObjectType]] = None self.hints: Optional[List[CWLObjectType]] = None self.overrides_list: List[CWLObjectType] = [] self.loader: Optional[Loader] = None self.avsc_names: Optional[Names] = None self.disable_js_validation: bool = False self.js_hint_options_file: Optional[str] = None self.do_validate: bool = True self.enable_dev: bool = False self.strict: bool = True self.resolver: Optional[ResolverType] = None self.fetcher_constructor: Optional[FetcherCallableType] = None self.construct_tool_object = default_make_tool self.research_obj: Optional[ResearchObject] = None self.orcid: str = "" self.cwl_full_name: str = "" self.host_provenance: bool = False self.user_provenance: bool = False self.prov_obj: Optional["ProvenanceProfile"] = None self.do_update: Optional[bool] = None self.jobdefaults: Optional[CommentedMap] = None self.doc_cache: bool = True self.relax_path_checks: bool = False self.singularity: bool = False self.podman: bool = False self.eval_timeout: float = 60 self.codegen_idx: Dict[str, Tuple[Any, "LoadingOptions"]] = {} self.fast_parser = False self.skip_resolve_all = False self.skip_schemas = False super().__init__(kwargs)
[docs] def copy(self) -> "LoadingContext": """Return a copy of this :py:class:`LoadingContext`.""" return copy.copy(self)
[docs]class RuntimeContext(ContextBase): outdir: Optional[str] = None tmpdir: str = "" tmpdir_prefix: str = DEFAULT_TMP_PREFIX tmp_outdir_prefix: str = "" stagedir: str = ""
[docs] def __init__(self, kwargs: Optional[Dict[str, Any]] = None) -> None: """Initialize the RuntimeContext from the kwargs.""" select_resources_callable = Callable[ [Dict[str, Union[int, float]], RuntimeContext], Dict[str, Union[int, float]], ] self.user_space_docker_cmd: Optional[str] = None self.secret_store: Optional["SecretStore"] = None self.no_read_only: bool = False self.custom_net: Optional[str] = None self.no_match_user: bool = False self.preserve_environment: Optional[Iterable[str]] = None self.preserve_entire_environment: bool = False self.use_container: bool = True self.force_docker_pull: bool = False self.rm_tmpdir: bool = True self.pull_image: bool = True self.rm_container: bool = True self.move_outputs: Union[Literal["move"], Literal["leave"], Literal["copy"]] = "move" self.log_dir: str = "" self.set_log_dir = set_log_dir self.log_dir_handler = log_handler self.streaming_allowed: bool = False self.singularity: bool = False self.podman: bool = False self.debug: bool = False self.compute_checksum: bool = True self.name: str = "" self.default_container: Optional[str] = "" self.find_default_container: Optional[Callable[[HasReqsHints], Optional[str]]] = None self.cachedir: Optional[str] = None self.part_of: str = "" self.basedir: str = "" self.toplevel: bool = False self.mutation_manager: Optional["MutationManager"] = None self.make_fs_access = StdFsAccess self.path_mapper = PathMapper self.builder: Optional["Builder"] = None self.docker_outdir: str = "" self.docker_tmpdir: str = "" self.docker_stagedir: str = "" self.js_console: bool = False self.job_script_provider: Optional[DependenciesConfiguration] = None self.select_resources: Optional[select_resources_callable] = None self.eval_timeout: float = 60 self.postScatterEval: Optional[Callable[[CWLObjectType], Optional[CWLObjectType]]] = None self.on_error: Union[Literal["stop"], Literal["continue"]] = "stop" self.strict_memory_limit: bool = False self.strict_cpu_limit: bool = False self.cidfile_dir: Optional[str] = None self.cidfile_prefix: Optional[str] = None self.workflow_eval_lock: Optional[threading.Condition] = None self.research_obj: Optional[ResearchObject] = None self.orcid: str = "" self.cwl_full_name: str = "" self.process_run_id: Optional[str] = None self.prov_obj: Optional[ProvenanceProfile] = None self.mpi_config: MpiConfig = MpiConfig() self.default_stdout: Optional[Union[IO[bytes], TextIO]] = None self.default_stderr: Optional[Union[IO[bytes], TextIO]] = None self.validate_only: bool = False self.validate_stdout: Optional[Union[IO[bytes], TextIO, IO[str]]] = None super().__init__(kwargs) if self.tmp_outdir_prefix == "": self.tmp_outdir_prefix = self.tmpdir_prefix
[docs] def get_outdir(self) -> str: """Return :py:attr:`outdir` or create one with :py:attr:`tmp_outdir_prefix`.""" if self.outdir: return self.outdir return self.create_outdir()
[docs] def get_tmpdir(self) -> str: """Return :py:attr:`tmpdir` or create one with :py:attr:`tmpdir_prefix`.""" if self.tmpdir: return self.tmpdir return self.create_tmpdir()
[docs] def get_stagedir(self) -> str: """Return :py:attr:`stagedir` or create one with :py:attr:`tmpdir_prefix`.""" if self.stagedir: return self.stagedir tmp_dir, tmp_prefix = os.path.split(self.tmpdir_prefix) return tempfile.mkdtemp(prefix=tmp_prefix, dir=tmp_dir)
[docs] def create_tmpdir(self) -> str: """Create a temporary directory that respects :py:attr:`tmpdir_prefix`.""" tmp_dir, tmp_prefix = os.path.split(self.tmpdir_prefix) return tempfile.mkdtemp(prefix=tmp_prefix, dir=tmp_dir)
[docs] def create_outdir(self) -> str: """Create a temporary directory that respects :py:attr:`tmp_outdir_prefix`.""" out_dir, out_prefix = os.path.split(self.tmp_outdir_prefix) return tempfile.mkdtemp(prefix=out_prefix, dir=out_dir)
[docs] def copy(self) -> "RuntimeContext": """Return a copy of this :py:class:`RuntimeContext`.""" return copy.copy(self)
[docs]def getdefault(val: Any, default: Any) -> Any: """Return the ``val`` using the ``default`` as backup in case the val is ``None``.""" if val is None: return default else: return val