Compare commits

...
40 Commits
Author SHA1 Message Date
Ruslan Kuprieiev 43b371c43a deps: bump dvc-data to 2.11.0 2023-08-04 15:42:25 +03:00
skshetry d1160a28e0 dvc.yaml: add matrix-do (#9725)
* dvc.yaml: add matrix-do

* remove docs

* no do in matrix

* add types
2023-08-04 13:29:15 +05:45
skshetry bc86fa0643 completion: fix broken completion generation for zsh (#9797)
`dvc commit --help` was also completely broken, which is also fixed.

This adds unit tests for completion generation for both zsh/bash, and
tests that --help works for all commands.
2023-08-04 09:21:01 +05:45
Saugat Pachhai (सौगात) 4067a86926 fix flaky test for repro --keep-going and --ignore 2023-08-04 05:46:58 +03:00
Ruslan Kuprieiev 9d1d51caa7 deps: bump dvc-data to 2.10.0 2023-08-04 03:40:36 +03:00
Peter Rowlands 761422638f output: add progress for commit 2023-08-03 18:19:19 +09:00
Peter Rowlands d01e2355d4 push/pull: generate explicit progress bars for dvc-data transfers 2023-08-03 18:19:19 +09:00
Peter Rowlands aa94bfd66b deps: bump dvc-data to 2.9.1 2023-08-03 18:19:19 +09:00
skshetry b9b9f91084 ruff: enable autofix (#9790) 2023-08-03 11:08:29 +05:45
Ruslan Kuprieiev 8a382f51fd deps: bump dvc-data to 2.9.0 2023-08-03 02:05:14 +03:00
Ruslan Kuprieiev b235487ecc ls/import/get: introduce --remote and --remote-config
Fixes #9656
2023-08-02 16:28:16 +03:00
pre-commit-ci[bot]andSaugat Pachhai 502b3df6c9 [pre-commit.ci] pre-commit autoupdate (#9756)
* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/asottile/pyupgrade: v3.9.0 → v3.10.1](https://github.com/asottile/pyupgrade/compare/v3.9.0...v3.10.1)
- [github.com/astral-sh/ruff-pre-commit: v0.0.278 → v0.0.281](https://github.com/astral-sh/ruff-pre-commit/compare/v0.0.278...v0.0.281)
- [github.com/PyCQA/flake8: 6.0.0 → 6.1.0](https://github.com/PyCQA/flake8/compare/6.0.0...6.1.0)

* fix issues reported by ruff

- also remove redundant flake8 plugins
- remove beautysh as we don't have any scripts now
- remove pyupgrade as it is redundant due to ruff

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>
2023-08-02 18:08:30 +05:45
dependabot[bot] 5c2fd67e6d build(deps-dev): bump pylint from 2.17.4 to 2.17.5 (#9769)
Bumps [pylint](https://github.com/pylint-dev/pylint) from 2.17.4 to 2.17.5.
- [Release notes](https://github.com/pylint-dev/pylint/releases)
- [Commits](https://github.com/pylint-dev/pylint/compare/v2.17.4...v2.17.5)

---
updated-dependencies:
- dependency-name: pylint
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2023-07-31 16:35:40 +05:45
Fabian Zills ed34c0f0bd dvc commit update description (#9760)
* Update commit.py

* fix formatting
2023-07-31 08:45:52 +00:00
Ruslan Kuprieiev 7c60dfd3c5 deps: bump dvc-data to >=2.8.1 2023-07-30 04:18:09 +03:00
Ruslan Kuprieiev 225814fdde dvc: don't use realpath where not needed 2023-07-30 02:24:00 +03:00
Ruslan Kuprieiev ca0ad239f7 deps: bump dvc-data to 0.28.0 2023-07-28 01:37:46 +03:00
Ruslan Kuprieiev f57ac609e9 deps: bump dvc-data to 0.27.1 2023-07-27 20:44:34 +03:00
Ruslan Kuprieiev cc4f505f99 api: open/read: introduce remote_config 2023-07-26 16:29:14 +03:00
Ruslan Kuprieiev 4d011718a6 out: str: don't try to relpath paths outside of the repo
This is not very useful overall, but also results in a ValueError from
relpath on windows if your repo and output are on different drives.
2023-07-25 23:49:27 +03:00
Ruslan Kuprieiev 9c89841637 fetch: support passing config
For now not exposing it in CLI since there is no need/use yet.

Requirement for https://github.com/iterative/studio/pull/6541
2023-07-25 20:03:38 +03:00
Peter Rowlands 92cae74512 exp.collect: strip metrics from queued experiments 2023-07-25 20:57:15 +09:00
Ruslan Kuprieiev 73dad4780f repo: brancher: reset parsed fields
Related #9754 and fixes `dvc fetch`.
2023-07-25 14:00:33 +03:00
Ruslan Kuprieiev 65ba9ba527 fs: get_cloud_fs: accept config instead of repo
Config is a more immutable thing and using `repo` is confusing and might be error
prone in the bigger picture.

Related to #9754
2023-07-25 00:58:47 +03:00
Peter Rowlands 7806afd4ad fs: prefer generic.copy over fsspec fs.get() 2023-07-24 17:01:34 +09:00
Ivan Shcheklein a9d0ac2ba9 fix(exceptions): add parent ctr init in InvalidArgumentError 2023-07-21 08:27:53 +02:00
Ruslan Kuprieiev 41df96fcdb cli: ls/import/get: fix help
Leftover from https://github.com/iterative/dvc/pull/9747
2023-07-20 21:53:14 +03:00
Ruslan Kuprieiev 53ec0fa41d ls/import/get: introduce --config
Adds support for `--config` to all three commands. For `ls` and `get`, it is
a purely runtime option that will merge the config you provide with target
repo's configs (similar to how .dvc/config.local works). E.g.

```
$ cat myconfig
[remote "myremote"]
    access_key_id = 12345
    secret_access_key = 98765

$ dvc get https://github.com/efiop/mydataregistry recent-grads.csv --config myconfig
```

In case of `dvc import`, the semantics are the same, but the value is also
recorded in the resulting dvcfile. E.g.

```
$ dvc import https://github.com/efiop/mydataregistry recent-grads.csv --config myconfig
...
$ cat recent-grads.csv.dvc
md5: 85c86eeb3afa6d1f7d11beedc644d28e
frozen: true
deps:
- path: recent-grads.csv
  repo:
    url: https://github.com/efiop/mydataregistry
    config: myconfig
    rev_lock: af22e06bbfe34d62a140437abf32f48b535944a7
outs:
- md5: f447e23a170a9f25a7799f069a1ba934
  size: 26872
  hash: md5
  path: recent-grads.csv
```

This is the most general and powerful mechanism that we can give, that works
for everything from reconfiguring your default remote to completely changing
your remote layout and beyond (will be handy in the future).

Fixes #2466
2023-07-20 18:41:10 +03:00
Dave BerenbaumandRuslan Kuprieiev 41194b7975 dvcfs: handle NotADirectoryError from datafs.ls (#9746)
* dvc_fs: ignore NotADirectoryError in ls

* deps: bump dvc-data to >= 2.6.0

* add test

---------

Co-authored-by: Ruslan Kuprieiev <[email protected]>
2023-07-19 20:54:55 +03:00
skshetry ea5f7d40c1 ruff: enable a few more rules, and merge tests' config with the main config (#9739) 2023-07-15 14:33:34 +05:45
dependabot[bot] eb06351062 build(deps-dev): bump mypy from 1.3.0 to 1.4.1 (#9664)
Bumps [mypy](https://github.com/python/mypy) from 1.3.0 to 1.4.1.
- [Commits](https://github.com/python/mypy/compare/v1.3.0...v1.4.1)

---
updated-dependencies:
- dependency-name: mypy
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2023-07-14 20:30:47 +05:45
pre-commit-ci[bot]andSaugat Pachhai ca3704997b [pre-commit.ci] pre-commit autoupdate (#9632)
* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/asottile/pyupgrade: v3.3.1 → v3.9.0](https://github.com/asottile/pyupgrade/compare/v3.3.1...v3.9.0)
- https://github.com/charliermarsh/ruff-pre-commit → https://github.com/astral-sh/ruff-pre-commit
- [github.com/astral-sh/ruff-pre-commit: v0.0.261 → v0.0.277](https://github.com/astral-sh/ruff-pre-commit/compare/v0.0.261...v0.0.277)
- [github.com/codespell-project/codespell: v2.2.4 → v2.2.5](https://github.com/codespell-project/codespell/compare/v2.2.4...v2.2.5)

* fix ruff complaints

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>
2023-07-14 20:11:15 +05:45
Saugat Pachhai (सौगात) f1764bdc77 stage add: remove --external 2023-07-14 15:38:57 +03:00
skshetry eb212a7ba3 repro: introduce --keep-going/--ignore to continue on error (#9712)
* repro: introduce --keep-going/--ignore to continue on error

`-k`/`--keep-going` will skip all dependents of a stage if it fails.
It'll still try to reproduce the other dependencies of the pending
targets, before it exits with a non-zero status.

`--ignore-errors` will treat any error during stage execution as a
success, and will continue executing other dependencies as if it
succeeded. Exits with 0 even on failure.

Also fixes:
* --force-downstream used to force everything even if one upstream nodes changed.
  This used to work when we supported a single targets, but there could be independent
  pipelines, so  we should not force everything if any node changes.
  Now we'll force reproduce the stage only if one of its dependency changed upstream
  or it's upstream node was also forced.

* Makes `--downstream` more correct, in relation to `--allow-missing` and `--pull`.
  Now we use an active graph for downstream/upstream logic instead of a subgraph.
  Note that this does not affect the downstream logic though, as dvc always goes
  downstream, so subgraph was no different than active graph.
  The subgraph only did not have the upstream nodes.

* add tests

* add spacing in the message
2023-07-12 21:52:41 +05:45
skshetry 4c8f539c56 plots: log failures during datapoint conversions (#9726)
Fixes #9441.

Note that we do merge props, and use that for all revision. So you may
see warning for multiple revisions.
2023-07-12 19:14:31 +05:45
skshetry 45aefc8423 parsing: pass config instead of loading every time (#9729) 2023-07-12 17:22:42 +05:45
Peter Rowlands 67a747a356 stage.cache: override both legacy+local cache link types on save/commit 2023-07-12 17:17:49 +09:00
Peter Rowlands a5dda1b8ad open_repo: return without entering context for both local and erepo 2023-07-12 14:45:11 +09:00
Peter Rowlands dd4035bc05 repo: add get_data_index_entry to handle subrepo lookup 2023-07-12 14:45:11 +09:00
Peter Rowlands f69b3533e4 erepo: drop dvcx for_write support 2023-07-12 14:45:11 +09:00
111 changed files with 1504 additions and 587 deletions
+8 -7
View File
@@ -1,19 +1,20 @@
[flake8]
ignore=
max_line_length = 88
extend-ignore=
# Whitespace before ':'
E203
# Too many leading '#' for block comment
E266
# Line break occurred before a binary operator
W503
# unindexed parameters in the str.format, see:
# https://pypi.org/project/flake8-string-format/
P1
# use `!r` instead of surrounding by quotes
B028
max_line_length = 88
max-complexity = 10
select = B,C,E,F,W,T4,B902,T,P,TYP
show_source = true
count = true
min_python_version = 3.8.0
per-file-ignores =
dvc/testing/benchmarks/conftest.py:F401
dvc/testing/conftest.py:F401
dvc/testing/plugin.py:F401
tests/conftest.py:F401
tests/func/experiments/executor/test_ssh.py:F401
+6 -19
View File
@@ -23,42 +23,29 @@ repos:
- id: mixed-line-ending
- id: sort-simple-yaml
- id: trailing-whitespace
- repo: https://github.com/lovesegfault/beautysh
rev: v6.2.1
hooks:
- id: beautysh
args: ["-i", "2"]
- repo: https://github.com/asottile/pyupgrade
rev: v3.3.1
hooks:
- id: pyupgrade
args:
- --py38-plus
- repo: https://github.com/psf/black
rev: 23.3.0
rev: 23.7.0
hooks:
- id: black
- repo: https://github.com/PyCQA/isort
rev: 5.12.0
hooks:
- id: isort
- repo: https://github.com/charliermarsh/ruff-pre-commit
rev: 'v0.0.261'
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: 'v0.0.282'
hooks:
- id: ruff
args: [--fix, --exit-non-zero-on-fix]
- repo: https://github.com/codespell-project/codespell
rev: v2.2.4
rev: v2.2.5
hooks:
- id: codespell
additional_dependencies: ["tomli"]
- repo: https://github.com/PyCQA/flake8
rev: 6.0.0
rev: 6.1.0
hooks:
- id: flake8
additional_dependencies:
- flake8-bugbear
- flake8-comprehensions
- flake8-debugger
- flake8-string-format
- flake8-typing-imports
- repo: https://github.com/PyCQA/bandit
+36 -26
View File
@@ -45,22 +45,13 @@ def get_url(path, repo=None, rev=None, remote=None):
with Repo.open(
repo, rev=rev, subrepos=True, uninitialized=True, **repo_kwargs
) as _repo:
with _wrap_exceptions(_repo, path):
fs_path = _repo.dvcfs.from_os_path(path)
fs = _repo.dvcfs.fs
# pylint: disable-next=protected-access
key = fs._get_key_from_relative(fs_path)
# pylint: disable-next=protected-access
subrepo, _, subkey = fs._get_subrepo_info(key)
index = subrepo.index.data["repo"]
with reraise(KeyError, OutputNotFoundError(path, repo)):
entry = index[subkey]
with reraise(
(StorageKeyError, ValueError),
NoRemoteError(f"no remote specified in {_repo}"),
):
remote_fs, remote_path = index.storage_map.get_remote(entry)
return remote_fs.unstrip_protocol(remote_path)
index, entry = _repo.get_data_index_entry(path)
with reraise(
(StorageKeyError, ValueError),
NoRemoteError(f"no remote specified in {_repo}"),
):
remote_fs, remote_path = index.storage_map.get_remote(entry)
return remote_fs.unstrip_protocol(remote_path)
class _OpenContextManager(GCM):
@@ -76,7 +67,7 @@ class _OpenContextManager(GCM):
raise AttributeError("dvc.api.open() should be used in a with statement.")
def open( # noqa, pylint: disable=redefined-builtin
def open( # noqa: A001, pylint: disable=redefined-builtin
path: str,
repo: Optional[str] = None,
rev: Optional[str] = None,
@@ -84,6 +75,7 @@ def open( # noqa, pylint: disable=redefined-builtin
mode: str = "r",
encoding: Optional[str] = None,
config: Optional[Dict[str, Any]] = None,
remote_config: Optional[Dict[str, Any]] = None,
):
"""
Opens a file tracked in a DVC project.
@@ -127,6 +119,9 @@ def open( # noqa, pylint: disable=redefined-builtin
Mirrors the namesake parameter in builtin `open()`_.
config(dict, optional): config to be passed to the DVC repository.
Defaults to None.
remote_config(dict, optional): remote config to be passed to the DVC
repository.
Defaults to None.
Returns:
_OpenContextManager: A context manager that generatse a corresponding
@@ -223,22 +218,27 @@ def open( # noqa, pylint: disable=redefined-builtin
"mode": mode,
"encoding": encoding,
"config": config,
"remote_config": remote_config,
}
return _OpenContextManager(_open, args, kwargs)
def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
if remote:
if config is not None:
raise ValueError(
"can't specify both `remote` and `config` at the same time"
)
config = {"core": {"remote": remote}}
def _open(
path,
repo=None,
rev=None,
remote=None,
mode="r",
encoding=None,
config=None,
remote_config=None,
):
repo_kwargs: Dict[str, Any] = {
"subrepos": True,
"uninitialized": True,
"remote": remote,
"config": config,
"remote_config": remote_config,
}
with Repo.open(repo, rev=rev, **repo_kwargs) as _repo:
@@ -274,7 +274,16 @@ def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, confi
raise DvcIsADirectoryError(f"'{path}' is a directory") from exc
def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
def read(
path,
repo=None,
rev=None,
remote=None,
mode="r",
encoding=None,
config=None,
remote_config=None,
):
"""
Returns the contents of a tracked file (by DVC or Git). For Git repos, HEAD
is used unless a rev argument is supplied. The default remote is tried
@@ -288,5 +297,6 @@ def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config
mode=mode,
encoding=encoding,
config=config,
remote_config=remote_config,
) as fd:
return fd.read()
+1 -1
View File
@@ -24,7 +24,7 @@ def _get_odb(
if not settings:
return None
cls, config, fs_path = get_cloud_fs(repo, **settings)
cls, config, fs_path = get_cloud_fs(repo.config, **settings)
fs = fs or cls(**config)
if prefix:
fs_path = fs.path.join(fs_path, *prefix)
+2 -2
View File
@@ -192,7 +192,7 @@ def main(argv=None): # noqa: C901, PLR0912, PLR0915
from dvc import PKG, __version__
pyv = " ".join([python_implementation(), python_version()])
pyv = f"{python_implementation()} {python_version()}"
pkg = f" ({PKG})" if PKG else ""
logger.debug("v%s%s, %s on %s", __version__, pkg, pyv, platform())
logger.debug("command: %s", " ".join(argv or sys.argv))
@@ -230,7 +230,7 @@ def main(argv=None): # noqa: C901, PLR0912, PLR0915
logger.exception("")
except DvcParserError:
ret = 254
except Exception as exc: # noqa, pylint: disable=broad-except
except Exception as exc: # noqa: BLE001, pylint: disable=broad-except
# pylint: disable=no-member
ret = _log_exceptions(exc) or 255
+25 -3
View File
@@ -1,3 +1,27 @@
import argparse
class DictAction(argparse.Action):
def __call__(self, parser, args, values, option_string=None): # noqa: ARG002
d = getattr(args, self.dest) or {}
if isinstance(values, list):
kvs = values
else:
kvs = [values]
for kv in kvs:
key, value = kv.split("=")
if not value:
raise argparse.ArgumentError(
self,
f'Could not parse argument "{values}" as k1=v1 k2=v2 ... format',
)
d[key] = value
setattr(args, self.dest, d)
def fix_subparsers(subparsers):
"""Workaround for bug in Python 3. See more info at:
https://bugs.python.org/issue16308
@@ -16,9 +40,7 @@ def append_doc_link(help_message, path):
if not path:
return help_message
doc_base = "https://man.dvc.org/"
return "{message}\nDocumentation: {link}".format(
message=help_message, link=format_link(doc_base + path)
)
return f"{help_message}\nDocumentation: {format_link(doc_base + path)}"
def hide_subparsers_from_help(subparsers):
+4 -1
View File
@@ -47,7 +47,10 @@ def add_parser(subparsers, parent_parser):
"--force",
action="store_true",
default=False,
help="Commit even if hash value for dependencies/outputs changed.",
help=(
"Commit data even if hash values for dependencies or "
"outputs did not change."
),
)
commit_parser.add_argument(
"-d",
+4 -1
View File
@@ -9,6 +9,9 @@ from dvc.ui import ui
logger = logging.getLogger(__name__)
SUPPORTED_SHELLS = ["bash", "zsh"]
class CmdCompletion(CmdBaseNoRepo):
def run(self):
import shtab
@@ -35,6 +38,6 @@ def add_parser(subparsers, parent_parser):
"--shell",
help="Shell syntax for completions.",
default="bash",
choices=["bash", "zsh"],
choices=SUPPORTED_SHELLS,
)
completion_parser.set_defaults(func=CmdCompletion)
+1 -3
View File
@@ -90,9 +90,7 @@ class CmdDataStatus(CmdBase):
ui.write(f" ({hint})")
if isinstance(stage_status, dict):
items = [
": ".join([state, file]) for file, state in stage_status.items()
]
items = [f"{state}: {file}" for file, state in stage_status.items()]
else:
items = stage_status
+1 -1
View File
@@ -21,7 +21,7 @@ class CmdDestroy(CmdBase):
)
if not self.args.force and not ui.confirm(statement):
raise DvcException(
raise DvcException( # noqa: TRY301
"cannot destroy without a confirmation from the user."
" Use `-f` to force."
)
+27 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBaseNoRepo
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.exceptions import DvcException
logger = logging.getLogger(__name__)
@@ -37,6 +37,9 @@ class CmdGet(CmdBaseNoRepo):
rev=self.args.rev,
jobs=self.args.jobs,
force=self.args.force,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
return 0
except CloneError:
@@ -102,4 +105,27 @@ def add_parser(subparsers, parent_parser):
default=False,
help="Override local file or folder if exists.",
)
get_parser.add_argument(
"--config",
type=str,
help=(
"Path to a config file that will be merged with the config "
"in the target repository."
),
)
get_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
get_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
get_parser.set_defaults(func=CmdGet)
+27 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBase
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.exceptions import DvcException
logger = logging.getLogger(__name__)
@@ -22,6 +22,9 @@ class CmdImport(CmdBase):
no_exec=self.args.no_exec,
no_download=self.args.no_download,
jobs=self.args.jobs,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
except CloneError:
logger.exception("failed to import '%s'", self.args.path)
@@ -94,4 +97,27 @@ def add_parser(subparsers, parent_parser):
),
metavar="<number>",
)
import_parser.add_argument(
"--config",
type=str,
help=(
"Path to a config file that will be merged with the config "
"in the target repository."
),
)
import_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
import_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
import_parser.set_defaults(func=CmdImport)
+27 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBaseNoRepo
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.commands.ls.ls_colors import LsColors
from dvc.exceptions import DvcException
from dvc.ui import ui
@@ -34,6 +34,9 @@ class CmdList(CmdBaseNoRepo):
rev=self.args.rev,
recursive=self.args.recursive,
dvc_only=self.args.dvc_only,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
if self.args.json:
ui.write_json(entries)
@@ -80,6 +83,29 @@ def add_parser(subparsers, parent_parser):
help="Git revision (e.g. SHA, branch, tag)",
metavar="<commit>",
)
list_parser.add_argument(
"--config",
type=str,
help=(
"Path to a config file that will be merged with the config "
"in the target repository."
),
)
list_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
list_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
list_parser.add_argument(
"path",
nargs="?",
+1 -3
View File
@@ -127,9 +127,7 @@ class CmdRemoteRename(CmdRemote):
all_config = self.config.load_config_to_level(None)
if self.args.new in all_config.get("remote", {}):
raise ConfigError(
"Rename failed. Remote name '{}' already exists.".format(
{self.args.new}
)
f"Rename failed. Remote name {self.args.new!r} already exists."
)
with self.config.edit(self.args.level) as conf:
+21
View File
@@ -33,6 +33,7 @@ class CmdRepro(CmdBase):
"force_downstream": self.args.force_downstream,
"pull": self.args.pull,
"allow_missing": self.args.allow_missing,
"on_error": self.args.on_error,
}
@property
@@ -137,6 +138,26 @@ and then the stage name name.
"Only print the commands that would be executed without actually executing."
),
)
repro_parser.add_argument(
"-k",
"--keep-going",
action="store_const",
default="fail",
const="keep-going",
dest="on_error",
help=(
"Continue executing, skipping stages having dependencies "
"on the failed stages"
),
)
repro_parser.add_argument(
"--ignore-errors",
action="store_const",
default="fail",
const="ignore",
dest="on_error",
help="Ignore errors from stages.",
)
def add_parser(subparsers, parent_parser):
-6
View File
@@ -195,12 +195,6 @@ def _add_common_args(parser):
help="Declare output file or directory (do not put into DVC cache).",
metavar="<filename>",
).complete = completion.FILE
parser.add_argument(
"--external",
action="store_true",
default=False,
help="Allow outputs that are outside of the DVC repository.",
)
parser.add_argument(
"--outs-persist",
action="append",
+4 -5
View File
@@ -57,7 +57,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
return self._columns[name]
def items(self) -> ItemsView[str, Column]:
projection = {k: self.column(k) for k in self.keys()} # noqa: SIM118
projection = {k: self.column(k) for k in self.keys()}
return projection.items()
def keys(self) -> List[str]:
@@ -171,8 +171,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
self.add_column(key)
row: List["CellT"] = [
with_value(d.get(key), self._fill_value)
for key in self.keys() # noqa: SIM118
with_value(d.get(key), self._fill_value) for key in self.keys()
]
self.append(row)
@@ -227,7 +226,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
to_drop -= match_line
if axis == "rows":
for name in self.keys(): # noqa: SIM118
for name in self.keys():
self._columns[name] = Column(
[x for n, x in enumerate(self._columns[name]) if n not in to_drop]
)
@@ -275,7 +274,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
else:
unique_rows.append(tuple_row)
for name in self.keys(): # noqa: SIM118
for name in self.keys():
self._columns[name] = Column(
[
x
+45 -15
View File
@@ -93,6 +93,8 @@ class Config(dict):
validate: bool = True,
fs: Optional["FileSystem"] = None,
config: Optional["DictStrAny"] = None,
remote: Optional[str] = None,
remote_config: Optional["DictStrAny"] = None,
): # pylint: disable=super-init-not-called
from dvc.fs import LocalFileSystem
@@ -101,9 +103,11 @@ class Config(dict):
self.fs = fs or self.wfs
if dvc_dir:
self.dvc_dir = self.fs.path.abspath(self.fs.path.realpath(dvc_dir))
self.dvc_dir = self.fs.path.abspath(dvc_dir)
self.load(validate=validate, config=config)
self.load(
validate=validate, config=config, remote=remote, remote_config=remote_config
)
@classmethod
def from_cwd(cls, fs: Optional["FileSystem"] = None, **kwargs):
@@ -154,7 +158,13 @@ class Config(dict):
with open(config_file, "w+", encoding="utf-8"):
return Config(dvc_dir)
def load(self, validate: bool = True, config: Optional["DictStrAny"] = None):
def load(
self,
validate: bool = True,
config: Optional["DictStrAny"] = None,
remote: Optional[str] = None,
remote_config: Optional["DictStrAny"] = None,
):
"""Loads config from all the config files.
Raises:
@@ -169,6 +179,17 @@ class Config(dict):
conf = self.validate(conf)
self.clear()
if remote:
conf["core"]["remote"] = remote
if remote_config:
remote = remote or conf["core"].get("remote")
if not remote:
raise ValueError("Missing remote name")
merge(conf, {"remote": {remote: remote_config}})
self.update(conf)
def _get_fs(self, level):
@@ -176,23 +197,32 @@ class Config(dict):
# the repo.
return self.fs if level == "repo" else self.wfs
def _load_config(self, level):
@staticmethod
def load_file(path, fs=None) -> dict:
from configobj import ConfigObj, ConfigObjError
from dvc.fs import localfs
fs = fs or localfs
with fs.open(path) as fobj:
try:
conf_obj = ConfigObj(fobj)
except UnicodeDecodeError as exc:
raise ConfigError(str(exc)) from exc
except ConfigObjError as exc:
raise ConfigError(str(exc)) from exc
return _parse_named(_lower_keys(conf_obj.dict()))
def _load_config(self, level):
filename = self.files[level]
fs = self._get_fs(level)
if fs.exists(filename):
with fs.open(filename) as fobj:
try:
conf_obj = ConfigObj(fobj)
except UnicodeDecodeError as exc:
raise ConfigError(str(exc)) from exc
except ConfigObjError as exc:
raise ConfigError(str(exc)) from exc
else:
conf_obj = ConfigObj()
return _parse_named(_lower_keys(conf_obj.dict()))
try:
return self.load_file(filename, fs=fs)
except FileNotFoundError:
return {}
def _save_config(self, level, conf_dict):
from configobj import ConfigObj
+1 -1
View File
@@ -75,7 +75,7 @@ def ByUrl(mapping): # noqa: N802
class RelPath(str):
pass
__slots__ = ()
class FeatureSchema(Schema):
+3 -1
View File
@@ -28,7 +28,9 @@ def _popen(cmd, **kwargs) -> Popen:
os.path.abspath(os.path.dirname(__file__)), "__main__.py"
)
prefix += [main_entrypoint]
return Popen(prefix + cmd, close_fds=True, shell=False, **kwargs) # nosec B603
return Popen(
prefix + cmd, close_fds=True, shell=False, **kwargs # nosec B603 # noqa: S603
)
def _spawn_windows(cmd, env):
+31 -20
View File
@@ -4,6 +4,7 @@ import logging
from typing import TYPE_CHECKING, Iterable, Optional, Set, Tuple
from dvc.config import NoRemoteError, RemoteConfigError
from dvc.fs.callbacks import Callback
from dvc.utils.objects import cached_property
from dvc_data.hashfile.db import get_index
from dvc_data.hashfile.transfer import TransferResult
@@ -90,7 +91,7 @@ class DataCloud:
if name:
from dvc.fs import get_cloud_fs
cls, config, fs_path = get_cloud_fs(self.repo, name=name)
cls, config, fs_path = get_cloud_fs(self.repo.config, name=name)
if config.get("worktree"):
version_aware = config.get("version_aware")
@@ -209,15 +210,20 @@ class DataCloud:
cache = self.repo.cache.legacy
else:
cache = self.repo.cache.local
return self.transfer(
cache,
odb,
objs,
jobs=jobs,
dest_index=get_index(odb),
cache_odb=cache,
validate_status=self._log_missing,
)
with Callback.as_tqdm_callback(
desc=f"Pushing to {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
return self.transfer(
cache,
odb,
objs,
jobs=jobs,
dest_index=get_index(odb),
cache_odb=cache,
validate_status=self._log_missing,
callback=cb,
)
def pull(
self,
@@ -263,16 +269,21 @@ class DataCloud:
cache = self.repo.cache.legacy
else:
cache = self.repo.cache.local
return self.transfer(
odb,
cache,
objs,
jobs=jobs,
src_index=get_index(odb),
cache_odb=cache,
verify=odb.verify,
validate_status=self._log_missing,
)
with Callback.as_tqdm_callback(
desc=f"Fetching from {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
return self.transfer(
odb,
cache,
objs,
jobs=jobs,
src_index=get_index(odb),
cache_odb=cache,
verify=odb.verify,
validate_status=self._log_missing,
callback=cb,
)
def status(
self,
+1 -1
View File
@@ -50,7 +50,7 @@ class Dependency(Output):
self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
def download(self, to, jobs=None):
fs_download(self.fs, self.fs_path, to, jobs=jobs)
fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
def save(self):
super().save()
+45 -3
View File
@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING, Dict, Optional, Union
from voluptuous import Required
from voluptuous import Any, Required
from dvc.utils import as_posix
@@ -16,12 +16,16 @@ class RepoDependency(Dependency):
PARAM_URL = "url"
PARAM_REV = "rev"
PARAM_REV_LOCK = "rev_lock"
PARAM_CONFIG = "config"
PARAM_REMOTE = "remote"
REPO_SCHEMA = {
PARAM_REPO: {
Required(PARAM_URL): str,
PARAM_REV: str,
PARAM_REV_LOCK: str,
PARAM_CONFIG: str,
PARAM_REMOTE: Any(str, dict),
}
}
@@ -60,7 +64,28 @@ class RepoDependency(Dependency):
self.def_repo[self.PARAM_REV_LOCK] = rev
def dumpd(self, **kwargs) -> Dict[str, Union[str, Dict[str, str]]]:
return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
repo = {self.PARAM_URL: self.def_repo[self.PARAM_URL]}
rev = self.def_repo.get(self.PARAM_REV)
if rev:
repo[self.PARAM_REV] = self.def_repo[self.PARAM_REV]
rev_lock = self.def_repo.get(self.PARAM_REV_LOCK)
if rev_lock:
repo[self.PARAM_REV_LOCK] = rev_lock
config = self.def_repo.get(self.PARAM_CONFIG)
if config:
repo[self.PARAM_CONFIG] = config
remote = self.def_repo.get(self.PARAM_REMOTE)
if remote:
repo[self.PARAM_REMOTE] = remote
return {
self.PARAM_PATH: self.def_path,
self.PARAM_REPO: repo,
}
def update(self, rev: Optional[str] = None):
if rev:
@@ -77,9 +102,24 @@ class RepoDependency(Dependency):
def _make_fs(
self, rev: Optional[str] = None, locked: bool = True
) -> "DVCFileSystem":
from dvc.config import Config
from dvc.fs import DVCFileSystem
config = {"cache": self.repo.config["cache"]}
rem = self.def_repo.get("remote")
if isinstance(rem, dict):
remote = None # type: ignore[unreachable]
remote_config = rem
else:
remote = rem
remote_config = None
conf = self.def_repo.get("config")
if conf:
config = Config.load_file(conf)
else:
config = {}
config["cache"] = self.repo.config["cache"]
config["cache"]["dir"] = self.repo.cache.local_cache_dir
return DVCFileSystem(
@@ -87,6 +127,8 @@ class RepoDependency(Dependency):
rev=rev or self._get_rev(locked=locked),
subrepos=True,
config=config,
remote=remote,
remote_config=remote_config,
)
def _get_rev(self, locked: bool = True):
+1 -3
View File
@@ -97,9 +97,7 @@ class FileMixin:
self.verify = verify
def __repr__(self):
return "{}: {}".format(
self.__class__.__name__, relpath(self.path, self.repo.root_dir)
)
return f"{self.__class__.__name__}: {relpath(self.path, self.repo.root_dir)}"
def __hash__(self):
return hash(self.path)
+13 -15
View File
@@ -17,6 +17,10 @@ class DvcException(Exception):
class InvalidArgumentError(ValueError, DvcException):
"""Thrown if arguments are invalid."""
def __init__(self, msg, *args):
self.msg = msg
super().__init__(msg, *args)
class OutputDuplicationError(DvcException):
"""Thrown if a file/directory is specified as an output in more than one
@@ -63,12 +67,14 @@ class OutputNotFoundError(DvcException):
self.output = output
self.repo = repo
super().__init__(
"Unable to find DVC file with output '{path}'".format(
path=relpath(self.output)
)
f"Unable to find DVC file with output {relpath(self.output)!r}"
)
class StageNotFoundError(DvcException):
pass
class StagePathAsOutputError(DvcException):
"""Thrown if directory that stage is going to be saved in is specified as
an output of another stage.
@@ -80,11 +86,7 @@ class StagePathAsOutputError(DvcException):
def __init__(self, stage, output):
assert isinstance(output, str)
super().__init__(
"{stage} is within an output '{output}' of another stage".format(
stage=stage, output=output
)
)
super().__init__(f"{stage} is within an output {output!r} of another stage")
class CircularDependencyError(DvcException):
@@ -156,9 +158,7 @@ class CyclicGraphError(DvcException):
class ConfirmRemoveError(DvcException):
def __init__(self, path):
super().__init__(
"unable to remove '{}' without a confirmation. Use `-f` to force.".format(
path
)
f"unable to remove {path!r} without a confirmation. Use `-f` to force."
)
@@ -167,9 +167,7 @@ class InitError(DvcException):
class ReproductionError(DvcException):
def __init__(self, name):
self.name = name
super().__init__(f"failed to reproduce '{name}'")
pass
class BadMetricError(DvcException):
@@ -299,7 +297,7 @@ class URLMissingError(DvcException):
super().__init__(f"The path '{url}' does not exist")
class IsADirectoryError(DvcException): # noqa,pylint:disable=redefined-builtin
class IsADirectoryError(DvcException): # noqa: A001,pylint:disable=redefined-builtin
"""Raised when a file operation is requested on a directory."""
+35 -11
View File
@@ -1,3 +1,4 @@
from typing import Optional
from urllib.parse import urlparse
from dvc_http import HTTPFileSystem, HTTPSFileSystem # noqa: F401
@@ -18,7 +19,7 @@ from dvc_objects.fs import ( # noqa: F401
system,
utils,
)
from dvc_objects.fs.base import AnyFSPath, FileSystem # noqa: F401
from dvc_objects.fs.base import AnyFSPath, FileSystem # noqa: F401, TCH001
from dvc_objects.fs.errors import ( # noqa: F401
AuthError,
ConfigError,
@@ -48,18 +49,45 @@ known_implementations.update(
# pylint: enable=unused-import
def download(fs, fs_path, to, jobs=None):
def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None):
with Callback.as_tqdm_callback(
desc=f"Downloading {fs.path.name(fs_path)}",
unit="files",
) as cb:
fs.get(fs_path, to.fs_path, batch_size=jobs, callback=cb)
# NOTE: We use dvc-objects generic.copy over fs.get since it makes file
# download atomic and avoids fsspec glob/regex path expansion.
if fs.isdir(fs_path):
from_infos = [
path
for path in fs.find(fs_path)
if not path.endswith(fs.path.flavour.sep)
]
if not from_infos:
return localfs.makedirs(to, exist_ok=True)
to_infos = [
localfs.path.join(to, *fs.path.relparts(info, fs_path))
for info in from_infos
]
else:
from_infos = [fs_path]
to_infos = [to]
cb.set_size(len(from_infos))
jobs = jobs or fs.jobs
generic.copy(
fs,
from_infos,
localfs,
to_infos,
callback=cb,
batch_size=jobs,
)
def parse_external_url(url, config=None):
remote_config = dict(config) if config else {}
remote_config["url"] = url
fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
fs_cls, fs_config, fs_path = get_cloud_fs({}, **remote_config)
fs = fs_cls(**fs_config)
return fs, fs_path
@@ -105,18 +133,14 @@ def _resolve_remote_refs(config, remote_conf):
return remote_conf
base = get_fs_config(config, name=parsed.netloc)
cls, _, _ = _get_cloud_fs(config, **base)
cls, _, _ = get_cloud_fs(config, **base)
relpath = parsed.path.lstrip("/").replace("/", cls.sep)
url = cls.sep.join((base["url"], relpath))
return {**base, **remote_conf, "url": url}
def get_cloud_fs(repo, **kwargs):
repo_config = repo.config if repo else {}
return _get_cloud_fs(repo_config, **kwargs)
def _get_cloud_fs(repo_config, **kwargs):
def get_cloud_fs(repo_config, **kwargs):
repo_config = repo_config or {}
core_config = repo_config.get("core", {})
remote_conf = get_fs_config(repo_config, **kwargs)
+3 -3
View File
@@ -288,7 +288,7 @@ class _DVCFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
dvc_infos = {}
if dvc_fs:
dvc_path = _get_dvc_path(dvc_fs, subkey)
with suppress(FileNotFoundError):
with suppress(FileNotFoundError, NotADirectoryError):
for info in dvc_fs.ls(dvc_path, detail=True):
dvc_infos[dvc_fs.path.name(info["name"])] = info
dvc_exists = bool(dvc_infos) or dvc_fs.exists(dvc_path)
@@ -352,7 +352,7 @@ class _DVCFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
dvc_info = dvc_fs.fs.index.info(subkey)
dvc_path = _get_dvc_path(dvc_fs, subkey)
dvc_info["name"] = dvc_path
except FileNotFoundError:
except KeyError:
pass
fs_info = None
@@ -412,7 +412,7 @@ class DVCFileSystem(FileSystem):
@functools.cached_property
# pylint: disable-next=invalid-overridden-method
def fs(self) -> "DVCFileSystem":
def fs(self) -> "_DVCFileSystem":
return _DVCFileSystem(**self.fs_args)
def isdvc(self, path, **kwargs) -> bool:
+2 -2
View File
@@ -56,7 +56,7 @@ def add_logging_level(level_name, level_num, method_name=None):
class LoggingException(Exception):
def __init__(self, record):
msg = f"failed to log {str(record)}"
msg = f"failed to log {record!s}"
super().__init__(msg)
@@ -160,7 +160,7 @@ class LoggerHandler(logging.StreamHandler):
self.flush()
except (BrokenPipeError, RecursionError):
raise
except Exception: # noqa, pylint: disable=broad-except
except Exception: # noqa: BLE001, pylint: disable=broad-except
self.handleError(record)
+62 -29
View File
@@ -354,7 +354,11 @@ class Output:
if fs_config is not None:
fs_kwargs.update(**fs_config)
fs_cls, fs_config, fs_path = get_cloud_fs(self.repo, url=path, **fs_kwargs)
fs_cls, fs_config, fs_path = get_cloud_fs(
self.repo.config if self.repo else {},
url=path,
**fs_kwargs,
)
self.fs = fs_cls(**fs_config)
if (
@@ -471,9 +475,7 @@ class Output:
return fs.path.abspath(fs.path.normpath(fs_path))
def __repr__(self):
return "{class_name}: '{def_path}'".format(
class_name=type(self).__name__, def_path=self.def_path
)
return f"{type(self).__name__}: {self.def_path!r}"
def __str__(self):
if self.fs.protocol != "local":
@@ -486,6 +488,9 @@ class Output:
):
return str(self.def_path)
if not self.fs.path.isin(self.fs_path, self.repo.root_dir):
return self.fs_path
cur_dir = self.fs.path.getcwd()
if self.fs.path.isin(cur_dir, self.repo.root_dir):
return self.fs.path.relpath(self.fs_path, cur_dir)
@@ -693,7 +698,7 @@ class Output:
def save(self) -> None:
if self.use_cache and not self.is_in_repo:
raise DvcException(
f"Saving cached external output {str(self)} is not supported "
f"Saving cached external output {self!s} is not supported "
"since DVC 3.0. See "
f"{format_link('https://dvc.org/doc/user-guide/upgrade')} "
"for more info."
@@ -782,13 +787,18 @@ class Output:
self.hash_name,
ignore=self.dvcignore,
)
otransfer(
staging,
self.cache,
{obj.hash_info},
shallow=False,
hardlink=hardlink,
)
with Callback.as_tqdm_callback(
desc=f"Committing {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{obj.hash_info},
shallow=False,
hardlink=hardlink,
callback=cb,
)
if relink:
rel = self.fs.path.relpath(filter_info or self.fs_path)
with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
@@ -817,13 +827,18 @@ class Output:
save_obj = obj.filter(prefix)
assert isinstance(save_obj, Tree)
checkout_obj = save_obj.get_obj(self.cache, prefix)
otransfer(
staging,
self.cache,
{save_obj.hash_info} | {oid for _, _, oid in save_obj},
shallow=True,
hardlink=hardlink,
)
with Callback.as_tqdm_callback(
desc=f"Committing {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{save_obj.hash_info} | {oid for _, _, oid in save_obj},
shallow=True,
hardlink=hardlink,
callback=cb,
)
return checkout_obj
def dumpd(self, **kwargs): # noqa: C901, PLR0912
@@ -1019,7 +1034,9 @@ class Output:
if odb is None:
odb = self.cache
cls, config, from_info = get_cloud_fs(self.repo, url=source)
cls, config, from_info = get_cloud_fs(
self.repo.config if self.repo else {}, url=source
)
from_fs = cls(**config)
# When running import-url --to-remote / add --to-remote/-o ... we
@@ -1041,14 +1058,19 @@ class Output:
upload=upload,
no_progress_bar=no_progress_bar,
)
otransfer(
staging,
odb,
{obj.hash_info},
jobs=jobs,
hardlink=False,
shallow=False,
)
with Callback.as_tqdm_callback(
desc=f"Transferring to {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
otransfer(
staging,
odb,
{obj.hash_info},
jobs=jobs,
hardlink=False,
shallow=False,
callback=cb,
)
self.hash_info = obj.hash_info
self.files = None
@@ -1411,7 +1433,18 @@ class Output:
assert staging
assert obj.hash_info
otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
with Callback.as_tqdm_callback(
desc=f"Adding {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{obj.hash_info},
hardlink=relink,
shallow=False,
callback=cb,
)
if relink:
with CheckoutCallback(
+130 -25
View File
@@ -2,6 +2,7 @@ import logging
import os
from collections.abc import Mapping, Sequence
from copy import deepcopy
from itertools import product
from typing import (
TYPE_CHECKING,
Any,
@@ -51,6 +52,7 @@ VARS_KWD = "vars"
WDIR_KWD = "wdir"
PARAMS_KWD = "params"
FOREACH_KWD = "foreach"
MATRIX_KWD = "matrix"
DO_KWD = "do"
DEFAULT_PARAMS_FILE = "params.yaml"
@@ -105,7 +107,7 @@ def is_map_or_seq(data: Any) -> bool:
return not isinstance(data, str) and _is_map_or_seq(data)
def split_foreach_name(name: str) -> Tuple[str, Optional[str]]:
def split_group_name(name: str) -> Tuple[str, Optional[str]]:
group, *keys = name.rsplit(JOIN, maxsplit=1)
return group, first(keys)
@@ -120,13 +122,15 @@ def check_interpolations(data: "DictStrAny", where: str, path: str):
return recurse(func)(data)
Definition = Union["ForeachDefinition", "EntryDefinition"]
Definition = Union["ForeachDefinition", "EntryDefinition", "MatrixDefinition"]
def make_definition(
resolver: "DataResolver", name: str, definition: "DictStrAny", **kwargs
) -> Definition:
args = resolver, resolver.context, name, definition
if MATRIX_KWD in definition:
return MatrixDefinition(*args, **kwargs)
if FOREACH_KWD in definition:
return ForeachDefinition(*args, **kwargs)
return EntryDefinition(*args, **kwargs)
@@ -135,6 +139,7 @@ def make_definition(
class DataResolver:
def __init__(self, repo: "Repo", wdir: str, d: dict):
self.fs = fs = repo.fs
self.parsing_config = repo.config.get("parsing", {})
if os.path.isabs(wdir):
wdir = fs.path.relpath(wdir)
@@ -158,7 +163,8 @@ class DataResolver:
self.tracked_vars: Dict[str, Mapping] = {}
stages_data = d.get(STAGES_KWD, {})
# we wrap the definitions into ForeachDefinition and EntryDefinition,
# we wrap the definitions into:
# ForeachDefinition, MatrixDefinition, and EntryDefinition
# that helps us to optimize, cache and selectively load each one of
# them as we need, and simplify all of this DSL/parsing logic.
self.definitions: Dict[str, Definition] = {
@@ -167,12 +173,13 @@ class DataResolver:
}
def resolve_one(self, name: str):
group, key = split_foreach_name(name)
group, key = split_group_name(name)
if not self._has_group_and_key(group, key):
raise EntryNotFound(f"Could not find '{name}'")
# all of the checks for `key` not being None for `ForeachDefinition`
# all of the checks for `key` not being None for
# `ForeachDefinition`/`MatrixDefinition`
# and/or `group` not existing in the `interim`, etc. should be
# handled by the `self.has_key()` above.
definition = self.definitions[group]
@@ -189,7 +196,7 @@ class DataResolver:
return {STAGES_KWD: data}
def has_key(self, key: str):
return self._has_group_and_key(*split_foreach_name(key))
return self._has_group_and_key(*split_group_name(key))
def _has_group_and_key(self, group: str, key: Optional[str] = None):
try:
@@ -197,16 +204,14 @@ class DataResolver:
except KeyError:
return False
if key:
return isinstance(definition, ForeachDefinition) and definition.has_member(
key
)
return not isinstance(definition, ForeachDefinition)
if not isinstance(definition, (ForeachDefinition, MatrixDefinition)):
return key is None
return key is not None and definition.has_member(key)
@collecting
def get_keys(self):
for name, definition in self.definitions.items():
if isinstance(definition, ForeachDefinition):
if isinstance(definition, (ForeachDefinition, MatrixDefinition)):
yield from definition.get_generated_names()
continue
yield name
@@ -255,8 +260,9 @@ class EntryDefinition:
name = self.name
if not skip_checks:
# we can check for syntax errors as we go for interpolated entries,
# but for foreach-generated ones, once is enough, which it does
# that itself. See `ForeachDefinition.do_definition`.
# but for foreach and matrix generated ones, once is enough, which it does
# that itself. See `ForeachDefinition.template`
# and `MatrixDefinition.template`.
check_syntax_errors(self.definition, name, self.relpath)
# we need to pop vars from generated/evaluated data
@@ -301,7 +307,10 @@ class EntryDefinition:
) -> "DictStrAny":
try:
return context.resolve(
value, skip_interpolation_checks=skip_checks, key=key
value,
skip_interpolation_checks=skip_checks,
key=key,
config=self.resolver.parsing_config,
)
except (ParseError, KeyNotInContext) as exc:
format_and_raise(exc, f"'{self.where}.{self.name}.{key}'", self.relpath)
@@ -327,17 +336,18 @@ class ForeachDefinition:
self.name = name
assert DO_KWD in definition
assert MATRIX_KWD not in definition
self.foreach_data = definition[FOREACH_KWD]
self._do_definition = definition[DO_KWD]
self._template = definition[DO_KWD]
self.pair = IterationPair()
self.where = where
@cached_property
def do_definition(self):
def template(self):
# optimization: check for syntax errors only once for `foreach` stages
check_syntax_errors(self._do_definition, self.name, self.relpath)
return self._do_definition
check_syntax_errors(self._template, self.name, self.relpath)
return self._template
@cached_property
def resolved_iterable(self):
@@ -440,16 +450,111 @@ class ForeachDefinition:
# i.e. quadratic complexity).
generated = self._generate_name(key)
entry = EntryDefinition(
self.resolver, self.context, generated, self.do_definition
self.resolver, self.context, generated, self.template
)
try:
# optimization: skip checking for syntax errors on each foreach
# generated stages. We do it once when accessing do_definition.
# generated stages. We do it once when accessing template.
return entry.resolve_stage(skip_checks=True)
except ContextError as exc:
format_and_raise(exc, f"stage '{generated}'", self.relpath)
# let mypy know that this state is unreachable as format_and_raise
# does not return at all (it's not able to understand it for some
# reason)
raise AssertionError("unreachable")
class MatrixDefinition:
def __init__(
self,
resolver: DataResolver,
context: Context,
name: str,
definition: "DictStrAny",
where: str = STAGES_KWD,
):
self.resolver = resolver
self.relpath = self.resolver.relpath
self.context = context
self.name = name
assert MATRIX_KWD in definition
assert DO_KWD not in definition
assert FOREACH_KWD not in definition
self._template = definition.copy()
self.matrix_data = self._template.pop(MATRIX_KWD)
self.pair = IterationPair()
self.where = where
@cached_property
def template(self) -> "DictStrAny":
# optimization: check for syntax errors only once for `matrix` stages
check_syntax_errors(self._template, self.name, self.relpath)
return self._template
@cached_property
def resolved_iterable(self) -> Dict[str, List]:
return self._resolve_matrix_data()
def _resolve_matrix_data(self) -> Dict[str, List]:
try:
iterable = self.context.resolve(self.matrix_data, unwrap=False)
except (ContextError, ParseError) as exc:
format_and_raise(exc, f"'{self.where}.{self.name}.matrix'", self.relpath)
return iterable
@cached_property
def normalized_iterable(self) -> Dict[str, "DictStrAny"]:
"""Convert sequence to Mapping with keys normalized."""
iterable = self.resolved_iterable
assert isinstance(iterable, Mapping)
ret: Dict[str, "DictStrAny"] = {}
matrix = {key: enumerate(v) for key, v in iterable.items()}
for combination in product(*matrix.values()):
d: "DictStrAny" = {}
fragments: List[str] = []
for k, (i, v) in zip(matrix.keys(), combination):
d[k] = v
fragments.append(f"{k}{i}" if is_map_or_seq(v) else to_str(v))
key = "-".join(fragments)
ret[key] = d
return ret
def has_member(self, key: str) -> bool:
return key in self.normalized_iterable
def get_generated_names(self) -> List[str]:
return list(map(self._generate_name, self.normalized_iterable))
def _generate_name(self, key: str) -> str:
return f"{self.name}{JOIN}{key}"
def resolve_all(self) -> "DictStrAny":
return join(map(self.resolve_one, self.normalized_iterable))
def resolve_one(self, key: str) -> "DictStrAny":
return self._each_iter(key)
def _each_iter(self, key: str) -> "DictStrAny":
err_message = f"Could not find '{key}' in matrix group '{self.name}'"
with reraise(KeyError, EntryNotFound(err_message)):
value = self.normalized_iterable[key]
temp_dict = {self.pair.value: value}
with self.context.set_temporarily(temp_dict, reserve=True):
# optimization: item and key can be removed on __exit__() as they
# are top-level values, and are not merged recursively.
# This helps us avoid cloning context, which is slower
# (increasing the size of the context might increase
# the no. of items to be generated which means more cloning,
# i.e. quadratic complexity).
generated = self._generate_name(key)
entry = EntryDefinition(
self.resolver, self.context, generated, self.template
)
try:
# optimization: skip checking for syntax errors on each matrix
# generated stages. We do it once when accessing template.
return entry.resolve_stage(skip_checks=True)
except ContextError as exc:
format_and_raise(exc, f"stage '{generated}'", self.relpath)
+19 -4
View File
@@ -495,7 +495,12 @@ class Context(CtxDict):
self.data.pop(key, None)
def resolve(
self, src, unwrap=True, skip_interpolation_checks=False, key=None
self,
src,
unwrap=True,
skip_interpolation_checks=False,
key=None,
config=None,
) -> Any:
"""Recursively resolves interpolation and returns resolved data.
@@ -511,10 +516,15 @@ class Context(CtxDict):
{'lst': [1, 2, 3]}
"""
func = recurse(self.resolve_str)
return func(src, unwrap, skip_interpolation_checks, key)
return func(src, unwrap, skip_interpolation_checks, key, config)
def resolve_str(
self, src: str, unwrap=True, skip_interpolation_checks=False, key=None
self,
src: str,
unwrap=True,
skip_interpolation_checks=False,
key=None,
config=None,
) -> str:
"""Resolves interpolated string to it's original value,
or in case of multiple interpolations, a combined string.
@@ -535,7 +545,12 @@ class Context(CtxDict):
return value
# but not "${num} days"
return str_interpolate(
src, matches, self, skip_checks=skip_interpolation_checks, key=key
src,
matches,
self,
skip_checks=skip_interpolation_checks,
key=key,
config=config,
)
+6 -7
View File
@@ -82,20 +82,18 @@ def escape_str(value):
@singledispatch
def to_str(obj) -> str:
def to_str(obj, config=None) -> str: # noqa: ARG001, pylint: disable=unused-argument
return str(obj)
@to_str.register(bool)
def _(obj: bool):
def _(obj: bool, config=None): # noqa: ARG001, pylint: disable=unused-argument
return "true" if obj else "false"
@to_str.register(dict)
def _(obj: dict): # noqa: C901
from dvc.config import Config
config = Config.from_cwd().get("parsing", {})
def _(obj: dict, config=None): # noqa: C901
config = config or {}
result = ""
for k, v in flatten(obj).items():
@@ -211,6 +209,7 @@ def str_interpolate(
context: "Context",
skip_checks: bool = False,
key=None,
config=None,
):
index, buf = 0, ""
for match in matches:
@@ -218,7 +217,7 @@ def str_interpolate(
expr = get_expression(match, skip_checks=skip_checks)
value = context.select(expr, unwrap=True)
validate_value(value, key)
buf += template[index:start] + to_str(value)
buf += template[index:start] + to_str(value, config=config)
index = end
buf += template[index:]
# regex already backtracks and avoids any `${` starting with
+3
View File
@@ -1,3 +1,4 @@
import logging
import os
from collections import defaultdict
from typing import TYPE_CHECKING, DefaultDict, Dict, List, NamedTuple, Optional
@@ -17,6 +18,7 @@ if TYPE_CHECKING:
dpath.options.ALLOW_EMPTY_STRING_KEYS = True
logger = logging.getLogger(__name__)
def _squash_plots_properties(data: List) -> Dict:
@@ -113,6 +115,7 @@ def match_defs_renderers( # noqa: C901, PLR0912
try:
dps, rev_props = converter.flat_datapoints(rev)
except Exception as e: # noqa: BLE001, pylint: disable=broad-except
logger.warning("In %r, %s", rev, str(e).lower())
def_errors[rev] = e
continue
+44 -6
View File
@@ -29,7 +29,7 @@ if TYPE_CHECKING:
from dvc.stage import Stage
from dvc.types import DictStrAny
from dvc_data.hashfile.state import StateBase
from dvc_data.index import DataIndex
from dvc_data.index import DataIndex, DataIndexEntry
from .experiments import Experiments
from .index import Index
@@ -133,7 +133,7 @@ class Repo:
assert root_dir
return root_dir, dvc_dir
def __init__( # noqa: PLR0915
def __init__( # noqa: PLR0915, PLR0913
self,
root_dir: Optional[str] = None,
fs: Optional["FileSystem"] = None,
@@ -144,9 +144,10 @@ class Repo:
url: Optional[str] = None,
repo_factory: Optional[Callable] = None,
scm: Optional[Union["Git", "NoSCM"]] = None,
remote: Optional[str] = None,
remote_config: Optional["DictStrAny"] = None,
):
from dvc.cachemgr import CacheManager
from dvc.config import Config
from dvc.data_cloud import DataCloud
from dvc.fs import GitFileSystem, LocalFileSystem, localfs
from dvc.lock import LockNoop, make_lock
@@ -163,6 +164,9 @@ class Repo:
self._fs_conf = {"repo_factory": repo_factory}
self._fs = fs or localfs
self._scm = scm
self._config = config
self._remote = remote
self._remote_config = remote_config
self._data_index = None
if rev and not fs:
@@ -182,7 +186,6 @@ class Repo:
scm=scm,
)
self.config: Config = Config(self.dvc_dir, fs=self.fs, config=config)
self._uninitialized = uninitialized
# used by DVCFileSystem to determine if it should traverse subrepos
@@ -234,6 +237,18 @@ class Repo:
def __str__(self):
return self.url or self.root_dir
@cached_property
def config(self):
from dvc.config import Config
return Config(
self.dvc_dir,
fs=self.fs,
config=self._config,
remote=self._remote,
remote_config=self._remote_config,
)
@cached_property
def local_dvc_dir(self):
from dvc.fs import GitFileSystem, LocalFileSystem
@@ -285,7 +300,7 @@ class Repo:
new.check_graph()
@staticmethod
def open(url, *args, **kwargs): # noqa: A003
def open(url: Optional[str], *args, **kwargs) -> "Repo": # noqa: A003
from .open_repo import open_repo
return open_repo(url, *args, **kwargs)
@@ -375,6 +390,28 @@ class Repo:
self.data_index.commit()
self._reset()
def get_data_index_entry(
self,
path: str,
workspace: str = "repo",
) -> Tuple["DataIndex", "DataIndexEntry"]:
if self.subrepos:
fs_path = self.dvcfs.from_os_path(path)
fs = self.dvcfs.fs
# pylint: disable-next=protected-access
key = fs._get_key_from_relative(fs_path)
# pylint: disable-next=protected-access
subrepo, _, key = fs._get_subrepo_info(key)
index = subrepo.index.data[workspace]
else:
index = self.index.data[workspace]
key = self.fs.path.relparts(path, self.root_dir)
try:
return index, index[key]
except KeyError as exc:
raise OutputNotFoundError(path, self) from exc
def __repr__(self):
return f"{self.__class__.__name__}: '{self.root_dir}'"
@@ -384,7 +421,7 @@ class Repo:
fs = fs or localfs
root = root or os.curdir
root_dir = fs.path.realpath(root)
root_dir = fs.path.abspath(root)
if not fs.isdir(root_dir):
raise NotDvcRepoError(f"directory '{root}' does not exist")
@@ -607,6 +644,7 @@ class Repo:
self.__dict__.pop("dvcignore", None)
self.__dict__.pop("dvcfs", None)
self.__dict__.pop("datafs", None)
self.__dict__.pop("config", None)
def __enter__(self):
return self
+8
View File
@@ -67,6 +67,7 @@ def brancher( # noqa: E302
saved_fs = self.fs
saved_root = self.root_dir
saved_dvc_dir = self.dvc_dir
scm = self.scm
@@ -101,6 +102,8 @@ def brancher( # noqa: E302
finally:
self.fs = saved_fs
self.root_dir = saved_root
self.dvc_dir = saved_dvc_dir
self._reset() # pylint: disable=protected-access
def _switch_fs(
@@ -125,6 +128,8 @@ def _switch_fs(
repo.fs = fs
repo.root_dir = root_dir
repo.dvc_dir = fs.path.join(root_dir, repo.DVC_DIR)
repo._reset() # pylint: disable=protected-access
if cwd_parts:
cwd = repo.fs.path.join("/", *cwd_parts)
@@ -149,9 +154,12 @@ def switch(repo: "Repo", rev: str) -> Iterator[str]:
saved_fs = repo.fs
saved_root = repo.root_dir
saved_dvc_dir = repo.dvc_dir
try:
_switch_fs(repo, rev, repo_root_parts, cwd_parts)
yield rev
finally:
repo.fs = saved_fs
repo.root_dir = saved_root
repo.dvc_dir = saved_dvc_dir
repo._reset() # pylint: disable=protected-access
+18 -1
View File
@@ -1,6 +1,7 @@
import itertools
import logging
import os
from dataclasses import fields
from datetime import datetime
from typing import (
TYPE_CHECKING,
@@ -152,7 +153,23 @@ def collect_queued(
"""
if not baseline_revs:
return {}
return repo.experiments.celery_queue.collect_queued_data(baseline_revs, **kwargs)
queued_data = {}
for rev, ranges in repo.experiments.celery_queue.collect_queued_data(
baseline_revs, **kwargs
).items():
for exp_range in ranges:
for exp_state in exp_range.revs:
if exp_state.data:
attrs = [f.name for f in fields(SerializableExp)]
exp_state.data = SerializableExp(
**{
attr: getattr(exp_state.data, attr)
for attr in attrs
if attr != "metrics"
}
)
queued_data[rev] = ranges
return queued_data
def collect_active(
+1 -1
View File
@@ -607,7 +607,7 @@ class BaseExecutor(ABC):
old_cwd = os.getcwd()
for path in copy_paths or []:
cls._copy_path(os.path.realpath(path), os.path.join(dvc.root_dir, path))
cls._copy_path(os.path.abspath(path), os.path.join(dvc.root_dir, path))
if info.wdir:
os.chdir(os.path.join(dvc.scm.root_dir, info.wdir))
+3 -1
View File
@@ -115,7 +115,9 @@ class TempDirQueue(WorkspaceQueue):
message=message,
)
if not exec_result.exp_hash:
raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")
raise DvcException( # noqa: TRY301
f"Failed to reproduce experiment '{rev[:7]}'"
)
if exec_result.ref_info:
results[rev].update(
self.collect_executor(self.repo.experiments, executor, exec_result)
+3 -1
View File
@@ -121,7 +121,9 @@ class WorkspaceQueue(BaseStashQueue):
message=kwargs.get("message"),
)
if not exec_result.exp_hash:
raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")
raise DvcException( # noqa: TRY301
f"Failed to reproduce experiment '{rev[:7]}'"
)
if exec_result.ref_info:
results[rev].update(
self.collect_executor(self.repo.experiments, executor, exec_result)
+1 -1
View File
@@ -234,7 +234,7 @@ def fix_exp_head(scm: Union["Git", "NoSCM"], ref: Optional[str]) -> Optional[str
if ref:
name, tail = Git.split_ref_pattern(ref)
if name == "HEAD" and scm.get_ref(EXEC_BASELINE):
return "".join((EXEC_BASELINE, tail))
return f"{EXEC_BASELINE}{tail}"
return ref
+19 -7
View File
@@ -21,19 +21,25 @@ def _collect_indexes( # noqa: PLR0913
revs=None,
max_size=None,
types=None,
config=None,
):
indexes = {}
collection_exc = None
config = config or {}
if remote:
core = config.get("core") or {}
core["remote"] = remote
config["core"] = core
for rev in repo.brancher(
revs=revs,
all_branches=all_branches,
all_tags=all_tags,
all_commits=all_commits,
):
saved_remote = repo.config["core"].get("remote")
try:
if remote:
repo.config["core"]["remote"] = remote
repo.config.update(config)
idx = repo.index.targets_view(
targets,
@@ -42,13 +48,17 @@ def _collect_indexes( # noqa: PLR0913
max_size=max_size,
types=types,
)
indexes[idx.data_tree.hash_info.value] = idx.data["repo"]
def onerror(_entry, _exc):
pass
data = idx.data["repo"]
data.onerror = onerror
indexes[idx.data_tree.hash_info.value] = data
except Exception as exc: # pylint: disable=broad-except
collection_exc = exc
logger.exception("failed to collect '%s'", rev or "workspace")
finally:
if remote:
repo.config["core"]["remote"] = saved_remote
if not indexes and collection_exc:
raise collection_exc
@@ -71,6 +81,7 @@ def fetch( # noqa: C901, PLR0913
revs=None,
max_size=None,
types=None,
config=None,
) -> int:
"""Download data items from a cloud and imported repositories
@@ -114,6 +125,7 @@ def fetch( # noqa: C901, PLR0913
revs=revs,
max_size=max_size,
types=types,
config=config,
)
cache_key = ("fetch", tokenize(sorted(indexes.keys())))
+26 -14
View File
@@ -19,22 +19,39 @@ class GetDVCFileError(DvcException):
)
def get(url, path, out=None, rev=None, jobs=None, force=False):
def get(
url,
path,
out=None,
rev=None,
jobs=None,
force=False,
config=None,
remote=None,
remote_config=None,
):
from dvc.config import Config
from dvc.dvcfile import is_valid_filename
from dvc.fs.callbacks import Callback
from dvc.repo import Repo
out = resolve_output(path, out, force=force)
if is_valid_filename(out):
raise GetDVCFileError()
raise GetDVCFileError
if config and not isinstance(config, dict):
config = Config.load_file(config)
with Repo.open(
url=url,
rev=rev,
subrepos=True,
uninitialized=True,
config=config,
remote=remote,
remote_config=remote_config,
) as repo:
from dvc.fs import download
from dvc.fs.data import DataFileSystem
fs: Union[DataFileSystem, "DVCFileSystem"]
@@ -44,14 +61,9 @@ def get(url, path, out=None, rev=None, jobs=None, force=False):
else:
fs = repo.dvcfs
fs_path = fs.from_os_path(path)
with Callback.as_tqdm_callback(
desc=f"Downloading {fs.path.name(path)}",
unit="files",
) as cb:
fs.get(
fs_path,
os.path.abspath(out),
batch_size=jobs,
callback=cb,
)
download(
fs,
fs_path,
os.path.abspath(out),
jobs=jobs,
)
+1 -1
View File
@@ -14,4 +14,4 @@ def get_url(url, out=None, *, config=None, jobs=None, force=False):
src_fs, src_path = parse_external_url(url, config)
if not src_fs.exists(src_path):
raise URLMissingError(url)
download(src_fs, src_path, out, jobs=jobs)
download(src_fs, src_path, out.fs_path, jobs=jobs)
+6
View File
@@ -31,6 +31,9 @@ def check_acyclic(graph: "DiGraph") -> None:
def get_pipeline(pipelines, node):
found = [i for i in pipelines if i.has_node(node)]
if not found:
return None
assert len(found) == 1
return found[0]
@@ -60,6 +63,9 @@ def collect_pipeline(stage: "Stage", graph: "DiGraph") -> Iterator["Stage"]:
import networkx as nx
pipeline = get_pipeline(get_pipelines(graph), stage)
if not pipeline:
return iter([])
return nx.dfs_postorder_nodes(pipeline, stage)
+27 -1
View File
@@ -1,6 +1,32 @@
def imp(self, url, path, out=None, rev=None, **kwargs):
def imp(
self,
url,
path,
out=None,
rev=None,
config=None,
remote=None,
remote_config=None,
**kwargs,
):
erepo = {"url": url}
if rev is not None:
erepo["rev"] = rev
if config is not None:
erepo["config"] = config
if remote is not None and remote_config is not None:
conf = erepo.get("config") or {}
remotes = conf.get("remote") or {}
remote_conf = remotes.get(remote) or {}
remote_conf.update(remote_config)
remotes[remote] = remote_conf
conf["remote"] = remotes
erepo["config"] = conf
elif remote is not None:
erepo["remote"] = remote
elif remote_config is not None:
erepo["remote"] = remote_config
return self.imp_url(path, out=out, erepo=erepo, frozen=True, **kwargs)
+1 -1
View File
@@ -37,7 +37,7 @@ def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False): # noqa:
"Cannot initialize repo with `--no-scm` and `--subdir`"
)
root_dir = os.path.realpath(root_dir)
root_dir = os.path.abspath(root_dir)
dvc_dir = os.path.join(root_dir, Repo.DVC_DIR)
try:
+22 -1
View File
@@ -13,6 +13,9 @@ def ls(
rev: Optional[str] = None,
recursive: Optional[bool] = None,
dvc_only: bool = False,
config: Optional[str] = None,
remote: Optional[str] = None,
remote_config: Optional[dict] = None,
):
"""Methods for getting files and outputs for the repo.
@@ -22,6 +25,9 @@ def ls(
rev (str, optional): SHA commit, branch or tag name
recursive (bool, optional): recursively walk the repo
dvc_only (bool, optional): show only DVC-artifacts
config (str, optional): path to config file
remote (str, optional): remote name to set as a default remote in the repo
remote_config (str, dict): remote config to merge with a remote in the repo
Returns:
list of `entry`
@@ -35,9 +41,24 @@ def ls(
"isexec": bool,
}
"""
from dvc.config import Config
from . import Repo
with Repo.open(url, rev=rev, subrepos=True, uninitialized=True) as repo:
if config and not isinstance(config, dict):
config_dict = Config.load_file(config)
else:
config_dict = None
with Repo.open(
url,
rev=rev,
subrepos=True,
uninitialized=True,
config=config_dict,
remote=remote,
remote_config=remote_config,
) as repo:
path = path or ""
ret = _ls(repo, path, recursive, dvc_only)
+11 -42
View File
@@ -2,8 +2,7 @@ import logging
import os
import tempfile
import threading
from contextlib import contextmanager
from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
from typing import TYPE_CHECKING, Dict, Optional, Tuple
from funcy import retry, wrap_with
@@ -18,16 +17,14 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
@contextmanager
@map_scm_exception()
def _external_repo(
url,
rev: Optional[str] = None,
for_write: bool = False,
**kwargs,
) -> Iterator["Repo"]:
) -> "Repo":
logger.debug("Creating external repo %s@%s", url, rev)
path = _cached_clone(url, rev, for_write=for_write)
path = _cached_clone(url, rev)
# Local HEAD points to the tip of whatever branch we first cloned from
# (which may not be the default branch), use origin/HEAD here to get
# the tip of the default branch
@@ -38,11 +35,6 @@ def _external_repo(
config.update(kwargs.pop("config", None) or {})
main_root = "/"
if for_write:
# we already checked out needed revision
rev = None
main_root = path
repo_kwargs = dict(
root_dir=path,
url=url,
@@ -52,14 +44,7 @@ def _external_repo(
**kwargs,
)
repo = Repo(**repo_kwargs)
try:
yield repo
finally:
repo.close()
if for_write:
_remove(path)
return Repo(**repo_kwargs)
def open_repo(url, *args, **kwargs):
@@ -140,20 +125,19 @@ def _get_remote_config(url):
repo.close()
def _cached_clone(url, rev, for_write=False):
def _cached_clone(url, rev):
"""Clone an external git repo to a temporary directory.
Returns the path to a local temporary directory with the specified
revision checked out. If for_write is set prevents reusing this dir via
cache.
revision checked out.
"""
from shutil import copytree
# even if we have already cloned this repo, we may need to
# fetch/fast-forward to get specified rev
clone_path, shallow = _clone_default_branch(url, rev, for_write=for_write)
clone_path, shallow = _clone_default_branch(url, rev)
if not for_write and (url) in CLONES:
if url in CLONES:
return CLONES[url][0]
# Copy to a new dir to keep the clone clean
@@ -161,16 +145,12 @@ def _cached_clone(url, rev, for_write=False):
logger.debug("erepo: making a copy of %s clone", url)
copytree(clone_path, repo_path)
# Check out the specified revision
if for_write:
_git_checkout(repo_path, rev)
else:
CLONES[url] = (repo_path, shallow)
CLONES[url] = (repo_path, shallow)
return repo_path
@wrap_with(threading.Lock())
def _clone_default_branch(url, rev, for_write=False): # noqa: C901, PLR0912
def _clone_default_branch(url, rev): # noqa: C901, PLR0912
"""Get or create a clean clone of the url.
The cloned is reactualized with git pull unless rev is a known sha.
@@ -204,7 +184,7 @@ def _clone_default_branch(url, rev, for_write=False): # noqa: C901, PLR0912
logger.debug("erepo: git clone '%s' to a temporary dir", url)
clone_path = tempfile.mkdtemp("dvc-clone")
if not for_write and rev and not Git.is_sha(rev):
if rev and not Git.is_sha(rev):
# If rev is a tag or branch name try shallow clone first
try:
@@ -249,17 +229,6 @@ def _merge_upstream(git: "Git"):
pass
def _git_checkout(repo_path, rev):
from dvc.scm import Git
logger.debug("erepo: git checkout %s@%s", repo_path, rev)
git = Git(repo_path)
try:
git.checkout(rev)
finally:
git.close()
def _remove(path):
from dvc.utils.fs import remove
+2 -2
View File
@@ -134,7 +134,7 @@ class Plots:
if "workspace" in revs:
# reorder revs to match repo.brancher ordering
revs.remove("workspace")
revs = ["workspace"] + revs
revs = ["workspace", *revs]
for rev in revs:
with switch_repo(self.repo, rev) as (repo, _):
res: Dict = {}
@@ -483,7 +483,7 @@ def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
if isinstance(elem, str):
dvcfile_defs_dict[elem] = None
else:
k, v = list(elem.items())[0]
k, v = next(iter(elem.items()))
dvcfile_defs_dict[k] = v
resolved = _resolve_definitions(
+101 -40
View File
@@ -1,11 +1,25 @@
import logging
from typing import TYPE_CHECKING, Iterable, List, Optional, Union, cast
from typing import (
TYPE_CHECKING,
Callable,
Dict,
Iterable,
List,
NoReturn,
Optional,
Set,
Tuple,
TypeVar,
Union,
cast,
)
from funcy import ldistinct
from dvc.exceptions import ReproductionError
from dvc.repo.scm_context import scm_context
from dvc.stage.cache import RunCacheNotSupported
from dvc.utils import humanize
from dvc.utils.collections import ensure_list
from . import locked
@@ -18,6 +32,7 @@ if TYPE_CHECKING:
from . import Repo
logger = logging.getLogger(__name__)
T = TypeVar("T")
def collect_stages(
@@ -50,7 +65,7 @@ def get_subgraph(
return nx.compose_all(used_pipelines)
def _remove_frozen_stages(graph: "DiGraph") -> "DiGraph":
def get_active_graph(graph: "DiGraph") -> "DiGraph":
g = cast("DiGraph", graph.copy())
for stage in graph:
if stage.frozen:
@@ -59,23 +74,12 @@ def _remove_frozen_stages(graph: "DiGraph") -> "DiGraph":
return g
def get_active_graph(
graph: "DiGraph",
stages: Optional[List["Stage"]] = None,
pipeline: bool = False,
downstream: bool = False,
) -> "DiGraph":
"""Return the graph to operate."""
processed = _remove_frozen_stages(graph)
return get_subgraph(processed, stages, pipeline=pipeline, downstream=downstream)
def plan_repro(
graph: "DiGraph",
stages: Optional[List["Stage"]] = None,
stages: Optional[List["T"]] = None,
pipeline: bool = False,
downstream: bool = False,
) -> List["Stage"]:
) -> List["T"]:
r"""Derive the evaluation of the given node for the given graph.
When you _reproduce a stage_, you want to _evaluate the descendants_
@@ -113,16 +117,14 @@ def plan_repro(
"""
import networkx as nx
active = get_active_graph(graph, stages, pipeline=pipeline, downstream=downstream)
return list(nx.dfs_postorder_nodes(active))
sub = get_subgraph(graph, stages, pipeline=pipeline, downstream=downstream)
return list(nx.dfs_postorder_nodes(sub))
def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
if stage.frozen and not stage.is_import:
logger.warning(
"%s is frozen. Its dependencies are not going to be reproduced.",
stage,
)
msg = "%s is frozen. Its dependencies are not going to be reproduced."
logger.warning(msg, stage)
ret = stage.reproduce(**kwargs)
if ret and not kwargs.get("dry", False):
@@ -130,31 +132,88 @@ def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
return ret
def _reproduce_stages(
def _get_upstream_downstream_nodes(
graph: Optional["DiGraph"], node: T
) -> Tuple[List[T], List[T]]:
succ = list(graph.successors(node)) if graph else []
pre = list(graph.predecessors(node)) if graph else []
return succ, pre
def _repr(stages: Iterable["Stage"]) -> str:
return humanize.join(repr(stage.addressing) for stage in stages)
def handle_error(
graph: Optional["DiGraph"], on_error: str, exc: Exception, stage: "Stage"
) -> Set["Stage"]:
import networkx as nx
logger.warning("%s%s", exc, " (ignored)" if on_error == "ignore" else "")
if not graph or on_error == "ignore":
return set()
dependents = set(nx.dfs_postorder_nodes(graph.reverse(), stage)) - {stage}
if dependents:
names = _repr(dependents)
msg = "%s %s will be skipped due to this failure"
logger.warning(msg, "Stages" if len(dependents) > 1 else "Stage", names)
return dependents
def _raise_error(exc: Optional[Exception], *stages: "Stage") -> NoReturn:
names = _repr(stages)
segment = " stages:" if len(stages) > 1 else ""
raise ReproductionError(f"failed to reproduce{segment} {names}") from exc
def _reproduce(
stages: List["Stage"],
graph: Optional["DiGraph"] = None,
force_downstream: bool = False,
on_error: str = "fail",
force: bool = False,
repro_fn: Callable = _reproduce_stage,
**kwargs,
) -> List["Stage"]:
result: List["Stage"] = []
for i, stage in enumerate(stages):
try:
ret = _reproduce_stage(stage, upstream=stages[:i], **kwargs)
except Exception as exc: # noqa: BLE001
raise ReproductionError(stage.addressing) from exc
assert on_error in ("fail", "keep-going", "ignore")
if not ret:
result: List["Stage"] = []
failed: List["Stage"] = []
to_skip: Dict["Stage", "Stage"] = {}
ret: Optional["Stage"] = None
force_state = {node: force for node in stages}
for stage in stages:
if stage in to_skip:
continue
result.append(ret)
if force_downstream:
# NOTE: we are walking our pipeline from the top to the
# bottom. If one stage is changed, it will be reproduced,
# which tells us that we should force reproducing all of
# the other stages down below, even if their direct
# dependencies didn't change.
kwargs["force"] = True
if i < len(stages) - 1:
if ret:
logger.info("") # add a newline
upstream, downstream = _get_upstream_downstream_nodes(graph, stage)
force_stage = force_state[stage]
try:
ret = repro_fn(stage, upstream=upstream, force=force_stage, **kwargs)
except Exception as exc: # noqa: BLE001, pylint: disable=broad-exception-caught
failed.append(stage)
if on_error == "fail":
_raise_error(exc, stage)
dependents = handle_error(graph, on_error, exc, stage)
to_skip.update({node: stage for node in dependents})
continue
if force_downstream and (ret or force_stage):
force_state.update({node: True for node in downstream})
if ret:
result.append(ret)
if on_error != "ignore" and failed:
_raise_error(None, *failed)
return result
@@ -169,6 +228,7 @@ def reproduce(
downstream: bool = False,
single_item: bool = False,
glob: bool = False,
on_error: Optional[str] = "fail",
**kwargs,
):
from dvc.dvcfile import PROJECT_FILE
@@ -192,8 +252,9 @@ def reproduce(
except RunCacheNotSupported as e:
logger.warning("Failed to pull run cache: %s", e)
graph = None
steps = stages
if not single_item:
graph = self.index.graph
graph = get_active_graph(self.index.graph)
steps = plan_repro(graph, stages, pipeline=pipeline, downstream=downstream)
return _reproduce_stages(steps, **kwargs)
return _reproduce(steps, graph=graph, on_error=on_error or "fail", **kwargs)
+11 -3
View File
@@ -37,11 +37,19 @@ StageSet = Set["Stage"]
def _collect_with_deps(stages: StageList, graph: "DiGraph") -> StageSet:
from dvc.exceptions import StageNotFoundError
from dvc.repo.graph import collect_pipeline
res: StageSet = set()
for stage in stages:
res.update(collect_pipeline(stage, graph=graph))
pl = list(collect_pipeline(stage, graph=graph))
if not pl:
raise StageNotFoundError(
f"Stage {stage} is not found in the project. "
"Check that there are no symlinks in the parents "
"leading up to it within the project."
)
res.update(pl)
return res
@@ -211,7 +219,7 @@ class StageLoad:
self, path: Optional[str] = None, name: Optional[str] = None
) -> str:
if path:
return self.repo.fs.path.realpath(path)
return self.repo.fs.path.abspath(path)
path = PROJECT_FILE
logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
@@ -235,7 +243,7 @@ class StageLoad:
) -> Iterable[str]:
if not name:
return stages.keys()
if accept_group and stages.is_foreach_generated(name):
if accept_group and stages.is_foreach_or_matrix_generated(name):
return self._get_group_keys(stages, name)
if glob:
return fnmatch.filter(stages.keys(), name)
+1 -1
View File
@@ -427,7 +427,7 @@ def _get_update_diff_index(
meta_cmp_key=partial(_meta_checksum, remote.fs),
with_unchanged=True,
):
if change.typ == ADD or change.typ == MODIFY:
if change.typ in (ADD, MODIFY):
entry = change.new
# preserve md5's which were calculated in out.save() after
# downloading
+2 -4
View File
@@ -31,15 +31,13 @@ RWLOCK_LOCK = "rwlock.lock"
class RWLockFileCorruptedError(DvcException):
def __init__(self, path):
super().__init__(
"Unable to read RWLock-file '{}'. JSON structure is corrupted".format(
relpath(path)
)
f"Unable to read RWLock-file {relpath(path)!r}. JSON structure is corrupted"
)
class RWLockFileFormatError(DvcException):
def __init__(self, path):
super().__init__(f"RWLock-file '{relpath(path)}' format error.")
super().__init__(f"RWLock-file {relpath(path)!r} format error.")
@contextmanager
+2 -1
View File
@@ -11,7 +11,7 @@ from dvc.output import (
META_SCHEMA,
Output,
)
from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
from dvc.parsing import DO_KWD, FOREACH_KWD, MATRIX_KWD, VARS_KWD
from dvc.stage.params import StageParams
STAGES = "stages"
@@ -77,6 +77,7 @@ PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]}
VARS_SCHEMA = [str, dict]
STAGE_DEFINITION = {
MATRIX_KWD: {str: Any(str, list)},
Required(StageParams.PARAM_CMD): Any(str, list),
Optional(StageParams.PARAM_WDIR): str,
Optional(StageParams.PARAM_DEPS): [str],
+1 -1
View File
@@ -67,7 +67,7 @@ def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
into = SCMError(str(exc))
if with_cause:
raise into from exc
raise into
raise into # noqa: B904
@overload
+1 -4
View File
@@ -426,10 +426,7 @@ class Stage(params.StageParams):
logger.info("Stage '%s' didn't change, skipping", self.addressing)
return None
msg = "Going to reproduce {stage}. Are you sure you want to continue?".format(
stage=self
)
msg = f"Going to reproduce {self}. Are you sure you want to continue?"
if interactive and not prompt.confirm(msg):
raise DvcException("reproduction aborted by the user")
+5 -2
View File
@@ -123,12 +123,15 @@ class StageCache:
@contextmanager
def _cache_type_copy(self):
cache_types = self.repo.cache.legacy.cache_types
cache_types = self.repo.cache.local.cache_types
legacy_cache_types = self.repo.cache.legacy.cache_types
self.repo.cache.local.cache_types = ["copy"]
self.repo.cache.legacy.cache_types = ["copy"]
try:
yield
finally:
self.repo.cache.legacy.cache_types = cache_types
self.repo.cache.local.cache_types = cache_types
self.repo.cache.legacy.cache_types = legacy_cache_types
def _uncached_outs(self, stage, cache):
# NOTE: using temporary stage to avoid accidentally modifying original
+6 -3
View File
@@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
from funcy import get_in, lcat, once, project
from dvc import dependency, output
from dvc.parsing import FOREACH_KWD, JOIN, EntryNotFound
from dvc.parsing import FOREACH_KWD, JOIN, MATRIX_KWD, EntryNotFound
from dvc.utils.objects import cached_property
from dvc_data.hashfile.meta import Meta
@@ -167,8 +167,11 @@ class StageLoader(Mapping):
def __contains__(self, name):
return self.resolver.has_key(name) # noqa: W601
def is_foreach_generated(self, name: str):
return name in self.stages_data and FOREACH_KWD in self.stages_data[name]
def is_foreach_or_matrix_generated(self, name: str) -> bool:
return (
name in self.stages_data
and {FOREACH_KWD, MATRIX_KWD} & self.stages_data[name].keys()
)
class SingleStageLoader(Mapping):
+1 -1
View File
@@ -80,7 +80,7 @@ def _run(executable, cmd, **kwargs):
exec_cmd = _make_cmd(executable, cmd)
try:
p = subprocess.Popen(exec_cmd, **kwargs) # nosec B603
p = subprocess.Popen(exec_cmd, **kwargs) # nosec B603 # noqa: S603
if main_thread:
old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
+2 -2
View File
@@ -32,14 +32,14 @@ def check_stage_path(repo, path, is_wdir=False):
wdir_or_path="stage working dir" if is_wdir else "file path", path=path
)
real_path = os.path.realpath(path)
real_path = os.path.abspath(path)
if not os.path.exists(real_path):
raise StagePathNotFoundError(error_msg.format("does not exist"))
if not os.path.isdir(real_path):
raise StagePathNotDirectoryError(error_msg.format("is not directory"))
proj_dir = os.path.realpath(repo.root_dir)
proj_dir = os.path.abspath(repo.root_dir)
if real_path != proj_dir and not path_isin(real_path, proj_dir):
raise StagePathOutsideError(error_msg.format("is outside of DVC repo"))
+1 -1
View File
@@ -1 +1 @@
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403 # pylint: disable=wildcard-import
+1 -1
View File
@@ -90,7 +90,7 @@ def make_dvc_bin(
dvc_bin = bench_config.dvc_bin
def _dvc_bin(*args):
return check_output([dvc_bin, *args], text=True) # nosec B603
return check_output([dvc_bin, *args], text=True) # nosec B603 # noqa: S603
_dvc_bin.version = parse_tuple(_dvc_bin("--version")) # type: ignore[attr-defined]
return _dvc_bin
+1 -1
View File
@@ -1 +1 @@
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403, pylint: disable=wildcard-import
+5 -3
View File
@@ -233,8 +233,8 @@ def docker_services(
pytest.skip("disabled for Windows on CI")
try:
subprocess.check_output( # nosec B607, B602
"docker ps", stderr=subprocess.STDOUT, shell=True
subprocess.check_output( # nosec B607, B602,
"docker ps", stderr=subprocess.STDOUT, shell=True # noqa: S602, S607
)
except subprocess.CalledProcessError as err:
out = (err.output or b"").decode("utf-8")
@@ -242,7 +242,9 @@ def docker_services(
try:
cmd = "docker-compose version"
subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True) # nosec
subprocess.check_output(
cmd, stderr=subprocess.STDOUT, shell=True # nosec B602 # noqa: S602
)
except subprocess.CalledProcessError as err:
out = (err.output or b"").decode("utf-8")
pytest.skip(f"docker-compose is not installed: {out}")
+2 -2
View File
@@ -1,5 +1,5 @@
from .benchmarks.fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .benchmarks.fixtures import * # noqa: F403, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403, pylint: disable=wildcard-import
def pytest_generate_tests(metafunc):
+1 -1
View File
@@ -147,7 +147,7 @@ class Console:
return self.write(json.text, stderr=stderr)
return self.rich_print(json, stderr=stderr, soft_wrap=True)
def rich_print(
def rich_print( # noqa: PLR0913
self,
*objects: Any,
sep: str = " ",
+2 -1
View File
@@ -59,7 +59,8 @@ def find_pager():
if not pager:
pager = os.getenv(PAGER_ENV)
if not pager:
if os.system(f"({DEFAULT_PAGER}) 2>{os.devnull}") != 0: # nosec B605
ret = os.system(f"({DEFAULT_PAGER}) 2>{os.devnull}") # nosec B605 # noqa: S605
if ret != 0:
logger.warning(
"Unable to find `less` in the PATH. Check out %s for more info.",
format_link("https://man.dvc.org/pipeline/show"),
+4 -14
View File
@@ -218,23 +218,13 @@ def _visual_center(line, width):
def relpath(path, start=os.curdir):
path = os.fspath(path)
path = os.path.abspath(os.fspath(path))
start = os.path.abspath(os.fspath(start))
# Windows path on different drive than curdir doesn't have relpath
if os.name == "nt":
# Since python 3.8 os.realpath resolves network shares to their UNC
# path. So, to be certain that relative paths correctly captured,
# we need to resolve to UNC path first. We resolve only the drive
# name so that we don't follow any 'real' symlinks on the path
def resolve_network_drive_windows(path_to_resolve):
drive, tail = os.path.splitdrive(path_to_resolve)
return os.path.join(os.path.realpath(drive), tail)
if os.name == "nt" and not os.path.commonprefix([start, path]):
return path
path = resolve_network_drive_windows(os.path.abspath(path))
start = resolve_network_drive_windows(start)
if not os.path.commonprefix([start, path]):
return path
return os.path.relpath(path, start)
@@ -319,7 +309,7 @@ def resolve_paths(repo, out, always_local=False):
def format_link(link):
return "<{blue}{link}{nc}>".format(
return "<{blue}{link}{nc}>".format( # noqa: UP032
blue=colorama.Fore.CYAN, link=link, nc=colorama.Fore.RESET
)
+5 -5
View File
@@ -1,11 +1,11 @@
from collections import defaultdict
from typing import DefaultDict
from ._common import * # noqa, pylint: disable=wildcard-import
from ._json import * # noqa, pylint: disable=wildcard-import
from ._py import * # noqa, pylint: disable=wildcard-import
from ._toml import * # noqa, pylint: disable=wildcard-import
from ._yaml import * # noqa, pylint: disable=wildcard-import
from ._common import * # noqa: F403, pylint: disable=wildcard-import
from ._json import * # noqa: F403, pylint: disable=wildcard-import
from ._py import * # noqa: F403, pylint: disable=wildcard-import
from ._toml import * # noqa: F403, pylint: disable=wildcard-import
from ._yaml import * # noqa: F403, pylint: disable=wildcard-import
LOADERS: DefaultDict[str, LoaderFn] = defaultdict( # noqa: F405
lambda: load_yaml # noqa: F405
+1 -1
View File
@@ -46,7 +46,7 @@ class ThreadPoolExecutor(futures.ThreadPoolExecutor):
if sys.version_info > (3, 9): # pylint: disable=no-else-return
# pylint: disable=unexpected-keyword-arg
return super().shutdown(wait=wait, cancel_futures=cancel_futures)
else:
else: # noqa: RET505
with self._shutdown_lock:
self._shutdown = True
if cancel_futures:
+16 -7
View File
@@ -34,7 +34,7 @@ dependencies = [
"configobj>=5.0.6",
"distro>=1.3",
"dpath<3,>=2.1.0",
"dvc-data>=2.5.0,<2.6.0",
"dvc-data>=2.11.0,<2.12.0",
"dvc-http>=2.29.0",
"dvc-render>=0.3.1,<1",
"dvc-studio-client>=0.9.2,<1",
@@ -75,15 +75,15 @@ gdrive = ["dvc-gdrive==2.20"]
gs = ["dvc-gs==2.22.1"]
hdfs = ["dvc-hdfs==2.19"]
lint = [
"mypy==1.3.0",
"pylint==2.17.4",
"mypy==1.4.1",
"pylint==2.17.5",
"types-colorama",
"types-psutil",
"types-pyinstaller",
"types-requests",
"types-tabulate",
"types-toml",
"types-tqdm",
"types-pyinstaller",
]
oss = ["dvc-oss==2.19"]
s3 = ["dvc-s3==2.23.0"]
@@ -268,8 +268,16 @@ ignored-argument-names = "_.*|args|kwargs"
[tool.ruff]
# external flake8 codes that should be preserved
external = ["B301", "C901", "E302", "W601"]
ignore = ["N818", "S101", "PT004", "PT007", "PT019", "SIM105", "SIM108", "SIM110", "SIM117", "TRY003", "TRY200", "TRY300", "TRY301", "PLR2004", "PLW2901", "RUF005"]
select = ["F", "E", "W", "C90", "N", "UP", "YTT", "S", "BLE", "B", "A", "C4", "T10", "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PT", "Q", "RET501", "RET504", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY", "RUF"]
ignore = [
"N818", "S101", "PT004", "PT007", "RET501", "RET502", "RET503", "SIM105", "SIM108",
"SIM117", "TRY003", "TRY200", "TRY300", "PLR2004", "PLW2901", "RUF012"
]
select = [
"F", "E", "W", "C90", "N", "UP", "YTT", "ASYNC", "S", "BLE", "B", "A", "C4", "T10",
"EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PYI", "PT", "Q", "RSE", "RET",
"SLOT", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY",
"FLY", "PERF101", "RUF",
]
show-source = true
[tool.ruff.flake8-pytest-style]
@@ -292,12 +300,13 @@ ignore-variadic-names = true
known-first-party = ["dvc", "dvc_data", "dvc_objects", "dvc_render", "dvc_task", "tests"]
[tool.ruff.pep8-naming]
ignore-names = ["M", "SCM"]
extend-ignore-names = ["M", "SCM"]
[tool.ruff.per-file-ignores]
"dvc/commands/**" = ["N806"]
"dvc/testing/**" = ["ARG002"]
"dvc/testing/benchmarks/**" = ["ARG001"]
"tests/**" = ["S", "ARG001", "ARG002", "TRY002", "TRY301"]
[tool.ruff.pylint]
max-args = 10
+4 -4
View File
@@ -7,11 +7,11 @@ import pytest
from dvc import env
from dvc.stage import PipelineStage
from dvc.testing.fixtures import * # noqa, pylint: disable=wildcard-import
from dvc.testing.fixtures import * # noqa: F403, pylint: disable=wildcard-import
from .dir_helpers import * # noqa, pylint: disable=wildcard-import
from .remotes import * # noqa, pylint: disable=wildcard-import
from .scripts import * # noqa, pylint: disable=wildcard-import
from .dir_helpers import * # noqa: F403 # pylint: disable=wildcard-import
from .remotes import * # noqa: F403 # pylint: disable=wildcard-import
from .scripts import * # noqa: F403, pylint: disable=wildcard-import
# Prevent updater and analytics from running their processes
os.environ["DVC_TEST"] = "true"
+1 -1
View File
@@ -5,7 +5,7 @@ from funcy import first, get_in
from dvc import api
from dvc.exceptions import OutputNotFoundError, PathMissingError
from dvc.testing.api_tests import TestAPI # noqa, pylint: disable=unused-import
from dvc.testing.api_tests import TestAPI # noqa: F401, pylint: disable=unused-import
from dvc.testing.tmp_dir import make_subrepo
from dvc.utils.fs import remove
+2 -2
View File
@@ -187,7 +187,7 @@ def test_params_show_stages(params_repo):
def test_params_show_stage_addressing(tmp_dir, dvc):
for subdir in {"subdir1", "subdir2"}:
for subdir in ("subdir1", "subdir2"):
subdir = tmp_dir / subdir
subdir.mkdir()
with subdir.chdir():
@@ -195,7 +195,7 @@ def test_params_show_stage_addressing(tmp_dir, dvc):
dvc.run(name="stage-0", cmd="echo stage-0", params=["foo"])
for s in {"subdir1", "subdir2"}:
for s in ("subdir1", "subdir2"):
dvcyaml = os.path.join(s, "dvc.yaml")
assert api.params_show(stages=f"{dvcyaml}:stage-0") == {"foo": 1}
+18
View File
@@ -1,4 +1,6 @@
import logging
import os
from copy import deepcopy
import pytest
@@ -39,6 +41,22 @@ def test_artifacts_read_subdir(tmp_dir, dvc):
}
def test_artifacts_read_bad_name(tmp_dir, dvc, caplog):
bad_name_dvcyaml = deepcopy(dvcyaml)
bad_name_dvcyaml["artifacts"]["bad_name"] = {"type": "model", "path": "bad.pkl"}
(tmp_dir / "dvc.yaml").dump(bad_name_dvcyaml)
artifacts = {
name: Artifact(**values)
for name, values in bad_name_dvcyaml["artifacts"].items()
}
with caplog.at_level(logging.WARNING):
assert tmp_dir.dvc.artifacts.read() == {"dvc.yaml": artifacts}
assert "Can't use 'bad_name' as artifact name (ID)" in caplog.text
def test_artifacts_add_subdir(tmp_dir, dvc):
subdir = tmp_dir / "subdir"
subdir.mkdir()
+2 -1
View File
@@ -1,6 +1,7 @@
import pytest
from tests.unit.repo.experiments.conftest import ( # noqa, pylint disable=unused-argument
# pylint: disable-next=unused-argument
from tests.unit.repo.experiments.conftest import ( # noqa: F401
exp_stage,
failed_exp_stage,
session_app,
+5 -3
View File
@@ -10,7 +10,9 @@ from dvc_ssh.tests.cloud import TEST_SSH_KEY_PATH, TEST_SSH_USER
from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
from dvc.repo.experiments.executor.ssh import SSHExecutor
from dvc.repo.experiments.refs import EXEC_HEAD, EXEC_MERGE
from tests.func.machine.conftest import * # noqa, pylint: disable=wildcard-import
# pylint: disable-next=wildcard-import
from tests.func.machine.conftest import * # noqa: F403
@contextmanager
@@ -119,7 +121,7 @@ def test_reproduce(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
rev,
fs_factory=partial(_ssh_factory, cloud),
)
assert mock_execute.called_once()
mock_execute.assert_called_once()
_name, args, _kwargs = mock_execute.mock_calls[0]
assert f"dvc exp exec-run --infofile {infofile}" in args[0]
@@ -148,7 +150,7 @@ def test_run_machine(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
tmp_dir.gen("params.yaml", "foo: 2")
dvc.experiments.run(exp_stage.addressing, machine="foo")
assert mock_repro.called_once()
mock_repro.assert_called_once()
_name, _args, kwargs = mock_repro.mock_calls[0]
info = kwargs["info"]
url = urlparse(info.git_url)
@@ -488,16 +488,6 @@ def test_run_celery(tmp_dir, scm, dvc, exp_stage, mocker):
assert expected == metrics
def test_run_metrics(tmp_dir, scm, dvc, exp_stage, mocker):
from dvc.cli import main
mocker.patch.object(dvc.experiments, "run", return_value={"abc123": "abc123"})
show_mock = mocker.patch.object(dvc.metrics, "show", return_value={})
main(["exp", "run", "-m"])
assert show_mock.called_once()
def test_checkout_targets_deps(tmp_dir, scm, dvc, exp_stage):
from dvc.utils.fs import remove
+2 -2
View File
@@ -452,7 +452,7 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
assert "metrics.yaml:foo,params.yaml:foo,copy.py" in cap.out
assert f",workspace,baseline,,,3,3,{data_hash}" in cap.out
assert (
",master,baseline,{},,1,1,{}".format(
",master,baseline,{},,1,1,{}".format( # noqa: UP032
_get_rev_isotimestamp(baseline_rev), data_hash
)
in cap.out
@@ -682,7 +682,7 @@ def test_metrics_renaming(tmp_dir, dvc, scm, capsys, copy_script):
assert f",master,baseline,{_get_rev_isotimestamp(scores_rev)},,1,,1" in cap.out
assert (
",{},baseline,{},,,1,1".format(
",{},baseline,{},,,1,1".format( # noqa: UP032
metrics_rev[:7], _get_rev_isotimestamp(metrics_rev)
)
in cap.out
+52
View File
@@ -0,0 +1,52 @@
import pytest
from dvc.parsing import DataResolver, MatrixDefinition
MATRIX_DATA = {
"os": ["win", "linux"],
"pyv": [3.7, 3.8],
"dict": [{"arg1": 1}, {"arg2": 2}],
"list": [["out1", "out11"], ["out2", "out22"]],
}
@pytest.mark.parametrize(
"matrix",
[
MATRIX_DATA,
{
"os": "${os}",
"pyv": "${pyv}",
"dict": "${dict}",
"list": "${list}",
},
],
)
def test_matrix_interpolated(tmp_dir, dvc, matrix):
(tmp_dir / "params.yaml").dump(MATRIX_DATA)
resolver = DataResolver(dvc, tmp_dir.fs_path, {})
data = {
"matrix": matrix,
"cmd": "echo ${item.os} ${item.pyv} ${item.dict}"
" -- ${item.list.0} ${item.list.1}",
}
definition = MatrixDefinition(resolver, resolver.context, "build", data)
assert definition.resolve_all() == {
"[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out2 out22"},
}
+75 -5
View File
@@ -10,7 +10,7 @@ from funcy import lsplit
from dvc.cli import main
from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
from dvc.exceptions import CyclicGraphError, ReproductionError
from dvc.fs import LocalFileSystem, system
from dvc.fs import system
from dvc.output import Output
from dvc.stage import PipelineStage, Stage
from dvc.stage.cache import RunCacheNotSupported
@@ -624,12 +624,14 @@ class TestReproAlreadyCached:
assert stage.outs[0].hash_info != saved_stage.outs[0].hash_info
def test_force_import(self, mocker, tmp_dir, dvc):
from dvc.dependency import base
tmp_dir.dvc_gen("foo", "foo")
ret = main(["import-url", "foo", "bar"])
assert ret == 0
spy_get = mocker.spy(LocalFileSystem, "get")
spy_get = mocker.spy(base, "fs_download")
spy_checkout = mocker.spy(Output, "checkout")
assert main(["unfreeze", "bar.dvc"]) == 0
@@ -848,9 +850,7 @@ def test_repro_dry_no_exec(tmp_dir, dvc):
idir,
"-o",
odir,
'python -c \'import shutil; shutil.copytree("{}", "{}")\''.format(
idir, odir
),
f'python -c \'import shutil; shutil.copytree("{idir}", "{odir}")\'',
]
)
assert ret == 0
@@ -943,6 +943,18 @@ def test_repro_force_downstream(tmp_dir, dvc, copy_script):
assert stages[1].addressing == file3_stage.addressing
def test_repro_force_downstream_do_not_force_independent_stages(tmp_dir, dvc, run_copy):
tmp_dir.gen({"foo": "foo", "bar": "bar"})
foo1 = run_copy("foo", "foo1", name="foo1")
foo2 = run_copy("foo1", "foo2", name="foo2")
run_copy("bar", "bar1", name="bar1")
run_copy("bar1", "bar2", name="bar2")
cat = dvc.run(cmd="cat bar2 foo2", deps=["foo2", "bar2"], name="cat")
tmp_dir.gen("foo", "foobar")
assert dvc.reproduce(force_downstream=True) == [foo1, foo2, cat]
def test_repro_pipeline(
tmp_dir,
dvc,
@@ -1296,3 +1308,61 @@ def test_repro_single_item_with_multiple_targets(tmp_dir, dvc, copy_script):
stage2,
stage1,
]
def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
from dvc.repo import reproduce
(bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
stage1 = dvc.stage.add(
cmd=["python copy.py bar foobar", "exit 1"],
deps=["bar"],
outs=["foobar"],
name="copy-bar-foobar",
)
dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
spy = mocker.spy(reproduce, "_reproduce_stage")
with pytest.raises(ReproductionError):
dvc.reproduce(on_error="keep-going", repro_fn=spy)
bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
stage1_call = mocker.call(
stage1, upstream=[bar_stage], force=False, interactive=False
)
foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
assert spy.call_args_list in (
[bar_call, stage1_call, foo_call],
[foo_call, bar_call, stage1_call],
)
def test_repro_ignore_errors(mocker, tmp_dir, dvc, copy_script):
from dvc.repo import reproduce
(bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
stage1 = dvc.stage.add(
cmd=["python copy.py bar foobar", "exit 1"],
deps=["bar"],
outs=["foobar"],
name="copy-bar-foobar",
)
stage2 = dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
spy = mocker.spy(reproduce, "_reproduce_stage")
dvc.reproduce(on_error="ignore", repro_fn=spy)
bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
stage1_call = mocker.call(
stage1, upstream=[bar_stage], force=False, interactive=False
)
stage2_call = mocker.call(
stage2,
upstream=[foo_stage, stage1],
force=False,
interactive=False,
)
assert spy.call_args_list in (
[bar_call, stage1_call, foo_call, stage2_call],
[foo_call, bar_call, stage1_call, stage2_call],
)
+2 -2
View File
@@ -452,9 +452,9 @@ def temporary_windows_drive(tmp_path_factory):
if len(s) > 0
]
new_drive_name = [
new_drive_name = next(
letter for letter in string.ascii_uppercase if letter not in drives
][0]
)
new_drive = f"{new_drive_name}:"
target_path = tmp_path_factory.mktemp("tmp_windows_drive")
+2 -4
View File
@@ -218,7 +218,7 @@ def test_list_bad_args(tmp_dir, dvc, caplog, args):
def test_set_invalid_key(dvc):
with pytest.raises(ConfigError, match=r"extra keys not allowed"): # noqa: PT012
with pytest.raises(ConfigError, match=r"extra keys not allowed"):
with dvc.config.edit() as conf:
conf["core"]["invalid_key"] = "value"
@@ -227,9 +227,7 @@ def test_merging_two_levels(dvc):
with dvc.config.edit() as conf:
conf["remote"]["test"] = {"url": "ssh://example.com"}
with pytest.raises( # noqa: PT012
ConfigError, match=r"expected 'url' for dictionary value"
):
with pytest.raises(ConfigError, match=r"expected 'url' for dictionary value"):
with dvc.config.edit("global") as conf:
conf["remote"]["test"] = {"password": "1"}
+5 -2
View File
@@ -10,7 +10,9 @@ from dvc.cli import main
from dvc.exceptions import CheckoutError
from dvc.repo.open_repo import clean_repos
from dvc.stage.exceptions import StageNotFound
from dvc.testing.remote_tests import TestRemote # noqa, pylint: disable=unused-import
from dvc.testing.remote_tests import ( # noqa: F401, pylint: disable=unused-import
TestRemote,
)
from dvc.utils.fs import remove
from dvc_data.hashfile.db import HashFileDB
from dvc_data.hashfile.db.local import LocalHashFileDB
@@ -203,7 +205,8 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
# Removing cache will invalidate existing state entries
dvc.cache.local.clear()
dvc.config["remote"]["upstream"]["verify"] = True
with dvc.config.edit() as conf:
conf["remote"]["upstream"]["verify"] = True
dvc.pull()
assert hash_spy.call_count == 10
+11 -31
View File
@@ -196,11 +196,7 @@ def test_gc_cloud_positive(tmp_dir, dvc, tmp_path_factory, local_remote):
assert main(["gc", "-vf", flag]) == 0
def test_gc_cloud_remove_order(tmp_dir, scm, dvc, tmp_path_factory, mocker):
storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
dvc.config["remote"]["local_remote"] = {"url": storage}
dvc.config["core"]["remote"] = "local_remote"
def test_gc_cloud_remove_order(tmp_dir, scm, dvc, mocker, local_remote):
(standalone, dir1, dir2) = tmp_dir.dvc_gen(
{
"file1": "standalone",
@@ -321,11 +317,7 @@ def test_date(tmp_dir, scm, dvc):
) # "modified, again"
def test_gc_not_in_remote(tmp_dir, scm, dvc, tmp_path_factory, mocker):
storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
dvc.config["remote"]["local_remote"] = {"url": storage}
dvc.config["core"]["remote"] = "local_remote"
def test_gc_not_in_remote(tmp_dir, scm, dvc, mocker, local_remote):
(standalone, dir1, dir2) = tmp_dir.dvc_gen(
{
"file1": "standalone",
@@ -358,12 +350,9 @@ def test_gc_not_in_remote(tmp_dir, scm, dvc, tmp_path_factory, mocker):
)
def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, tmp_path_factory, mocker):
storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
dvc.config["remote"]["local_remote"] = {"url": storage}
dvc.config["core"]["remote"] = "local_remote"
other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
dvc.config["remote"]["other_remote"] = {"url": other_storage}
def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, mocker, make_remote):
make_remote("local_remote", typ="local")
make_remote("other_remote", typ="local", default=False)
tmp_dir.dvc_gen(
{
@@ -385,15 +374,9 @@ def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, tmp_path_factory, mocker
assert len(mocked_remove.mock_calls) == 3
def test_gc_not_in_remote_with_remote_field(
tmp_dir, scm, dvc, tmp_path_factory, mocker
):
storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
dvc.config["remote"]["local_remote"] = {"url": storage}
dvc.config["core"]["remote"] = "local_remote"
other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
dvc.config["remote"]["other_remote"] = {"url": other_storage}
def test_gc_not_in_remote_with_remote_field(tmp_dir, scm, dvc, mocker, make_remote):
make_remote("local_remote", typ="local")
make_remote("other_remote", typ="local", default=False)
text = textwrap.dedent(
"""\
@@ -420,12 +403,9 @@ def test_gc_not_in_remote_cloud(tmp_dir, scm, dvc):
dvc.gc(workspace=True, not_in_remote=True, cloud=True)
def test_gc_cloud_remote_field(tmp_dir, scm, dvc, tmp_path_factory, mocker):
storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
dvc.config["remote"]["local_remote"] = {"url": storage}
dvc.config["core"]["remote"] = "local_remote"
other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
dvc.config["remote"]["other_remote"] = {"url": other_storage}
def test_gc_cloud_remote_field(tmp_dir, scm, dvc, mocker, make_remote):
make_remote("local_remote", typ="local")
make_remote("other_remote", typ="local", default=False)
text = textwrap.dedent(
"""\
+16
View File
@@ -64,6 +64,22 @@ def test_get_repo_dir(tmp_dir, erepo_dir):
assert (tmp_dir / "dir_imported").read_text() == {"file": "contents"}
def test_get_repo_broken_dir(tmp_dir, erepo_dir):
import shutil
from dvc_data.index import DataIndexDirError
with erepo_dir.chdir():
erepo_dir.dvc_gen({"broken": {"file": "contents"}})
erepo_dir.dvc.cache.local.clear()
shutil.rmtree(erepo_dir / "broken")
with pytest.raises(DataIndexDirError):
Repo.get(os.fspath(erepo_dir), "broken", "out")
assert not (tmp_dir / "out").exists()
@pytest.mark.parametrize(
"erepo", [pytest.lazy_fixture("git_dir"), pytest.lazy_fixture("erepo_dir")]
)
+16 -1
View File
@@ -12,6 +12,7 @@ from dvc.scm import Git
from dvc.stage.exceptions import StagePathNotFoundError
from dvc.testing.tmp_dir import make_subrepo
from dvc.utils.fs import remove
from dvc_data.index.index import DataIndexDirError
def test_import(tmp_dir, scm, dvc, erepo_dir):
@@ -96,6 +97,19 @@ def test_import_dir(tmp_dir, scm, dvc, erepo_dir):
}
def test_import_broken_dir(tmp_dir, scm, dvc, erepo_dir):
with erepo_dir.chdir():
erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create dir")
erepo_dir.dvc.cache.local.clear()
remove(erepo_dir / "dir")
with pytest.raises(DataIndexDirError):
dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
assert not (tmp_dir / "dir_imported").exists()
assert not (tmp_dir / "dir_imported.dvc").exists()
def test_import_file_from_dir(tmp_dir, scm, dvc, erepo_dir):
with erepo_dir.chdir():
erepo_dir.dvc_gen(
@@ -346,7 +360,8 @@ def test_push_wildcard_from_bare_git_repo(
with dvc_repo.chdir():
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dirextra")
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
with pytest.raises(DataIndexDirError):
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
@pytest.mark.parametrize("dname", [".", "dir", "dir/subdir"])
+37
View File
@@ -583,3 +583,40 @@ def test_broken_symlink(tmp_dir, dvc):
"path": "link",
},
]
def test_ls_broken_dir(tmp_dir, dvc):
from dvc_data.index import DataIndexDirError
tmp_dir.dvc_gen(
{
"broken": {"baz": "baz"},
}
)
shutil.rmtree(tmp_dir / "broken")
dvc.cache.local.clear()
tmp_dir.dvc_gen(
{
"foo": "foo",
"dir": {"bar": "bar"},
}
)
entries = Repo.ls(os.fspath(tmp_dir))
assert entries == [
{"isdir": False, "isexec": False, "isout": False, "path": ".dvcignore"},
{"isdir": True, "isexec": False, "isout": True, "path": "broken"},
{"isdir": False, "isexec": False, "isout": False, "path": "broken.dvc"},
{"isdir": True, "isexec": False, "isout": True, "path": "dir"},
{"isdir": False, "isexec": False, "isout": False, "path": "dir.dvc"},
{"isdir": False, "isexec": False, "isout": True, "path": "foo"},
{"isdir": False, "isexec": False, "isout": False, "path": "foo.dvc"},
]
with pytest.raises(DataIndexDirError):
Repo.ls(os.fspath(tmp_dir), "broken")
with pytest.raises(DataIndexDirError):
Repo.ls(os.fspath(tmp_dir), recursive=True)
+23 -5
View File
@@ -84,7 +84,7 @@ def test_default_wdir_ignored_in_checksum(tmp_dir, dvc):
stage = dvc.run(cmd="cp bar foo", deps=["bar"], outs=["foo"], name="copy-foo-bar")
d = stage.dumpd()
assert Stage.PARAM_WDIR not in d.keys()
assert Stage.PARAM_WDIR not in d
d = load_yaml("dvc.yaml")
assert Stage.PARAM_WDIR not in d["stages"]["copy-foo-bar"]
@@ -230,17 +230,35 @@ def test_parent_repo_collect_stages(tmp_dir, scm, dvc):
@pytest.mark.parametrize("with_deps", (False, True))
def test_collect_symlink(tmp_dir, dvc, with_deps):
from dvc.exceptions import StageNotFoundError
tmp_dir.gen({"data": {"foo": "foo contents"}})
foo_path = os.path.join("data", "foo")
dvc.add(foo_path)
data_link = tmp_dir / "data_link"
data_link.symlink_to("data")
stage = list(
dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
)[0]
assert stage.addressing == f"{foo_path}.dvc"
if with_deps:
# NOTE: with_deps means that we'll need to collect and use dvcfiles in the repo
# and we currently don't follow symlinks when collecting those, so it will not
# be able to find the target stage.
with pytest.raises(StageNotFoundError):
dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
else:
stage = next(
iter(
dvc.stage.collect(
target=str(data_link / "foo.dvc"), with_deps=with_deps
)
)
)
assert stage.addressing == os.path.join("data_link", "foo.dvc")
stage = next(iter(dvc.stage.collect(target=f"{foo_path}.dvc", with_deps=with_deps)))
assert stage.addressing == os.path.join("data", "foo.dvc")
def test_stage_strings_representation(tmp_dir, dvc, run_copy):
+1 -1
View File
@@ -377,7 +377,7 @@ def test_repo_with_removed_plots(tmp_dir, capsys, repo_with_plots):
remove("confusion.json")
remove("image.png")
for s in {"show", "diff"}:
for s in ("show", "diff"):
_, json_result, split_json_result = call(capsys, subcommand=s)
errors = [
{
+1 -1
View File
@@ -1,4 +1,4 @@
from dvc_ssh.tests.fixtures import ( # noqa, pylint: disable=unused-import
from dvc_ssh.tests.fixtures import ( # noqa: F401, pylint: disable=unused-import
make_ssh,
ssh,
ssh_server,
-3
View File
@@ -1,3 +0,0 @@
extend = "../pyproject.toml"
extend-select = ["RUF005"]
extend-ignore = ["S", "RUF001", "ARG001", "ARG002", "TRY002"]
+76 -5
View File
@@ -18,32 +18,103 @@ def _test_cli(mocker, *args):
def test_list(mocker):
url = "local_dir"
m = _test_cli(mocker, url)
m.assert_called_once_with(url, None, recursive=False, rev=None, dvc_only=False)
m.assert_called_once_with(
url,
None,
recursive=False,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
def test_list_recursive(mocker):
url = "local_dir"
m = _test_cli(mocker, url, "-R")
m.assert_called_once_with(url, None, recursive=True, rev=None, dvc_only=False)
m.assert_called_once_with(
url,
None,
recursive=True,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
def test_list_git_ssh_rev(mocker):
url = "[email protected]:repo"
m = _test_cli(mocker, url, "--rev", "123")
m.assert_called_once_with(url, None, recursive=False, rev="123", dvc_only=False)
m.assert_called_once_with(
url,
None,
recursive=False,
rev="123",
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
def test_list_targets(mocker):
url = "local_dir"
target = "subdir"
m = _test_cli(mocker, url, target)
m.assert_called_once_with(url, target, recursive=False, rev=None, dvc_only=False)
m.assert_called_once_with(
url,
target,
recursive=False,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
def test_list_outputs_only(mocker):
url = "local_dir"
m = _test_cli(mocker, url, None, "--dvc-only")
m.assert_called_once_with(url, None, recursive=False, rev=None, dvc_only=True)
m.assert_called_once_with(
url,
None,
recursive=False,
rev=None,
dvc_only=True,
config=None,
remote=None,
remote_config=None,
)
def test_list_config(mocker):
url = "local_dir"
m = _test_cli(
mocker,
url,
None,
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
)
m.assert_called_once_with(
url,
None,
recursive=False,
rev=None,
dvc_only=False,
config="myconfig",
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
def test_show_json(mocker, capsys):
+17
View File
@@ -0,0 +1,17 @@
import logging
import pytest
from dvc.cli import main
from dvc.commands.completion import SUPPORTED_SHELLS
@pytest.mark.parametrize("shell", SUPPORTED_SHELLS)
def test_completion(caplog, capsys, shell):
with caplog.at_level(logging.INFO):
assert main(["completion", "-s", shell]) == 0
assert not caplog.text
out, err = capsys.readouterr()
assert not err
assert out
+16 -1
View File
@@ -14,6 +14,13 @@ def test_get(mocker):
"version",
"--jobs",
"4",
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
]
)
assert cli_args.func == CmdGet
@@ -24,7 +31,15 @@ def test_get(mocker):
assert cmd.run() == 0
m.assert_called_once_with(
"repo_url", path="src", out="out", rev="version", jobs=4, force=False
"repo_url",
path="src",
out="out",
rev="version",
jobs=4,
config="myconfig",
force=False,
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
+48
View File
@@ -0,0 +1,48 @@
import logging
from argparse import SUPPRESS, ArgumentParser
from typing import Tuple
import pytest
import shtab
from dvc.cli import main
from dvc.cli.parser import get_main_parser
def command_tuples():
root: Tuple[str, ...] = ()
commands = [root]
def recurse_parser(parser: ArgumentParser, parents: Tuple[str, ...] = root) -> None:
for positional in parser._get_positional_actions():
if positional.help != SUPPRESS and isinstance(positional.choices, dict):
public_cmds = shtab.get_public_subcommands(positional)
for subcmd, subparser in positional.choices.items():
cmd = (*parents, subcmd)
if subcmd in public_cmds:
commands.append(cmd)
recurse_parser(subparser, cmd)
main_parser = get_main_parser()
recurse_parser(main_parser)
return sorted(commands)
def pytest_generate_tests(metafunc):
def ids(values):
if values:
return "-".join(values)
return "dvc"
metafunc.parametrize("command_tuples", command_tuples(), ids=ids)
def test_help(caplog, capsys, command_tuples):
with caplog.at_level(logging.INFO), pytest.raises(SystemExit) as e:
main([*command_tuples, "--help"])
assert e.value.code == 0
assert not caplog.text
out, err = capsys.readouterr()
assert not err
assert out
+16
View File
@@ -14,6 +14,13 @@ def test_import(mocker, dvc):
"version",
"--jobs",
"3",
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
]
)
assert cli_args.func == CmdImport
@@ -31,6 +38,9 @@ def test_import(mocker, dvc):
no_exec=False,
no_download=False,
jobs=3,
config="myconfig",
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
@@ -61,6 +71,9 @@ def test_import_no_exec(mocker, dvc):
no_exec=True,
no_download=False,
jobs=None,
config=None,
remote=None,
remote_config=None,
)
@@ -91,4 +104,7 @@ def test_import_no_download(mocker, dvc):
no_exec=False,
no_download=True,
jobs=None,
config=None,
remote=None,
remote_config=None,
)
+1
View File
@@ -14,6 +14,7 @@ common_arguments = {
"pull": False,
"allow_missing": False,
"targets": [],
"on_error": "fail",
}
repro_arguments = {
"run_cache": True,
-1
View File
@@ -47,7 +47,6 @@ def test_stage_add(mocker, dvc, command, parsed_command):
"file:param1,param2",
"--params",
"param3",
"--external",
"--desc",
"description",
"--force",
+10
View File
@@ -162,6 +162,16 @@ def test_isdir_mixed(tmp_dir, dvc):
assert not fs.isfile("dir")
def test_ls_dirty(tmp_dir, dvc):
tmp_dir.dvc_gen({"data": "data"})
(tmp_dir / "data").unlink()
tmp_dir.gen({"data": {"foo": "foo", "bar": "bar"}})
fs = DVCFileSystem(repo=dvc)
assert set(fs.ls("data")) == {"data/foo", "data/bar"}
@pytest.mark.parametrize(
"dvcfiles,extra_expected",
[

Some files were not shown because too many files have changed in this diff Show More