iterative__dvc-8904

{
  "status": "ok",
  "stdout": "dvc/repo/stage.py:230:        if accept_group and stages.is_foreach_generated(name):\ndvc/repo/stage.py:249:                foreach are returned.\ndvc/repo/stage.py:308:                - a foreach group name or a stage name in the `dvc.yaml` file.\ndvc/repo/stage.py:309:                - a generated stage name from a foreach group.\ndvc/repo/stage.py:356:        - foreach_group_name or stage_name\ndvc/repo/stage.py:357:        - generated stage name from a foreach group\ndvc/parsing/__init__.py:53:FOREACH_KWD = \"foreach\"\ndvc/parsing/__init__.py:108:def split_foreach_name(name: str) -> Tuple[str, Optional[str]]:\ndvc/parsing/__init__.py:170:        group, key = split_foreach_name(name)\ndvc/parsing/__init__.py:192:        return self._has_group_and_key(*split_foreach_name(key))\ndvc/parsing/__init__.py:258:            # but for foreach-generated ones, once is enough, which it does\ndvc/parsing/__init__.py:330:        self.foreach_data = definition[FOREACH_KWD]\ndvc/parsing/__init__.py:338:        # optimization: check for syntax errors only once for `foreach` stages\ndvc/parsing/__init__.py:344:        return self._resolve_foreach_data()\ndvc/parsing/__init__.py:346:    def _resolve_foreach_data(self) -> \"SeqOrMap\":\ndvc/parsing/__init__.py:348:            iterable = self.context.resolve(self.foreach_data, unwrap=False)\ndvc/parsing/__init__.py:350:            format_and_raise(exc, f\"'{self.where}.{self.name}.foreach'\", self.relpath)\ndvc/parsing/__init__.py:352:        # foreach data can be a resolved dictionary/list.\ndvc/parsing/__init__.py:354:        # foreach stages will have `item` and `key` added to the context\ndvc/parsing/__init__.py:366:                f\"failed to resolve '{self.where}.{self.name}.foreach'\"\ndvc/parsing/__init__.py:421:        err_message = f\"Could not find '{key}' in foreach group '{self.name}'\"\ndvc/parsing/__init__.py:425:        # NOTE: we need to use resolved iterable/foreach-data,\ndvc/parsing/__init__.py:446:                # optimization: skip checking for syntax errors on each foreach\ndvc/parsing/interpolate.py:201:    not_foreach = key is not None and \"foreach\" not in key\ndvc/parsing/interpolate.py:202:    if not_primitive and not_foreach:\ndvc/stage/loader.py:174:    def is_foreach_generated(self, name: str):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            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:\n            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)",
  "start_line": 100,
  "end_line": 249,
  "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": "        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\n    ) -> \"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",
  "start_line": 250,
  "end_line": 399,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            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)\n\n            # let mypy know that this state is unreachable as format_and_raise\n            # does not return at all (it's not able to understand it for some\n            # reason)\n            raise AssertionError(\"unreachable\")",
  "start_line": 400,
  "end_line": 455,
  "total_lines": 455,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 4, in <module>\nImportError: cannot import name 'GitFS' 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 'GitFS' from 'dvc.fs' (/workspace/dvc/fs/__init__.py)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "dvc/fs/__init__.py",
  "content": "from urllib.parse import urlparse\n\nfrom dvc_http import HTTPFileSystem, HTTPSFileSystem  # noqa: F401\n\nfrom dvc.config import ConfigError as RepoConfigError\nfrom dvc.config_schema import SCHEMA, Invalid\n\n# pylint: disable=unused-import\nfrom dvc_objects.fs import (  # noqa: F401\n    LocalFileSystem,\n    MemoryFileSystem,\n    Schemes,\n    generic,\n    get_fs_cls,\n    known_implementations,\n    localfs,\n    registry,\n    system,\n    utils,\n)\nfrom dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401\nfrom dvc_objects.fs.errors import (  # noqa: F401\n    AuthError,\n    ConfigError,\n    RemoteMissingDepsError,\n)\nfrom dvc_objects.fs.path import Path  # noqa: F401\n\nfrom .callbacks import Callback\nfrom .data import DataFileSystem  # noqa: F401\nfrom .dvc import DVCFileSystem  # noqa: F401\nfrom .git import GitFileSystem  # noqa: F401\n\nknown_implementations.update(\n    {\n        \"dvc\": {\n            \"class\": \"dvc.fs.dvc.DVCFileSystem\",\n            \"err\": \"dvc is supported, but requires 'dvc' to be installed\",\n        },\n        \"git\": {\n            \"class\": \"dvc.fs.git.GitFileSystem\",\n            \"err\": \"git is supported, but requires 'dvc' to be installed\",\n        },\n    }\n)\n\n\n# pylint: enable=unused-import\n\n\ndef download(fs, fs_path, to, jobs=None):\n    with Callback.as_tqdm_callback(\n        desc=f\"Downloading {fs.path.name(fs_path)}\",\n        unit=\"files\",\n    ) as cb:\n        fs.get(fs_path, to.fs_path, batch_size=jobs, callback=cb)\n\n\ndef parse_external_url(url, config=None):\n    remote_config = dict(config) if config else {}\n    remote_config[\"url\"] = url\n    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)\n    fs = fs_cls(**fs_config)\n    return fs, fs_path\n\n\ndef get_fs_config(config, **kwargs):\n    name = kwargs.get(\"name\")\n    if name:\n        try:\n            remote_conf = config[\"remote\"][name.lower()]\n        except KeyError:\n            from dvc.config import RemoteNotFoundError\n\n            raise RemoteNotFoundError(f\"remote '{name}' doesn't exist\")  # noqa: B904\n    else:\n        remote_conf = kwargs\n    return _resolve_remote_refs(config, remote_conf)\n\n\ndef _resolve_remote_refs(config, remote_conf):\n    # Support for cross referenced remotes.\n    # This will merge the settings, shadowing base ref with remote_conf.\n    # For example, having:\n    #\n    #       dvc remote add server ssh://localhost\n    #       dvc remote modify server user root\n    #       dvc remote modify server ask_password true\n    #\n    #       dvc remote add images remote://server/tmp/pictures\n    #       dvc remote modify images user alice\n    #       dvc remote modify images ask_password false\n    #       dvc remote modify images password asdf1234\n    #\n    # Results on a config dictionary like:\n    #\n    #       {\n    #           \"url\": \"ssh://localhost/tmp/pictures\",\n    #           \"user\": \"alice\",\n    #           \"password\": \"asdf1234\",\n    #           \"ask_password\": False,\n    #       }\n    parsed = urlparse(remote_conf[\"url\"])\n    if parsed.scheme != \"remote\":\n        return remote_conf\n\n    base = get_fs_config(config, name=parsed.netloc)\n    cls, _, _ = _get_cloud_fs(config, **base)\n    relpath = parsed.path.lstrip(\"/\").replace(\"/\", cls.sep)\n    url = cls.sep.join((base[\"url\"], relpath))\n    return {**base, **remote_conf, \"url\": url}\n\n\ndef get_cloud_fs(repo, **kwargs):\n    repo_config = repo.config if repo else {}\n    return _get_cloud_fs(repo_config, **kwargs)\n\n\ndef _get_cloud_fs(repo_config, **kwargs):\n    core_config = repo_config.get(\"core\", {})\n\n    remote_conf = get_fs_config(repo_config, **kwargs)\n    try:\n        remote_conf = SCHEMA[\"remote\"][str](remote_conf)  # type: ignore[index]\n    except Invalid as exc:\n        raise RepoConfigError(str(exc)) from None\n\n    if \"checksum_jobs\" not in remote_conf:\n        checksum_jobs = core_config.get(\"checksum_jobs\")\n        if checksum_jobs:\n            remote_conf[\"checksum_jobs\"] = checksum_jobs\n\n    cls = get_fs_cls(remote_conf)\n\n    url = remote_conf.pop(\"url\")\n    if cls.protocol in [\"webdav\", \"webdavs\"]:\n        # For WebDAVFileSystem, provided url is the base path itself, so it\n        # should be treated as being a root path.\n        fs_path = cls.root_marker\n    else:\n        fs_path = cls._strip_protocol(url)  # pylint:disable=protected-access\n\n    extras = cls._get_kwargs_from_urls(url)  # pylint:disable=protected-access\n    conf = {**extras, **remote_conf}  # remote config takes priority\n    return cls, conf, fs_path",
  "start_line": 1,
  "end_line": 145,
  "total_lines": 145,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"/workspace/dvc/parsing/__init__.py\", line 152, in __init__\n    self.context.load_from_vars(*args, default=DEFAULT_PARAMS_FILE)\n  File \"/workspace/dvc/parsing/context.py\", line 436, in load_from_vars\n    self.merge_from(fs, default, wdir)\n  File \"/workspace/dvc/parsing/context.py\", line 397, in merge_from\n    ctx = Context.load_from(fs, path, select_keys)\n  File \"/workspace/dvc/parsing/context.py\", line 360, in load_from\n    raise ParamsLoadError(f\"'{path}' is a directory\")\ndvc.parsing.context.ParamsLoadError: 'dvc.yaml' is a directory\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"<stdin>\", line 33, in <module>\n  File \"<stdin>\", line 26, in test_foreach_numeric_keys\n  File \"/workspace/dvc/parsing/__init__.py\", line 154, in __init__\n    format_and_raise(exc, \"'vars'\", self.relpath)\n  File \"/workspace/dvc/parsing/__init__.py\", line 81, in format_and_raise\n    _reraise_err(ResolveError, message, from_exc=exc)\n  File \"/workspace/dvc/parsing/__init__.py\", line 90, in _reraise_err\n    raise err\ndvc.parsing.ResolveError: failed to parse 'vars' in 'dvc.yaml': 'dvc.yaml' is a directory\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"/workspace/dvc/parsing/__init__.py\", line 152, in __init__\n    self.context.load_from_vars(*args, default=DEFAULT_PARAMS_FILE)\n  File \"/workspace/dvc/parsing/context.py\", line 436, in load_from_vars\n    self.merge_from(fs, default, wdir)\n  File \"/workspace/dvc/parsing/context.py\", line 397, in merge_from\n    ctx = Context.load_from(fs, path, select_keys)\n  File \"/workspace/dvc/parsing/context.py\", line 360, in load_from\n    raise ParamsLoadError(f\"'{path}' is a directory\")\ndvc.parsing.context.ParamsLoadError: 'dvc.yaml' is a directory\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"<stdin>\", line 33, in <module>\n  File \"<stdin>\", line 26, in test_foreach_numeric_keys\n  File \"/workspace/dvc/parsing/__init__.py\", line 154, in __init__\n    format_and_raise(exc, \"'vars'\", self.relpath)\n  File \"/workspace/dvc/parsing/__init__.py\", line 81, in format_and_raise\n    _reraise_err(ResolveError, message, from_exc=exc)\n  File \"/workspace/dvc/parsing/__init__.py\", line 90, in _reraise_err\n    raise err\ndvc.parsing.ResolveError: failed to parse 'vars' in 'dvc.yaml': 'dvc.yaml' is a directory\n",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 25, in <module>\n  File \"<stdin>\", line 17, in test_foreach_numeric_keys\nTypeError: __init__() missing 1 required positional argument: 'definition'\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 25, in <module>\n  File \"<stdin>\", line 17, in test_foreach_numeric_keys\nTypeError: __init__() missing 1 required positional argument: 'definition'\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            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:\n            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:",
  "start_line": 100,
  "end_line": 200,
  "total_lines": 455,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            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\n    ) -> \"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)",
  "start_line": 201,
  "end_line": 350,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "Error resolving build@2021: Could not find '2021' in foreach group 'build'\n",
  "stderr": "Traceback (most recent call last):\n  File \"/opt/miniconda3/envs/testbed/lib/python3.9/site-packages/funcy/flow.py\", line 84, in reraise\n    yield\n  File \"/workspace/dvc/parsing/__init__.py\", line 423, in _each_iter\n    value = self.normalized_iterable[key]\nKeyError: '2021'\n\nThe above exception was the direct cause of the following exception:\n\nTraceback (most recent call last):\n  File \"<stdin>\", line 35, in test_foreach_numeric_keys\n  File \"/workspace/dvc/parsing/__init__.py\", line 418, in resolve_one\n    return self._each_iter(key)\n  File \"/workspace/dvc/parsing/__init__.py\", line 423, in _each_iter\n    value = self.normalized_iterable[key]\n  File \"/opt/miniconda3/envs/testbed/lib/python3.9/contextlib.py\", line 137, in __exit__\n    self.gen.throw(typ, value, traceback)\n  File \"/opt/miniconda3/envs/testbed/lib/python3.9/site-packages/funcy/flow.py\", line 88, in reraise\n    raise into from e\ndvc.parsing.EntryNotFound: Could not find '2021' in foreach group 'build'\n",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "\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)\n\n            # let mypy know that this state is unreachable as format_and_raise\n            # does not return at all (it's not able to understand it for some\n            # reason)\n            raise AssertionError(\"unreachable\")",
  "start_line": 351,
  "end_line": 455,
  "total_lines": 455,
  "is_truncated": false
}

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

{
  "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)",
  "start_line": 1,
  "end_line": 100,
  "total_lines": 455,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Current: {2021: {'epochs': 3, 'thresh': 10}, 2022: {'epochs': 10, 'thresh': 15}}\nFixed: {'2021': {'epochs': 3, 'thresh': 10}, '2022': {'epochs': 10, 'thresh': 15}}\nFix verification successful!\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 36, in <module>\n  File \"<stdin>\", line 29, in test_foreach_numeric_keys_real\nAssertionError\n",
  "details": {
    "stdout": "Normalized: {2021: {'epochs': 3, 'thresh': 10}, 2022: {'epochs': 10, 'thresh': 15}}\n",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 36, in <module>\n  File \"<stdin>\", line 29, in test_foreach_numeric_keys_real\nAssertionError\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "        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\n    ) -> \"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:",
  "start_line": 200,
  "end_line": 349,
  "total_lines": 455,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dvc/parsing/__init__.py",
  "content": "            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)\n\n            # let mypy know that this state is unreachable as format_and_raise\n            # does not return at all (it's not able to understand it for some\n            # reason)\n            raise AssertionError(\"unreachable\")",
  "start_line": 350,
  "end_line": 455,
  "total_lines": 455,
  "is_truncated": false,
  "budget_warning": "Only 10 tool call(s) remaining (20/30 used). Finalize your edits and call submit_patch soon."
}