iterative__dvc-8904

{
  "status": "ok",
  "stdout": "Initialized DVC repository.\n\n+---------------------------------------------------------------------+\n|                                                                     |\n|        DVC has enabled anonymous aggregate usage analytics.         |\n|     Read the analytics documentation (and how to opt-out) here:     |\n|             <https://dvc.org/doc/user-guide/analytics>              |\n|                                                                     |\n+---------------------------------------------------------------------+\n\nWhat's next?\n------------\n- Check out the documentation: <https://dvc.org/doc>\n- Get help and share ideas: <https://dvc.org/chat>\n- Star us on GitHub: <https://github.com/iterative/dvc>\n",
  "stderr": "ERROR: Stage 'build@2021' not found inside 'dvc.yaml' file\n",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "./dvc/stage/exceptions.py:97:        super().__init__(f\"Stage '{self.name}' not found inside '{self.file}' file\")\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dvc/stage/exceptions.py",
  "content": "        source = \"source\"\n        if len(missing_files) > 1:\n            source += \"s\"\n\n        msg = \"missing data '{}': {}\".format(source, \", \".join(missing_files))\n        super().__init__(msg)\n\n\nclass DataSourceChanged(DvcException):\n    def __init__(self, path: str):\n        super().__init__(f\"data source changed: {path}\")\n\n\nclass StageNotFound(DvcException, KeyError):\n    def __init__(self, file, name):\n        self.file = file.relpath\n        self.name = name\n        super().__init__(f\"Stage '{self.name}' not found inside '{self.file}' file\")\n\n    def __str__(self):\n        # `KeyError` quotes the message\n        # see: https://bugs.python.org/issue2651\n        return self.msg\n\n\nclass StageNameUnspecified(DvcException):\n    def __init__(self, file):\n        super().__init__(\n            \"Stage name not provided.\"\n            \"Please specify the name as: `{}:stage_name`\".format(file.relpath)\n        )",
  "start_line": 80,
  "end_line": 110,
  "total_lines": 119,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "./tests/func/test_stage_load.py:13:    StageNotFound,\n./tests/func/test_stage_load.py:77:    with pytest.raises(StageNotFound):\n./tests/func/test_stage_load.py:294:    with pytest.raises(StageNotFound):\n./tests/func/test_stage_load.py:316:    with pytest.raises(StageNotFound):\n./tests/func/test_stage_load.py:345:    with pytest.raises(StageNotFound):\n./tests/func/test_dvcfile.py:14:from dvc.stage.loader import StageNotFound\n./tests/func/test_dvcfile.py:63:    with pytest.raises(StageNotFound):\n./tests/func/test_repro_multistage.py:12:from dvc.stage.exceptions import StageNotFound\n./tests/func/test_repro_multistage.py:19:    with pytest.raises(StageNotFound):\n./tests/func/test_data_cloud.py:12:from dvc.stage.exceptions import StageNotFound\n./tests/func/test_data_cloud.py:383:    with pytest.raises(StageNotFound):\n./tests/func/test_data_cloud.py:386:    with pytest.raises(StageNotFound):\n./dvc/repo/remove.py:8:    StageNotFound,\n./dvc/repo/remove.py:23:    except (StageNotFound, StageFileDoesNotExistError) as e:\n./dvc/repo/stage.py:47:    from dvc.stage.exceptions import StageNotFound\n./dvc/repo/stage.py:51:        with suppress(StageNotFound):\n./dvc/repo/stage.py:385:            from dvc.stage.exceptions import StageFileDoesNotExistError, StageNotFound\n./dvc/repo/stage.py:401:            except StageNotFound as exc:\n./dvc/stage/exceptions.py:93:class StageNotFound(DvcException, KeyError):\n./dvc/stage/__init__.py:116:    from .exceptions import StageNotFound\n./dvc/stage/__init__.py:123:    except StageNotFound:\n./dvc/stage/__init__.py:501:        from .exceptions import StageFileDoesNotExistError, StageNotFound\n./dvc/stage/__init__.py:513:        except (StageFileDoesNotExistError, StageNotFound):\n./dvc/stage/loader.py:17:from .exceptions import StageNameUnspecified, StageNotFound\n./dvc/stage/loader.py:141:            raise StageNotFound(self.dvcfile, name)  # noqa: B904\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dvc/stage/loader.py",
  "content": "                stage.PARAM_OUTS,\n                stage.PARAM_METRICS,\n                stage.PARAM_PLOTS,\n            ],\n        )\n        stage.outs = lcat(\n            output.load_from_pipeline(stage, data, typ=key)\n            for key, data in outs.items()\n        )\n\n        if lock_data:\n            stage.cmd_changed = lock_data.get(Stage.PARAM_CMD) != stage.cmd\n\n        cls.fill_from_lock(stage, lock_data)\n        return stage\n\n    @once\n    def lockfile_needs_update(self):\n        # if lockfile does not have all of the entries that dvc.yaml says it\n        # should have, provide a debug message once\n        # pylint: disable=protected-access\n        lockfile = self.dvcfile._lockfile.relpath\n        logger.debug(\"Lockfile '%s' needs to be updated.\", lockfile)\n\n    def __getitem__(self, name):\n        if not name:\n            raise StageNameUnspecified(self.dvcfile)\n\n        try:\n            resolved_data = self.resolver.resolve_one(name)\n        except EntryNotFound:\n            raise StageNotFound(self.dvcfile, name)  # noqa: B904\n\n        if self.lockfile_data and name not in self.lockfile_data:\n            self.lockfile_needs_update()\n            logger.trace(  # type: ignore[attr-defined]\n                \"No lock entry found for '%s:%s'\", self.dvcfile.relpath, name\n            )\n\n        resolved_stage = resolved_data[name]\n        stage = self.load_stage(\n            self.dvcfile,\n            name,\n            resolved_stage,\n            self.lockfile_data.get(name, {}),\n        )\n\n        stage.tracked_vars = self.resolver.tracked_vars.get(name, {})\n        group, *keys = name.rsplit(JOIN, maxsplit=1)\n        if group and keys and name not in self.stages_data:\n            stage.raw_data.generated_from = group",
  "start_line": 110,
  "end_line": 160,
  "total_lines": 223,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "dvc/stage/loader.py",
  "content": "import logging\nfrom collections.abc import Mapping\nfrom copy import deepcopy\nfrom itertools import chain\nfrom typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple\n\nfrom funcy import get_in, lcat, once, project\n\nfrom dvc import dependency, output\nfrom dvc.parsing import FOREACH_KWD, JOIN, EntryNotFound\nfrom dvc.parsing.versions import LOCKFILE_VERSION\nfrom dvc.utils.objects import cached_property\nfrom dvc_data.hashfile.hash_info import HashInfo\nfrom dvc_data.hashfile.meta import Meta\n\nfrom . import PipelineStage, Stage, loads_from\nfrom .exceptions import StageNameUnspecified, StageNotFound\nfrom .params import StageParams\nfrom .utils import fill_stage_dependencies, resolve_paths\n\nif TYPE_CHECKING:\n    from dvc.dvcfile import ProjectFile, SingleStageFile\n    from dvc.output import Output\n\nlogger = logging.getLogger(__name__)\n\n\nclass StageLoader(Mapping):\n    def __init__(\n        self,\n        dvcfile: \"ProjectFile\",\n        data,\n        lockfile_data=None,\n    ):\n        self.dvcfile = dvcfile\n        self.resolver = self.dvcfile.resolver\n        self.data = data or {}\n        self.stages_data = self.data.get(\"stages\", {})\n        self.repo = self.dvcfile.repo\n\n        lockfile_data = lockfile_data or {}\n        version = LOCKFILE_VERSION.from_dict(lockfile_data)\n        if version == LOCKFILE_VERSION.V1:\n            self._lockfile_data = lockfile_data\n        else:\n            self._lockfile_data = lockfile_data.get(\"stages\", {})\n\n    @cached_property\n    def lockfile_data(self) -> Dict[str, Any]:\n        if not self._lockfile_data:\n            logger.debug(\"Lockfile for '%s' not found\", self.dvcfile.relpath)\n        return self._lockfile_data\n\n    @staticmethod\n    def fill_from_lock(stage, lock_data=None):\n        \"\"\"Fill values for params, checksums for outs and deps from lock.\"\"\"\n        if not lock_data:\n            return\n\n        from dvc.output import merge_file_meta_from_cloud\n\n        assert isinstance(lock_data, dict)\n        items: Iterable[Tuple[str, \"Output\"]] = chain(\n            ((StageParams.PARAM_DEPS, dep) for dep in stage.deps),\n            ((StageParams.PARAM_OUTS, out) for out in stage.outs),\n        )\n\n        checksums = {\n            key: {item[\"path\"]: item for item in lock_data.get(key, {})}\n            for key in [StageParams.PARAM_DEPS, StageParams.PARAM_OUTS]\n        }\n        for key, item in items:\n            path = item.def_path\n            if isinstance(item, dependency.ParamsDependency):\n                item.fill_values(get_in(lock_data, [stage.PARAM_PARAMS, path]))\n                continue\n            info = get_in(checksums, [key, path], {})\n            info = info.copy()\n            info.pop(\"path\", None)\n\n            item.meta = Meta.from_dict(merge_file_meta_from_cloud(info))\n            hash_value = getattr(item.meta, item.hash_name, None)\n            item.hash_info = HashInfo(item.hash_name, hash_value)\n            files = get_in(checksums, [key, path, item.PARAM_FILES], None)\n            if files:\n                item.files = [merge_file_meta_from_cloud(f) for f in files]\n            # pylint: disable-next=protected-access\n            item._compute_meta_hash_info_from_files()\n\n    @classmethod\n    def load_stage(cls, dvcfile: \"ProjectFile\", name, stage_data, lock_data=None):\n        assert all([name, dvcfile, dvcfile.repo, dvcfile.path])\n        assert stage_data\n        assert isinstance(stage_data, dict)\n\n        path, wdir = resolve_paths(\n            dvcfile.repo.fs, dvcfile.path, stage_data.get(Stage.PARAM_WDIR)\n        )\n        stage = loads_from(PipelineStage, dvcfile.repo, path, wdir, stage_data)\n        stage.name = name",
  "start_line": 1,
  "end_line": 100,
  "total_lines": 223,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "dvc/dvcfile.py",
  "content": "import contextlib\nimport logging\nimport os\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Callable,\n    Dict,\n    List,\n    Optional,\n    Tuple,\n    TypeVar,\n    Union,\n)\n\nfrom dvc.exceptions import DvcException\nfrom dvc.parsing.versions import LOCKFILE_VERSION, SCHEMA_KWD\nfrom dvc.stage import serialize\nfrom dvc.stage.exceptions import (\n    StageFileBadNameError,\n    StageFileDoesNotExistError,\n    StageFileIsNotDvcFileError,\n)\nfrom dvc.utils import relpath\nfrom dvc.utils.collections import apply_diff\nfrom dvc.utils.objects import cached_property\nfrom dvc.utils.serialize import dump_yaml, modify_yaml\n\nif TYPE_CHECKING:\n    from dvc.repo import Repo\n    from dvc.types import StrOrBytesPath\n\n    from .parsing import DataResolver\n    from .stage import Stage\n\nlogger = logging.getLogger(__name__)\n_T = TypeVar(\"_T\")\n\nDVC_FILE = \"Dvcfile\"\nDVC_FILE_SUFFIX = \".dvc\"\nPROJECT_FILE = \"dvc.yaml\"\nLOCK_FILE = \"dvc.lock\"\n\n\nclass FileIsGitIgnored(DvcException):\n    def __init__(self, path, pipeline_file=False):\n        super().__init__(\n            \"{}'{}' is git-ignored.\".format(\n                \"bad DVC file name \" if pipeline_file else \"\", path\n            )\n        )\n\n\nclass ParametrizedDumpError(DvcException):\n    pass\n\n\ndef is_valid_filename(path):\n    return path.endswith(DVC_FILE_SUFFIX) or os.path.basename(path) in [\n        DVC_FILE,\n        PROJECT_FILE,\n    ]\n\n\ndef is_dvc_file(path):\n    return os.path.isfile(path) and (is_valid_filename(path) or is_lock_file(path))\n\n\ndef is_lock_file(path):\n    return os.path.basename(path) == LOCK_FILE\n\n\ndef is_git_ignored(repo, path):\n    from dvc.fs import LocalFileSystem\n    from dvc.scm import NoSCMError\n\n    try:\n        return isinstance(repo.fs, LocalFileSystem) and repo.scm.is_ignored(path)\n    except NoSCMError:\n        return False\n\n\ndef check_dvcfile_path(repo, path):\n    if not is_valid_filename(path):\n        raise StageFileBadNameError(\n            \"bad DVC file name '{}'. DVC files should be named \"\n            \"'{}' or have a '.dvc' suffix (e.g. '{}.dvc').\".format(\n                relpath(path), PROJECT_FILE, os.path.basename(path)\n            )\n        )\n\n    if is_git_ignored(repo, path):\n        raise FileIsGitIgnored(relpath(path), True)\n\n\nclass FileMixin:\n    SCHEMA: Callable[[_T], _T]\n\n    def __init__(self, repo, path, verify=True, **kwargs):\n        self.repo = repo\n        self.path = path\n        self.verify = verify\n\n    def __repr__(self):\n        return \"{}: {}\".format(\n            self.__class__.__name__, relpath(self.path, self.repo.root_dir)\n        )\n\n    def __hash__(self):\n        return hash(self.path)\n\n    def __eq__(self, other):\n        return self.repo == other.repo and os.path.abspath(\n            self.path\n        ) == os.path.abspath(other.path)\n\n    def __str__(self):\n        return f\"{self.__class__.__name__}: {self.relpath}\"\n\n    @property\n    def relpath(self):\n        return relpath(self.path)\n\n    def exists(self):\n        is_ignored = self.repo.dvcignore.is_ignored_file(self.path)\n        return self.repo.fs.exists(self.path) and not is_ignored\n\n    def _is_git_ignored(self):\n        return is_git_ignored(self.repo, self.path)\n\n    def _verify_filename(self):\n        if self.verify:\n            check_dvcfile_path(self.repo, self.path)\n\n    def _check_gitignored(self):\n        if self._is_git_ignored():\n            raise FileIsGitIgnored(self.path)\n\n    def load(self, **kwargs: Any) -> Any:\n        d, _ = self._load(**kwargs)\n        return d\n\n    def _load(self, **kwargs: Any) -> Tuple[Any, str]:\n        # it raises the proper exceptions by priority:\n        # 1. when the file doesn't exists\n        # 2. filename is not a DVC file\n        # 3. path doesn't represent a regular file\n        # 4. when the file is git ignored\n        if not self.exists():\n            dvc_ignored = self.repo.dvcignore.is_ignored_file(self.path)",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 457,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "import logging\nimport os\nfrom collections.abc import Mapping, Sequence\nfrom copy import deepcopy\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Dict,\n    List,\n    NamedTuple,\n    Optional,\n    Tuple,\n    Type,\n    Union,\n)\n\nfrom funcy import collecting, first, isa, join, reraise\n\nfrom dvc.exceptions import DvcException\nfrom dvc.parsing.interpolate import ParseError\nfrom dvc.utils.objects import cached_property\n\nfrom .context import (\n    Context,\n    ContextError,\n    KeyNotInContext,\n    MergeError,\n    Node,\n    VarsAlreadyLoaded,\n)\nfrom .interpolate import (\n    check_recursive_parse_errors,\n    is_interpolated_string,\n    recurse,\n    to_str,\n)\n\nif TYPE_CHECKING:\n    from typing import NoReturn\n\n    from dvc.repo import Repo\n    from dvc.types import DictStrAny\n\n    from .context import SeqOrMap\n\n\nlogger = logging.getLogger(__name__)\n\nSTAGES_KWD = \"stages\"\nVARS_KWD = \"vars\"\nWDIR_KWD = \"wdir\"\nPARAMS_KWD = \"params\"\nFOREACH_KWD = \"foreach\"\nDO_KWD = \"do\"\n\nDEFAULT_PARAMS_FILE = \"params.yaml\"\n\nJOIN = \"@\"\n\n\nclass ResolveError(DvcException):\n    pass\n\n\nclass EntryNotFound(DvcException):\n    pass\n\n\ndef _format_preamble(msg: str, path: str, spacing: str = \" \") -> str:\n    return f\"failed to parse {msg} in '{path}':{spacing}\"\n\n\ndef format_and_raise(exc: Exception, msg: str, path: str) -> \"NoReturn\":\n    spacing = (\n        \"\\n\" if isinstance(exc, (ParseError, MergeError, VarsAlreadyLoaded)) else \" \"\n    )\n    message = _format_preamble(msg, path, spacing) + str(exc)\n\n    # FIXME: cannot reraise because of how we log \"cause\" of the exception\n    # the error message is verbose, hence need control over the spacing\n    _reraise_err(ResolveError, message, from_exc=exc)\n\n\ndef _reraise_err(\n    exc_cls: Type[Exception], *args, from_exc: Optional[Exception] = None\n) -> \"NoReturn\":\n    err = exc_cls(*args)\n    if from_exc and logger.isEnabledFor(logging.DEBUG):\n        raise err from from_exc\n    raise err\n\n\ndef check_syntax_errors(\n    definition: \"DictStrAny\", name: str, path: str, where: str = \"stages\"\n):\n    for key, d in definition.items():\n        try:\n            check_recursive_parse_errors(d)\n        except ParseError as exc:\n            format_and_raise(exc, f\"'{where}.{name}.{key}'\", path)\n\n\ndef is_map_or_seq(data: Any) -> bool:\n    _is_map_or_seq = isa(Mapping, Sequence)\n    return not isinstance(data, str) and _is_map_or_seq(data)\n\n\ndef split_foreach_name(name: str) -> Tuple[str, Optional[str]]:\n    group, *keys = name.rsplit(JOIN, maxsplit=1)\n    return group, first(keys)\n\n\ndef check_interpolations(data: \"DictStrAny\", where: str, path: str):\n    def func(s: \"DictStrAny\") -> None:\n        if is_interpolated_string(s):\n            raise ResolveError(\n                _format_preamble(f\"'{where}'\", path) + \"interpolating is not allowed\"\n            )\n\n    return recurse(func)(data)\n\n\nDefinition = Union[\"ForeachDefinition\", \"EntryDefinition\"]\n\n\ndef make_definition(\n    resolver: \"DataResolver\", name: str, definition: \"DictStrAny\", **kwargs\n) -> Definition:\n    args = resolver, resolver.context, name, definition\n    if FOREACH_KWD in definition:\n        return ForeachDefinition(*args, **kwargs)\n    return EntryDefinition(*args, **kwargs)\n\n\nclass DataResolver:\n    def __init__(self, repo: \"Repo\", wdir: str, d: dict):\n        self.fs = fs = repo.fs\n\n        if os.path.isabs(wdir):\n            wdir = fs.path.relpath(wdir)\n            wdir = \"\" if wdir == os.curdir else wdir\n\n        self.wdir = wdir\n        self.relpath = fs.path.normpath(fs.path.join(self.wdir, \"dvc.yaml\"))\n\n        vars_ = d.get(VARS_KWD, [])\n        check_interpolations(vars_, VARS_KWD, self.relpath)\n        self.context: Context = Context()\n\n        try:",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "import logging\nimport os\nfrom collections.abc import Mapping, Sequence\nfrom copy import deepcopy\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Dict,\n    List,\n    NamedTuple,\n    Optional,\n    Tuple,\n    Type,\n    Union,\n)\n\nfrom funcy import collecting, first, isa, join, reraise\n\nfrom dvc.exceptions import DvcException\nfrom dvc.parsing.interpolate import ParseError\nfrom dvc.utils.objects import cached_property\n\nfrom .context import (\n    Context,\n    ContextError,\n    KeyNotInContext,\n    MergeError,\n    Node,\n    VarsAlreadyLoaded,\n)\nfrom .interpolate import (\n    check_recursive_parse_errors,\n    is_interpolated_string,\n    recurse,\n    to_str,\n)\n\nif TYPE_CHECKING:\n    from typing import NoReturn\n\n    from dvc.repo import Repo\n    from dvc.types import DictStrAny\n\n    from .context import SeqOrMap\n\n\nlogger = logging.getLogger(__name__)\n\nSTAGES_KWD = \"stages\"\nVARS_KWD = \"vars\"\nWDIR_KWD = \"wdir\"\nPARAMS_KWD = \"params\"\nFOREACH_KWD = \"foreach\"\nDO_KWD = \"do\"\n\nDEFAULT_PARAMS_FILE = \"params.yaml\"\n\nJOIN = \"@\"\n\n\nclass ResolveError(DvcException):\n    pass\n\n\nclass EntryNotFound(DvcException):\n    pass\n\n\ndef _format_preamble(msg: str, path: str, spacing: str = \" \") -> str:\n    return f\"failed to parse {msg} in '{path}':{spacing}\"\n\n\ndef format_and_raise(exc: Exception, msg: str, path: str) -> \"NoReturn\":\n    spacing = (\n        \"\\n\" if isinstance(exc, (ParseError, MergeError, VarsAlreadyLoaded)) else \" \"\n    )\n    message = _format_preamble(msg, path, spacing) + str(exc)\n\n    # FIXME: cannot reraise because of how we log \"cause\" of the exception\n    # the error message is verbose, hence need control over the spacing\n    _reraise_err(ResolveError, message, from_exc=exc)\n\n\ndef _reraise_err(\n    exc_cls: Type[Exception], *args, from_exc: Optional[Exception] = None\n) -> \"NoReturn\":\n    err = exc_cls(*args)\n    if from_exc and logger.isEnabledFor(logging.DEBUG):\n        raise err from from_exc\n    raise err\n\n\ndef check_syntax_errors(\n    definition: \"DictStrAny\", name: str, path: str, where: str = \"stages\"\n):\n    for key, d in definition.items():\n        try:\n            check_recursive_parse_errors(d)\n        except ParseError as exc:\n            format_and_raise(exc, f\"'{where}.{name}.{key}'\", path)\n\n\ndef is_map_or_seq(data: Any) -> bool:\n    _is_map_or_seq = isa(Mapping, Sequence)\n    return not isinstance(data, str) and _is_map_or_seq(data)\n\n\ndef split_foreach_name(name: str) -> Tuple[str, Optional[str]]:\n    group, *keys = name.rsplit(JOIN, maxsplit=1)\n    return group, first(keys)\n\n\ndef check_interpolations(data: \"DictStrAny\", where: str, path: str):\n    def func(s: \"DictStrAny\") -> None:\n        if is_interpolated_string(s):\n            raise ResolveError(\n                _format_preamble(f\"'{where}'\", path) + \"interpolating is not allowed\"\n            )\n\n    return recurse(func)(data)\n\n\nDefinition = Union[\"ForeachDefinition\", \"EntryDefinition\"]\n\n\ndef make_definition(\n    resolver: \"DataResolver\", name: str, definition: \"DictStrAny\", **kwargs\n) -> Definition:\n    args = resolver, resolver.context, name, definition\n    if FOREACH_KWD in definition:\n        return ForeachDefinition(*args, **kwargs)\n    return EntryDefinition(*args, **kwargs)\n\n\nclass DataResolver:\n    def __init__(self, repo: \"Repo\", wdir: str, d: dict):\n        self.fs = fs = repo.fs\n\n        if os.path.isabs(wdir):\n            wdir = fs.path.relpath(wdir)\n            wdir = \"\" if wdir == os.curdir else wdir\n\n        self.wdir = wdir\n        self.relpath = fs.path.normpath(fs.path.join(self.wdir, \"dvc.yaml\"))\n\n        vars_ = d.get(VARS_KWD, [])\n        check_interpolations(vars_, VARS_KWD, self.relpath)\n        self.context: Context = Context()\n\n        try:",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            args = fs, vars_, wdir  # load from `vars` section\n            self.context.load_from_vars(*args, default=DEFAULT_PARAMS_FILE)\n        except ContextError as exc:\n            format_and_raise(exc, \"'vars'\", self.relpath)\n\n        # we use `tracked_vars` to keep a dictionary of used variables\n        # by the interpolated entries.\n        self.tracked_vars: Dict[str, Mapping] = {}\n\n        stages_data = d.get(STAGES_KWD, {})\n        # we wrap the definitions into ForeachDefinition and EntryDefinition,\n        # that helps us to optimize, cache and selectively load each one of\n        # them as we need, and simplify all of this DSL/parsing logic.\n        self.definitions: Dict[str, Definition] = {\n            name: make_definition(self, name, definition)\n            for name, definition in stages_data.items()\n        }\n\n    def resolve_one(self, name: str):\n        group, key = split_foreach_name(name)\n\n        if not self._has_group_and_key(group, key):\n            raise EntryNotFound(f\"Could not find '{name}'\")\n\n        # all of the checks for `key` not being None for `ForeachDefinition`\n        # and/or `group` not existing in the `interim`, etc. should be\n        # handled by the `self.has_key()` above.\n        definition = self.definitions[group]\n        if isinstance(definition, EntryDefinition):\n            return definition.resolve()\n\n        assert key\n        return definition.resolve_one(key)\n\n    def resolve(self):\n        \"\"\"Used for testing purposes, otherwise use resolve_one().\"\"\"\n        data = join(map(self.resolve_one, self.get_keys()))\n        logger.trace(\"Resolved dvc.yaml:\\n%s\", data)  # type: ignore[attr-defined]\n        return {STAGES_KWD: data}\n\n    def has_key(self, key: str):\n        return self._has_group_and_key(*split_foreach_name(key))\n\n    def _has_group_and_key(self, group: str, key: Optional[str] = None):\n        try:\n            definition = self.definitions[group]\n        except KeyError:\n            return False\n\n        if key:\n            return isinstance(definition, ForeachDefinition) and definition.has_member(\n                key\n            )\n        return not isinstance(definition, ForeachDefinition)\n\n    @collecting\n    def get_keys(self):\n        for name, definition in self.definitions.items():\n            if isinstance(definition, ForeachDefinition):\n                yield from definition.get_generated_names()\n                continue\n            yield name\n\n    def track_vars(self, name: str, vars_) -> None:\n        self.tracked_vars[name] = vars_\n\n\nclass EntryDefinition:\n    def __init__(\n        self,\n        resolver: DataResolver,\n        context: Context,\n        name: str,\n        definition: \"DictStrAny\",\n        where: str = STAGES_KWD,\n    ):\n        self.resolver = resolver\n        self.wdir = self.resolver.wdir\n        self.relpath = self.resolver.relpath\n        self.context = context\n        self.name = name\n        self.definition = definition\n        self.where = where\n\n    def _resolve_wdir(\n        self, context: Context, name: str, wdir: Optional[str] = None\n    ) -> str:\n        if not wdir:\n            return self.wdir\n\n        try:\n            wdir = to_str(context.resolve_str(wdir))\n        except (ContextError, ParseError) as exc:\n            format_and_raise(exc, f\"'{self.where}.{name}.wdir'\", self.relpath)\n        return self.resolver.fs.path.join(self.wdir, wdir)\n\n    def resolve(self, **kwargs):\n        try:\n            return self.resolve_stage(**kwargs)\n        except ContextError as exc:\n            format_and_raise(exc, f\"stage '{self.name}'\", self.relpath)\n\n    def resolve_stage(self, skip_checks: bool = False) -> \"DictStrAny\":\n        context = self.context\n        name = self.name\n        if not skip_checks:\n            # we can check for syntax errors as we go for interpolated entries,\n            # but for foreach-generated ones, once is enough, which it does\n            # that itself. See `ForeachDefinition.do_definition`.\n            check_syntax_errors(self.definition, name, self.relpath)\n\n        # we need to pop vars from generated/evaluated data\n        definition = deepcopy(self.definition)\n\n        wdir = self._resolve_wdir(context, name, definition.get(WDIR_KWD))\n        vars_ = definition.pop(VARS_KWD, [])\n        # FIXME: Should `vars` be templatized?\n        check_interpolations(vars_, f\"{self.where}.{name}.vars\", self.relpath)\n        if vars_:\n            # Optimization: Lookahead if it has any vars, if it does not, we\n            # don't need to clone them.\n            context = Context.clone(context)\n\n        try:\n            fs = self.resolver.fs\n            context.load_from_vars(fs, vars_, wdir, stage_name=name)\n        except VarsAlreadyLoaded as exc:\n            format_and_raise(exc, f\"'{self.where}.{name}.vars'\", self.relpath)\n\n        logger.trace(  # type: ignore[attr-defined]\n            \"Context during resolution of stage %s:\\n%s\", name, context\n        )\n\n        with context.track() as tracked_data:\n            # NOTE: we do not pop \"wdir\", and resolve it again\n            # this does not affect anything and is done to try to\n            # track the source of `wdir` interpolation.\n            # This works because of the side-effect that we do not\n            # allow overwriting and/or str interpolating complex objects.\n            # Fix if/when those assumptions are no longer valid.\n            resolved = {\n                key: self._resolve(context, value, key, skip_checks)\n                for key, value in definition.items()\n            }\n\n        self.resolver.track_vars(name, tracked_data)\n        return {name: resolved}\n\n    def _resolve(\n        self, context: \"Context\", value: Any, key: str, skip_checks: bool",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 455,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "    ) -> \"DictStrAny\":\n        try:\n            return context.resolve(\n                value, skip_interpolation_checks=skip_checks, key=key\n            )\n        except (ParseError, KeyNotInContext) as exc:\n            format_and_raise(exc, f\"'{self.where}.{self.name}.{key}'\", self.relpath)\n\n\nclass IterationPair(NamedTuple):\n    key: str = \"key\"\n    value: str = \"item\"\n\n\nclass ForeachDefinition:\n    def __init__(\n        self,\n        resolver: DataResolver,\n        context: Context,\n        name: str,\n        definition: \"DictStrAny\",\n        where: str = STAGES_KWD,\n    ):\n        self.resolver = resolver\n        self.relpath = self.resolver.relpath\n        self.context = context\n        self.name = name\n\n        assert DO_KWD in definition\n        self.foreach_data = definition[FOREACH_KWD]\n        self._do_definition = definition[DO_KWD]\n\n        self.pair = IterationPair()\n        self.where = where\n\n    @cached_property\n    def do_definition(self):\n        # optimization: check for syntax errors only once for `foreach` stages\n        check_syntax_errors(self._do_definition, self.name, self.relpath)\n        return self._do_definition\n\n    @cached_property\n    def resolved_iterable(self):\n        return self._resolve_foreach_data()\n\n    def _resolve_foreach_data(self) -> \"SeqOrMap\":\n        try:\n            iterable = self.context.resolve(self.foreach_data, unwrap=False)\n        except (ContextError, ParseError) as exc:\n            format_and_raise(exc, f\"'{self.where}.{self.name}.foreach'\", self.relpath)\n\n        # foreach data can be a resolved dictionary/list.\n        self._check_is_map_or_seq(iterable)\n        # foreach stages will have `item` and `key` added to the context\n        # so, we better warn them if they have them already in the context\n        # from the global vars. We could add them in `set_temporarily`, but\n        # that'd make it display for each iteration.\n        self._warn_if_overwriting(self._inserted_keys(iterable))\n        return iterable\n\n    def _check_is_map_or_seq(self, iterable):\n        if not is_map_or_seq(iterable):\n            node = iterable.value if isinstance(iterable, Node) else iterable\n            typ = type(node).__name__\n            raise ResolveError(\n                f\"failed to resolve '{self.where}.{self.name}.foreach'\"\n                f\" in '{self.relpath}': expected list/dictionary, got \" + typ\n            )\n\n    def _warn_if_overwriting(self, keys: List[str]):\n        warn_for = [k for k in keys if k in self.context]\n        if warn_for:\n            linking_verb = \"is\" if len(warn_for) == 1 else \"are\"\n            logger.warning(\n                (\n                    \"%s %s already specified, \"\n                    \"will be overwritten for stages generated from '%s'\"\n                ),\n                \" and \".join(warn_for),\n                linking_verb,\n                self.name,\n            )\n\n    def _inserted_keys(self, iterable) -> List[str]:\n        keys = [self.pair.value]\n        if isinstance(iterable, Mapping):\n            keys.append(self.pair.key)\n        return keys\n\n    @cached_property\n    def normalized_iterable(self):\n        \"\"\"Convert sequence to Mapping with keys normalized.\"\"\"\n        iterable = self.resolved_iterable\n        if isinstance(iterable, Mapping):\n            return iterable\n\n        assert isinstance(iterable, Sequence)\n        if any(map(is_map_or_seq, iterable)):\n            # if the list contains composite data, index are the keys\n            return {to_str(idx): value for idx, value in enumerate(iterable)}\n\n        # for simple lists, eg: [\"foo\", \"bar\"],  contents are the key itself\n        return {to_str(value): value for value in iterable}\n\n    def has_member(self, key: str) -> bool:\n        return key in self.normalized_iterable\n\n    def get_generated_names(self):\n        return list(map(self._generate_name, self.normalized_iterable))\n\n    def _generate_name(self, key: str) -> str:\n        return f\"{self.name}{JOIN}{key}\"\n\n    def resolve_all(self) -> \"DictStrAny\":\n        return join(map(self.resolve_one, self.normalized_iterable))\n\n    def resolve_one(self, key: str) -> \"DictStrAny\":\n        return self._each_iter(key)\n\n    def _each_iter(self, key: str) -> \"DictStrAny\":\n        err_message = f\"Could not find '{key}' in foreach group '{self.name}'\"\n        with reraise(KeyError, EntryNotFound(err_message)):\n            value = self.normalized_iterable[key]\n\n        # NOTE: we need to use resolved iterable/foreach-data,\n        # not the normalized ones to figure out whether to make item/key\n        # available\n        inserted = self._inserted_keys(self.resolved_iterable)\n        temp_dict = {self.pair.value: value}\n        key_str = self.pair.key\n        if key_str in inserted:\n            temp_dict[key_str] = key\n\n        with self.context.set_temporarily(temp_dict, reserve=True):\n            # optimization: item and key can be removed on __exit__() as they\n            # are top-level values, and are not merged recursively.\n            # This helps us avoid cloning context, which is slower\n            # (increasing the size of the context might increase\n            # the no. of items to be generated which means more cloning,\n            # i.e. quadratic complexity).\n            generated = self._generate_name(key)\n            entry = EntryDefinition(\n                self.resolver, self.context, generated, self.do_definition\n            )\n            try:\n                # optimization: skip checking for syntax errors on each foreach\n                # generated stages. We do it once when accessing do_definition.\n                return entry.resolve_stage(skip_checks=True)\n            except ContextError as exc:\n                format_and_raise(exc, f\"stage '{generated}'\", self.relpath)",
  "start_line": 301,
  "end_line": 450,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 4, in <module>\nImportError: cannot import name 'FS' from 'dvc.fs' (/workspace/dvc/fs/__init__.py)\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 4, in <module>\nImportError: cannot import name 'FS' from 'dvc.fs' (/workspace/dvc/fs/__init__.py)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "dvc/parsing/context.py",
  "content": "import logging\nfrom abc import ABC, abstractmethod\nfrom collections import defaultdict\nfrom collections.abc import Mapping, MutableMapping, MutableSequence, Sequence\nfrom contextlib import contextmanager\nfrom copy import deepcopy\nfrom dataclasses import dataclass, field, replace\nfrom typing import Any, Dict, List, Optional, Union\n\nfrom funcy import identity, lfilter, nullcontext, select\n\nfrom dvc.exceptions import DvcException\nfrom dvc.parsing.interpolate import (\n    get_expression,\n    get_matches,\n    is_exact_string,\n    normalize_key,\n    recurse,\n    str_interpolate,\n    validate_value,\n)\n\nlogger = logging.getLogger(__name__)\nSeqOrMap = Union[Sequence, Mapping]\nDictStr = Dict[str, Any]\n\n\nclass ContextError(DvcException):\n    pass\n\n\nclass ReservedKeyError(ContextError):\n    def __init__(self, keys, path=None):\n        from dvc.utils.humanize import join\n\n        self.keys = keys\n        self.path = path\n\n        n = \"key\" + (\"s\" if len(keys) > 1 else \"\")\n        msg = f\"attempted to modify reserved {n} {join(keys)}\"\n        if path:\n            msg += f\" in '{path}'\"\n        super().__init__(msg)\n\n\nclass MergeError(ContextError):\n    def __init__(self, key, new, into):\n        self.key = key\n        to_node = into[key]\n        if not isinstance(to_node, Node) or not isinstance(new, Node):\n            super().__init__(f\"cannot merge '{key}' as it already exists in {into}\")\n            return\n\n        assert isinstance(to_node, Node)\n        assert isinstance(new, Node)\n        preexisting = to_node.meta.source\n        new_src = new.meta.source\n        path = new.meta.path()\n        super().__init__(\n            f\"cannot redefine '{path}' from '{new_src}'\"\n            f\" as it already exists in '{preexisting}'\"\n        )\n\n\nclass ParamsLoadError(ContextError):\n    pass\n\n\nclass KeyNotInContext(ContextError, KeyError):\n    def __init__(self, key: str) -> None:\n        self.key: str = key\n        super().__init__(f\"Could not find '{key}'\")\n\n    def __str__(self):\n        return self.msg\n\n\nclass VarsAlreadyLoaded(ContextError):\n    pass\n\n\ndef _merge(into, update, overwrite):\n    for key, val in update.items():\n        if isinstance(into.get(key), Mapping) and isinstance(val, Mapping):\n            _merge(into[key], val, overwrite)\n        else:\n            if key in into and not overwrite:\n                raise MergeError(key, val, into)\n            into[key] = val\n            assert isinstance(into[key], Node)\n\n\ndef recurse_not_a_node(data: dict):\n    def func(item):\n        assert not isinstance(item, Node)\n\n    return recurse(func)(data)\n\n\n@dataclass\nclass Meta:\n    source: Optional[str] = None\n    dpaths: List[str] = field(default_factory=list)\n    local: bool = True\n\n    @staticmethod\n    def update_path(meta: \"Meta\", path: Union[str, int]):\n        dpaths = meta.dpaths[:] + [str(path)]\n        return replace(meta, dpaths=dpaths)\n\n    def __str__(self):\n        string = self.source or \"<local>\"\n        string += \":\" + self.path()\n        return string\n\n    def path(self):\n        return \".\".join(self.dpaths)\n\n\ndef _default_meta() -> Meta:\n    return Meta()\n\n\nclass Node:\n    meta: Meta\n\n    def get_sources(self):\n        raise NotImplementedError\n\n    @property\n    @abstractmethod\n    def value(self):\n        pass\n\n\n@dataclass\nclass Value(Node):\n    _value: Any\n    meta: Meta = field(compare=False, default_factory=_default_meta, repr=False)\n\n    def __repr__(self):\n        return repr(self._value)\n\n    def __str__(self) -> str:\n        return str(self._value)\n\n    def get_sources(self):\n        return {self.meta.source: self.meta.path()}\n\n    @property",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 545,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/context.py",
  "content": "    def value(self):\n        return self._value\n\n\nPRIMITIVES = (int, float, str, bytes, bool)\n\n\nclass Container(Node, ABC):\n    meta: Meta\n    data: Union[list, dict]\n    _key_transform = staticmethod(identity)\n\n    def __init__(self, meta=None) -> None:\n        self.meta = meta or _default_meta()\n\n    def _convert(self, key, value):\n        meta = Meta.update_path(self.meta, key)\n        return self._convert_with_meta(value, meta)\n\n    @staticmethod\n    def _convert_with_meta(value, meta: Optional[Meta] = None):\n        if value is None or isinstance(value, PRIMITIVES):\n            assert meta\n            return Value(value, meta=meta)\n        if isinstance(value, Node):\n            return value\n        if isinstance(value, (list, dict)):\n            assert meta\n            container = CtxDict if isinstance(value, dict) else CtxList\n            return container(value, meta=meta)\n        msg = f\"Unsupported value of type '{type(value).__name__}' in '{meta}'\"\n        raise TypeError(msg)\n\n    def __repr__(self):\n        return repr(self.data)\n\n    def __getitem__(self, key):\n        return self.data[key]\n\n    def __setitem__(self, key, value):\n        self.data[key] = self._convert(key, value)\n\n    def __delitem__(self, key):\n        del self.data[key]\n\n    def __len__(self):\n        return len(self.data)\n\n    def __iter__(self):\n        return iter(self.data)\n\n    def __eq__(self, o):\n        container = type(self)\n        if isinstance(o, container):\n            return o.data == self.data\n        return container(o) == self\n\n    def select(self, key: str):\n        index, *rems = key.split(sep=\".\", maxsplit=1)\n        index = index.strip()\n        index = self._key_transform(index)\n        try:\n            d = self[index]\n        except LookupError as exc:\n            raise ValueError(f\"Could not find '{index}' in {self.data}\") from exc\n\n        if not rems:\n            return d\n\n        rem = rems[0]\n        if not isinstance(d, Container):\n            raise ValueError(  # noqa: TRY004\n                f\"{index} is a primitive value, cannot get '{rem}'\"\n            )\n        return d.select(rem)\n\n    def get_sources(self):\n        return {}\n\n\nclass CtxList(Container, MutableSequence):\n    _key_transform = staticmethod(int)\n\n    def __init__(self, values: Sequence, meta: Optional[Meta] = None):\n        super().__init__(meta=meta)\n        self.data: list = []\n        self.extend(values)\n\n    def insert(self, index: int, value):\n        self.data.insert(index, self._convert(index, value))\n\n    def get_sources(self):\n        return {self.meta.source: self.meta.path()}\n\n    @property\n    def value(self):\n        return [node.value for node in self]\n\n    def __deepcopy__(self, _):\n        # optimization: we don't support overriding a list\n        new = CtxList([])\n        new.data = self.data[:]  # Short-circuiting __setitem__\n        return new\n\n\nclass CtxDict(Container, MutableMapping):\n    def __init__(\n        self,\n        mapping: Optional[Mapping] = None,\n        meta: Optional[Meta] = None,\n        **kwargs,\n    ):\n        super().__init__(meta=meta)\n\n        self.data: dict = {}\n        if mapping:\n            self.update(mapping)\n        self.update(kwargs)\n\n    def __setitem__(self, key, value):\n        if not isinstance(key, str):\n            # limitation for the interpolation\n            # ignore other kinds of keys\n            return\n        return super().__setitem__(key, value)\n\n    def merge_update(self, other, overwrite=False):\n        _merge(self, other, overwrite=overwrite)\n\n    @property\n    def value(self):\n        return {key: node.value for key, node in self.items()}\n\n    def __deepcopy__(self, _):\n        new = CtxDict()\n        for k, v in self.items():\n            new.data[k] = (\n                deepcopy(v) if isinstance(v, Container) else v\n            )  # short-circuiting __setitem__\n        return new\n\n\nclass Context(CtxDict):\n    def __init__(self, *args, **kwargs):\n        \"\"\"\n        Top level mutable dict, with some helpers to create context and track\n        \"\"\"\n        super().__init__(*args, **kwargs)\n        self._track = False\n        self._tracked_data: Dict[str, Dict] = defaultdict(dict)",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 545,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/context.py",
  "content": "        self.imports = {}\n        self._reserved_keys = {}\n\n    @contextmanager\n    def track(self):\n        self._track = True\n        yield self._tracked_data\n\n        self._track = False\n        self._tracked_data = defaultdict(dict)\n\n    def _track_data(self, node):\n        if not self._track or not isinstance(node, Node):\n            return\n\n        assert isinstance(node, Node)\n        if node.meta and node.meta.local:\n            return\n\n        for source, keys in node.get_sources().items():\n            if not source:\n                continue\n            params_file = self._tracked_data[source]\n            keys = [keys] if isinstance(keys, str) else keys\n            params_file.update({key: node.value for key in keys})\n\n    def select(\n        self, key: str, unwrap: bool = False\n    ):  # pylint: disable=arguments-differ\n        \"\"\"Select the item using key, similar to `__getitem__`\n           but can track the usage of the data on interpolation\n           as well and can get from nested data structure by using\n           \".\" separated key (eg: \"key1.key2.key3\")\n\n        Args:\n            key: key to select value from\n            unwrap: Convert CtxList/CtxDict/Value items to it's original data\n                    Defaults to False. Note that the default is different from\n                    `resolve`.\n        \"\"\"\n        normalized = normalize_key(key)\n        try:\n            node = super().select(normalized)\n        except ValueError as exc:\n            raise KeyNotInContext(key) from exc\n\n        assert isinstance(node, Node)\n        self._track_data(node)\n        return node.value if unwrap else node\n\n    @classmethod\n    def load_from(\n        cls, fs, path: str, select_keys: Optional[List[str]] = None\n    ) -> \"Context\":\n        from dvc.utils.serialize import load_path\n\n        if not fs.exists(path):\n            raise ParamsLoadError(f\"'{path}' does not exist\")\n        if fs.isdir(path):\n            raise ParamsLoadError(f\"'{path}' is a directory\")\n\n        data = load_path(path, fs)\n        if not isinstance(data, Mapping):\n            typ = type(data).__name__\n            raise ParamsLoadError(\n                f\"expected a dictionary, got '{typ}' in file '{path}'\"\n            )\n\n        if select_keys:\n            try:\n                data = {key: data[key] for key in select_keys}\n            except KeyError as exc:\n                key, *_ = exc.args\n                raise ParamsLoadError(f\"could not find '{key}' in '{path}'\") from exc\n\n        meta = Meta(source=path, local=False)\n        ctx = cls(data, meta=meta)\n        ctx.imports[path] = select_keys\n        return ctx\n\n    def merge_update(self, other: \"Context\", overwrite=False):\n        matches = select(lambda key: key in other, self._reserved_keys.keys())\n        if matches:\n            raise ReservedKeyError(matches)\n        return super().merge_update(other, overwrite=overwrite)\n\n    def merge_from(self, fs, item: str, wdir: str, overwrite=False):\n        path, _, keys_str = item.partition(\":\")\n        path = fs.path.normpath(fs.path.join(wdir, path))\n\n        select_keys = lfilter(bool, keys_str.split(\",\")) if keys_str else None\n        if path in self.imports:\n            if not select_keys and self.imports[path] is None:\n                return  # allow specifying complete filepath multiple times\n            self.check_loaded(path, item, select_keys)\n\n        ctx = Context.load_from(fs, path, select_keys)\n\n        try:\n            self.merge_update(ctx, overwrite=overwrite)\n        except ReservedKeyError as exc:\n            raise ReservedKeyError(exc.keys, item) from exc\n\n        cp = ctx.imports[path]\n        if path not in self.imports:\n            self.imports[path] = cp\n        elif cp:\n            self.imports[path].extend(cp)\n\n    def check_loaded(self, path, item, keys):\n        imported = self.imports[path]\n        if not keys and isinstance(imported, list):\n            raise VarsAlreadyLoaded(\n                f\"cannot load '{item}' as it's partially loaded already\"\n            )\n        if keys and imported is None:\n            raise VarsAlreadyLoaded(\n                f\"cannot partially load '{item}' as it's already loaded.\"\n            )\n        if isinstance(imported, list) and set(keys) & set(imported):\n            raise VarsAlreadyLoaded(\n                f\"cannot load '{item}' as it's partially loaded already\"\n            )\n\n    def load_from_vars(\n        self,\n        fs,\n        vars_: List,\n        wdir: str,\n        stage_name: Optional[str] = None,\n        default: Optional[str] = None,\n    ):\n        if default:\n            to_import = fs.path.join(wdir, default)\n            if fs.exists(to_import):\n                self.merge_from(fs, default, wdir)\n            else:\n                msg = \"%s does not exist, it won't be used in parametrization\"\n                logger.trace(msg, to_import)  # type: ignore[attr-defined]\n\n        stage_name = stage_name or \"\"\n        for index, item in enumerate(vars_):\n            assert isinstance(item, (str, dict))\n            if isinstance(item, str):\n                self.merge_from(fs, item, wdir)\n            else:\n                joiner = \".\" if stage_name else \"\"\n                meta = Meta(source=f\"{stage_name}{joiner}vars[{index}]\")\n                self.merge_update(Context(item, meta=meta))\n",
  "start_line": 301,
  "end_line": 450,
  "total_lines": 545,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/context.py",
  "content": "    def __deepcopy__(self, _):\n        new = Context(super().__deepcopy__(_))\n        new.meta = deepcopy(self.meta)\n        new.imports = deepcopy(self.imports)\n        new._reserved_keys = deepcopy(self._reserved_keys)\n        return new\n\n    @classmethod\n    def clone(cls, ctx: \"Context\") -> \"Context\":\n        \"\"\"Clones given context.\"\"\"\n        return deepcopy(ctx)\n\n    @contextmanager\n    def reserved(self, *keys: str):\n        \"\"\"Allow reserving some keys so that they cannot be overwritten.\n\n        Ideally, we should delegate this to a separate container\n        and support proper namespacing so that we could support `env` features.\n        But for now, just `item` and `key`, this should do.\n        \"\"\"\n        # using dict to make the error messages ordered\n        new = dict.fromkeys([key for key in keys if key not in self._reserved_keys])\n        self._reserved_keys.update(new)\n        try:\n            yield\n        finally:\n            for key in new:\n                self._reserved_keys.pop(key)\n\n    @contextmanager\n    def set_temporarily(self, to_set: DictStr, reserve: bool = False):\n        cm = self.reserved(*to_set) if reserve else nullcontext()\n\n        non_existing = frozenset(to_set.keys() - self.keys())\n        prev = {key: self[key] for key in to_set if key not in non_existing}\n        temp = CtxDict(to_set)\n        self.update(temp)\n\n        try:\n            with cm:\n                yield\n        finally:\n            self.update(prev)\n            for key in non_existing:\n                self.data.pop(key, None)\n\n    def resolve(\n        self, src, unwrap=True, skip_interpolation_checks=False, key=None\n    ) -> Any:\n        \"\"\"Recursively resolves interpolation and returns resolved data.\n\n        Args:\n            src: Data (str/list/dict etc.) to resolve\n            unwrap: Unwrap CtxDict/CtxList/Value to it's original data if\n                inside `src`. Defaults to True.\n            skip_interpolation_checks: Skip interpolation checks for error\n                The callee is responsible to check for errors in advance.\n\n        >>> c = Context({\"three\": 3})\n        >>> c.resolve({\"lst\": [1, 2, \"${three}\"]})\n        {'lst': [1, 2, 3]}\n        \"\"\"\n        func = recurse(self.resolve_str)\n        return func(src, unwrap, skip_interpolation_checks, key)\n\n    def resolve_str(\n        self, src: str, unwrap=True, skip_interpolation_checks=False, key=None\n    ) -> str:\n        \"\"\"Resolves interpolated string to it's original value,\n        or in case of multiple interpolations, a combined string.\n\n        >>> c = Context({\"enabled\": True})\n        >>> c.resolve_str(\"${enabled}\")\n        True\n        >>> c.resolve_str(\"enabled? ${enabled}\")\n        'enabled? true'\n        \"\"\"\n        matches = get_matches(src)\n        if is_exact_string(src, matches):\n            # replace \"${enabled}\", if `enabled` is a boolean, with it's actual\n            # value rather than it's string counterparts.\n            expr = get_expression(matches[0], skip_checks=skip_interpolation_checks)\n            value = self.select(expr, unwrap=unwrap)\n            validate_value(value, key)\n            return value\n        # but not \"${num} days\"\n        return str_interpolate(\n            src, matches, self, skip_checks=skip_interpolation_checks, key=key\n        )\n\n\nif __name__ == \"__main__\":\n    import doctest\n\n    doctest.testmod()",
  "start_line": 451,
  "end_line": 545,
  "total_lines": 545,
  "is_truncated": false
}