diff --git a/dvc/_debug.py b/dvc/_debug.py index 4197619b2..63b6d9774 100644 --- a/dvc/_debug.py +++ b/dvc/_debug.py @@ -119,7 +119,7 @@ def debug(): try: import ipdb as pdb # noqa: T100, pylint: disable=import-error except ImportError: - import pdb # noqa: T100 + import pdb # type: ignore[no-redef] # noqa: T100 pdb.post_mortem() raise # prevent from jumping ahead diff --git a/dvc/analytics.py b/dvc/analytics.py index 3aa467bf0..7f196c02c 100644 --- a/dvc/analytics.py +++ b/dvc/analytics.py @@ -125,7 +125,7 @@ def _system_info(): system = platform.system() if system == "Windows": - version = sys.getwindowsversion() + version = sys.getwindowsversion() # type: ignore[attr-defined] return { "os": "windows", diff --git a/dvc/api/data.py b/dvc/api/data.py index a063612f1..a26885909 100644 --- a/dvc/api/data.py +++ b/dvc/api/data.py @@ -38,7 +38,11 @@ class _OpenContextManager(GCM): self, func, args, kwds ): # pylint: disable=super-init-not-called self.gen = func(*args, **kwds) - self.func, self.args, self.kwds = func, args, kwds + self.func, self.args, self.kwds = ( # type: ignore[assignment] + func, + args, + kwds, + ) def __getattr__(self, name): raise AttributeError( diff --git a/dvc/api/params.py b/dvc/api/params.py index 6cba9e98a..00c03f0d9 100644 --- a/dvc/api/params.py +++ b/dvc/api/params.py @@ -1,3 +1,4 @@ +import typing from collections import Counter from typing import Dict, Iterable, Optional, Union @@ -242,11 +243,11 @@ def params_show( raise exception def _postprocess(params): - processed = {} + processed: Dict[str, Dict] = {} for rev, rev_data in params.items(): processed[rev] = {} - counts = Counter() + counts: typing.Counter[str] = Counter() for file_data in rev_data["data"].values(): for k in file_data["data"]: counts[k] += 1 diff --git a/dvc/cli/__init__.py b/dvc/cli/__init__.py index 4d96e3ebf..b85a0dd9f 100644 --- a/dvc/cli/__init__.py +++ b/dvc/cli/__init__.py @@ -167,12 +167,12 @@ def main(argv=None): # noqa: C901 elif args.verbose == 1: level = logging.DEBUG elif args.verbose > 1: - level = logging.TRACE + level = logging.TRACE # type: ignore[attr-defined] if level is not None: set_loggers_level(level) - logger.trace(args) + logger.trace(args) # type: ignore[attr-defined] if not sys.stdout.closed and not args.quiet: from dvc.ui import ui diff --git a/dvc/commands/data_sync.py b/dvc/commands/data_sync.py index a8fad6468..3258a0b60 100644 --- a/dvc/commands/data_sync.py +++ b/dvc/commands/data_sync.py @@ -122,7 +122,7 @@ def shared_parent_parser(): "Limit command scope to these tracked files/directories, " ".dvc files and stage names." ), - ).complete = completion.DVC_FILE + ).complete = completion.DVC_FILE # type: ignore[attr-defined] return parent_parser diff --git a/dvc/commands/experiments/show.py b/dvc/commands/experiments/show.py index 208d3a347..4bb7a44f6 100644 --- a/dvc/commands/experiments/show.py +++ b/dvc/commands/experiments/show.py @@ -4,7 +4,7 @@ import os import re from collections import Counter, OrderedDict, defaultdict from datetime import date, datetime -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, Dict, Set from funcy import lmap @@ -37,9 +37,9 @@ def _update_names(names, items): def _collect_names(all_experiments, **kwargs): - metric_names = defaultdict(dict) - param_names = defaultdict(dict) - deps_names = set() + metric_names: Dict[str, Dict[str, Any]] = defaultdict(dict) + param_names: Dict[str, Dict[str, Any]] = defaultdict(dict) + deps_names: Set[str] = set() for _, experiments in all_experiments.items(): for exp_data in experiments.values(): @@ -328,7 +328,7 @@ def show_experiments( # noqa: C901 metric_names, param_names, deps_names = _collect_names(all_experiments) - headers = [ + headers_list = [ "Experiment", "rev", "typ", @@ -340,13 +340,13 @@ def show_experiments( # noqa: C901 names = {**metric_names, **param_names} counter = Counter(flatten_list([list(a.keys()) for a in names.values()])) - counter.update(headers) + counter.update(headers_list) metric_headers = _normalize_headers(metric_names, counter) param_headers = _normalize_headers(param_names, counter) td = experiments_table( all_experiments, - headers, + headers_list, metric_headers, metric_names, param_headers, diff --git a/dvc/commands/machine.py b/dvc/commands/machine.py index c89f0ba30..a208e73f3 100644 --- a/dvc/commands/machine.py +++ b/dvc/commands/machine.py @@ -177,6 +177,7 @@ class CmdMachineRename(CmdBase): self._check_exists(conf) conf["machine"][self.args.new] = conf["machine"][self.args.name] try: + assert self.repo.machine self.repo.machine.rename(self.args.name, self.args.new) except DvcException as error: del conf["machine"][self.args.new] diff --git a/dvc/commands/plots.py b/dvc/commands/plots.py index 9451db875..71e3e5f0f 100644 --- a/dvc/commands/plots.py +++ b/dvc/commands/plots.py @@ -89,6 +89,7 @@ class CmdPlots(CmdBase): "html_template", None ) if html_template_path and not os.path.isabs(html_template_path): + assert self.repo.dvc_dir html_template_path = os.path.join( self.repo.dvc_dir, html_template_path ) diff --git a/dvc/commands/stage.py b/dvc/commands/stage.py index 6c3262bc8..85936d4fb 100644 --- a/dvc/commands/stage.py +++ b/dvc/commands/stage.py @@ -95,7 +95,7 @@ class CmdStageList(CmdBase): stages = self._get_stages() data = prepare_stages_data(stages, description=not self.args.name_only) - ui.table(data.items()) + ui.table(list(data.items())) return 0 diff --git a/dvc/compare.py b/dvc/compare.py index bbd19a406..d21b4d0db 100644 --- a/dvc/compare.py +++ b/dvc/compare.py @@ -328,8 +328,9 @@ def _format_field( ) -> str: def _format(_val): if isinstance(_val, float) and precision: - func = round if round_digits else _normalize_float - return func(_val, precision) + if round_digits: + return round(val, precision) + return _normalize_float(val, precision) if isinstance(_val, abc.Mapping): return {k: _format(v) for k, v in _val.items()} if isinstance(_val, list): diff --git a/dvc/config.py b/dvc/config.py index 2388c4d86..0e41f02a1 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -291,7 +291,7 @@ class Config(dict): return Schema(dirs_schema, extra=ALLOW_EXTRA)(conf) def load_config_to_level(self, level=None): - merged_conf = {} + merged_conf: Dict = {} for merge_level in self.LEVELS: if merge_level == level: break @@ -337,7 +337,7 @@ class Config(dict): def _parse_named(conf): - result = {"remote": {}, "machine": {}} + result: Dict[str, Dict] = {"remote": {}, "machine": {}} for section, val in conf.items(): match = re_find(r'^\s*(remote|machine)\s*"(.*)"\s*$', section) diff --git a/dvc/daemon.py b/dvc/daemon.py index 50e0d82c8..5dfde30ae 100644 --- a/dvc/daemon.py +++ b/dvc/daemon.py @@ -25,21 +25,24 @@ def _popen(cmd, **kwargs): def _spawn_windows(cmd, env): - from subprocess import ( - CREATE_NEW_PROCESS_GROUP, - CREATE_NO_WINDOW, - STARTF_USESHOWWINDOW, - STARTUPINFO, - ) + if sys.platform == "win32": + from subprocess import ( + CREATE_NEW_PROCESS_GROUP, + CREATE_NO_WINDOW, + STARTF_USESHOWWINDOW, + STARTUPINFO, + ) - # https://stackoverflow.com/a/7006424 - # https://bugs.python.org/issue41619 - creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW + # https://stackoverflow.com/a/7006424 + # https://bugs.python.org/issue41619 + creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW - startupinfo = STARTUPINFO() - startupinfo.dwFlags |= STARTF_USESHOWWINDOW + startupinfo = STARTUPINFO() + startupinfo.dwFlags |= STARTF_USESHOWWINDOW - _popen(cmd, env=env, creationflags=creationflags, startupinfo=startupinfo) + _popen( + cmd, env=env, creationflags=creationflags, startupinfo=startupinfo + ) def _spawn_posix(cmd, env): diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 8c70bf878..965e75168 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -1,5 +1,5 @@ from collections import defaultdict -from typing import Any, Mapping +from typing import Any, Mapping, Set from dvc.output import ARTIFACT_SCHEMA, DIR_FILES_SCHEMA, Output @@ -59,7 +59,7 @@ def _merge_params(s_list): default_file = ParamsDependency.DEFAULT_PARAMS_FILE # figure out completely tracked params file, and ignore specific keys - wholly_tracked = set() + wholly_tracked: Set[str] = set() for key in s_list: if not isinstance(key, dict): continue diff --git a/dvc/dependency/param.py b/dvc/dependency/param.py index 10257f3bd..4b13b0cb6 100644 --- a/dvc/dependency/param.py +++ b/dvc/dependency/param.py @@ -104,7 +104,7 @@ class ParamsDependency(Dependency): from funcy import ldistinct - status = defaultdict(dict) + status: Dict[str, Any] = defaultdict(dict) info = self.hash_info.value if self.hash_info else {} actual = self.read_params() @@ -165,7 +165,7 @@ class ParamsDependency(Dependency): if not self.exists: raise self.DoesNotExistError(self) - if not self.isfile and not self.isdir: + if not self.isfile() and not self.isdir(): raise self.IsNotFileOrDirError(self) self.ignore() diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index 7013c71ba..fe19d4ec8 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -5,7 +5,6 @@ from typing import ( TYPE_CHECKING, ContextManager, Dict, - Iterable, Optional, Set, Tuple, @@ -169,7 +168,7 @@ class RepoDependency(Dependency): return used_obj_ids, meta, obj def _check_circular_import( - self, odb: "ObjectDB", obj_ids: Iterable[str] + self, odb: "ObjectDB", obj_ids: Set["HashInfo"] ) -> None: from dvc.exceptions import CircularImportError from dvc.fs.dvc import DVCFileSystem diff --git a/dvc/external_repo.py b/dvc/external_repo.py index b34032a93..de235ec4f 100644 --- a/dvc/external_repo.py +++ b/dvc/external_repo.py @@ -3,7 +3,7 @@ import os import tempfile import threading from contextlib import contextmanager -from typing import TYPE_CHECKING, Collection, Dict, Iterator, Optional +from typing import TYPE_CHECKING, Collection, Dict, Iterator, Optional, Tuple from funcy import retry, wrap_with @@ -114,7 +114,7 @@ def erepo_factory(url, root_dir, cache_config): return make_repo -CLONES: Dict[str, str] = {} +CLONES: Dict[str, Tuple[str, bool]] = {} CACHE_DIRS: Dict[str, str] = {} @@ -198,7 +198,7 @@ def _clone_default_branch(url, rev, for_write=False): # noqa: C901 """ from scmrepo.git import Git - clone_path, shallow = CLONES.get(url, (None, False)) + clone_path, shallow = CLONES.get(url) or (None, False) git = None try: diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py index 165ce0984..b9e2af672 100644 --- a/dvc/fs/dvc.py +++ b/dvc/fs/dvc.py @@ -31,7 +31,7 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -RepoFactory = Union[Callable[[str], "Repo"], Type["Repo"]] +RepoFactory = Union[Callable[..., "Repo"], Type["Repo"]] Key = Tuple[str, ...] @@ -129,7 +129,7 @@ class _DVCFileSystem(AbstractFileSystem): # pylint:disable=abstract-method repo = self._make_repo( url=url, rev=rev, subrepos=subrepos, **repo_kwargs ) - assert repo + assert repo is not None # pylint: disable=protected-access repo_factory = repo._fs_conf["repo_factory"] @@ -142,6 +142,7 @@ class _DVCFileSystem(AbstractFileSystem): # pylint:disable=abstract-method def _getcwd(): relparts = () + assert repo is not None if repo.fs.path.isin(repo.fs.path.getcwd(), repo.root_dir): relparts = repo.fs.path.relparts( repo.fs.path.getcwd(), repo.root_dir @@ -182,8 +183,6 @@ class _DVCFileSystem(AbstractFileSystem): # pylint:disable=abstract-method @property def repo_url(self): - if self.repo is None: - return None return self.repo.url @classmethod diff --git a/dvc/lock.py b/dvc/lock.py index 646188602..050c0d01a 100644 --- a/dvc/lock.py +++ b/dvc/lock.py @@ -131,6 +131,7 @@ class Lock(LockBase): if not self.is_locked: raise DvcException("Unlock called on an unlocked lock") + assert self._lock self._lock.close() self._lock = None diff --git a/dvc/logger.py b/dvc/logger.py index b48457533..f43475b1c 100644 --- a/dvc/logger.py +++ b/dvc/logger.py @@ -204,7 +204,7 @@ def disable_other_loggers(): loggerDict = logging.root.manager.loggerDict # pylint: disable=no-member for logger_name, logger in loggerDict.items(): if logger_name != "dvc" and not logger_name.startswith("dvc."): - logger.disabled = True + logger.disabled = True # type: ignore[union-attr] def set_loggers_level(level: int = logging.INFO) -> None: diff --git a/dvc/output.py b/dvc/output.py index d77b334fd..6122c5fdc 100644 --- a/dvc/output.py +++ b/dvc/output.py @@ -144,7 +144,7 @@ def _split_dict(d, keys): def _merge_data(s_list): - d = defaultdict(dict) + d: Dict[str, Dict] = defaultdict(dict) for key in s_list: if isinstance(key, str): d[key].update({}) @@ -316,7 +316,7 @@ class Output: and isinstance(stage.repo.fs, LocalFileSystem) and path_isin(path, stage.repo.root_dir) ): - self.def_path = relpath(path, stage.wdir) + self.def_path: str = relpath(path, stage.wdir) self.fs = stage.repo.fs else: self.def_path = path diff --git a/dvc/parsing/context.py b/dvc/parsing/context.py index 54046ba3b..f2c02fe81 100644 --- a/dvc/parsing/context.py +++ b/dvc/parsing/context.py @@ -306,7 +306,7 @@ class Context(CtxDict): """ super().__init__(*args, **kwargs) self._track = False - self._tracked_data = defaultdict(dict) + self._tracked_data: Dict[str, Dict] = defaultdict(dict) self.imports = {} self._reserved_keys = {} diff --git a/dvc/progress.py b/dvc/progress.py index 13e7a5ca9..082c875bf 100644 --- a/dvc/progress.py +++ b/dvc/progress.py @@ -147,10 +147,16 @@ class Tqdm(tqdm): def format_dict(self): """inject `ncols_desc` to fill the display width (`ncols`)""" d = super().format_dict - ncols = d["ncols"] or 80 + ncols: int = d["ncols"] or 80 # assumes `bar_format` has max one of ("ncols_desc" & "ncols_info") ncols_left = ( - ncols - len(self.format_meter(ncols_desc=1, ncols_info=1, **d)) + 1 + ncols + - len( + self.format_meter( # type: ignore[call-arg] + ncols_desc=1, ncols_info=1, **d + ) + ) + + 1 ) ncols_left = max(ncols_left, 0) if ncols_left: diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 69059dab4..9723918ff 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -244,9 +244,9 @@ class Repo: self._ignore() - self.metrics = Metrics(self) + self.metrics: Metrics = Metrics(self) self.plots: Plots = Plots(self) - self.params = Params(self) + self.params: Params = Params(self) self.stage_collection_error_handler: Optional[ Callable[[str, Exception], None] @@ -403,10 +403,11 @@ class Repo: return self.odb.repo.unprotect(target) def _ignore(self): - flist = [self.config.files["local"], self.tmp_dir] - + flist = [self.config.files["local"]] + if tmp_dir := self.tmp_dir: + flist.append(tmp_dir) if path_isin(self.odb.repo.path, self.root_dir): - flist += [self.odb.repo.path] + flist.append(self.odb.repo.path) for file in flist: self.scm_context.ignore(file) diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py index 88611b394..8e5db918e 100644 --- a/dvc/repo/checkout.py +++ b/dvc/repo/checkout.py @@ -1,6 +1,6 @@ import logging import os -from typing import TYPE_CHECKING, Set +from typing import TYPE_CHECKING, Dict, List, Set from dvc.exceptions import ( CheckoutError, @@ -81,7 +81,12 @@ def checkout( ): from dvc.fs.callbacks import Callback - stats = {"added": [], "deleted": [], "modified": [], "failed": []} + stats: Dict[str, List[str]] = { + "added": [], + "deleted": [], + "modified": [], + "failed": [], + } if not targets: targets = [None] stats["deleted"] = _remove_unused_links(self) diff --git a/dvc/repo/index.py b/dvc/repo/index.py index 2c25c486d..2e1e6b99a 100644 --- a/dvc/repo/index.py +++ b/dvc/repo/index.py @@ -46,7 +46,9 @@ def log_walk(seq): start = time.perf_counter() yield root, dirs, files duration = format_time(time.perf_counter() - start) - logger.trace("%s in collecting stages from %s", duration, root) + logger.trace( # type: ignore[attr-defined] + "%s in collecting stages from %s", duration, root + ) def collect_files( diff --git a/dvc/repo/plots/__init__.py b/dvc/repo/plots/__init__.py index 3e69f9afb..ad89af4c7 100644 --- a/dvc/repo/plots/__init__.py +++ b/dvc/repo/plots/__init__.py @@ -605,7 +605,7 @@ def _load_sv(path, fs, delimiter=",", header=True): fieldnames=[str(i) for i in range(len(first_row))], ) - fieldnames = reader.fieldnames + fieldnames = reader.fieldnames or [] data = list(reader) return [ diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 37f1c0dd1..586087ab0 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -1,6 +1,6 @@ import logging from functools import partial -from typing import TYPE_CHECKING, Iterator +from typing import TYPE_CHECKING, Iterator, List from dvc.exceptions import DvcException, ReproductionError from dvc.repo.scm_context import scm_context @@ -16,7 +16,7 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -def _reproduce_stage(stage: "Stage", **kwargs): +def _reproduce_stage(stage: "Stage", **kwargs) -> List["Stage"]: def _run_callback(repro_callback): stage.dump(update_pipeline=False) _track_stage(stage) @@ -173,9 +173,9 @@ def _reproduce_stages( # noqa: C901 force_downstream = kwargs.pop("force_downstream", False) result = [] - unchanged = [] + unchanged: List["Stage"] = [] # `ret` is used to add a cosmetic newline. - ret = [] + ret: List["Stage"] = [] checkpoint_func = kwargs.pop("checkpoint_func", None) for i, stage in enumerate(steps): @@ -234,7 +234,7 @@ def _get_steps(G, stages, downstream, single_item): # NOTE: disconnect frozen stage from its dependencies active.remove_edges_from(G.out_edges(stage)) - all_pipelines = [] + all_pipelines: List["Stage"] = [] for stage in stages: if downstream: # NOTE (py3 only): diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py index 7362c48cf..31e3cc59b 100644 --- a/dvc/stage/cache.py +++ b/dvc/stage/cache.py @@ -2,7 +2,7 @@ import logging import os import tempfile from contextlib import contextmanager -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, List, Optional, Tuple from funcy import first @@ -236,7 +236,7 @@ class StageCache: message = f"run-cache is not supported for http filesystem: {path}" raise RunCacheNotSupported(message) - ret = [] + ret: List[Tuple[str, str]] = [] if not from_fs.exists(runs): return ret diff --git a/dvc/stage/loader.py b/dvc/stage/loader.py index 0ea7ce694..e71e5779b 100644 --- a/dvc/stage/loader.py +++ b/dvc/stage/loader.py @@ -200,7 +200,10 @@ class SingleStageLoader(Mapping): @classmethod def load_stage( - cls, dvcfile: "SingleStageFile", d: Dict[str, Any], stage_text: str + cls, + dvcfile: "SingleStageFile", + d: Dict[str, Any], + stage_text: Optional[str], ) -> Stage: path, wdir = resolve_paths( dvcfile.repo.fs, dvcfile.path, d.get(Stage.PARAM_WDIR) diff --git a/dvc/stage/serialize.py b/dvc/stage/serialize.py index b37eb4193..686b302d0 100644 --- a/dvc/stage/serialize.py +++ b/dvc/stage/serialize.py @@ -1,7 +1,14 @@ from collections import OrderedDict -from functools import partial from operator import attrgetter -from typing import TYPE_CHECKING, List, no_type_check +from typing import ( + TYPE_CHECKING, + Dict, + Iterable, + List, + Optional, + Union, + no_type_check, +) from funcy import post_processing @@ -34,9 +41,6 @@ PARAM_PUSH = Output.PARAM_PUSH DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE -sort_by_path = partial(sorted, key=attrgetter("def_path")) - - @post_processing(OrderedDict) def _get_flags(out): annot = out.annot.to_dict() @@ -67,7 +71,7 @@ def _serialize_out(out): @no_type_check def _serialize_outs(outputs: List[Output]): outs, metrics, plots = [], [], [] - for out in sort_by_path(outputs): + for out in sorted(outputs, key=attrgetter("def_path")): bucket = outs if out.plot: bucket = plots @@ -77,7 +81,7 @@ def _serialize_outs(outputs: List[Output]): return outs, metrics, plots -def _serialize_params_keys(params): +def _serialize_params_keys(params: Iterable["ParamsDependency"]): """ Returns the following format of data: ['lr', 'train', {'params2.yaml': ['lr']}] @@ -86,12 +90,12 @@ def _serialize_params_keys(params): at the first, and then followed by entry of other files in lexicographic order. The keys of those custom files are also sorted in the same order. """ - keys = [] - for param_dep in sort_by_path(params): + keys: List[Union[str, Dict[str, Optional[List[str]]]]] = [] + for param_dep in sorted(params, key=attrgetter("def_path")): # when on no_exec, params are not filled and are saved as list - k = sorted(param_dep.params) + k: List[str] = sorted(param_dep.params) if k and param_dep.def_path == DEFAULT_PARAMS_FILE: - keys = k + keys + keys = k + keys # type: ignore[operator,assignment] else: keys.append({param_dep.def_path: k or None}) return keys @@ -106,7 +110,7 @@ def _serialize_params_values(params: List[ParamsDependency]): alphabetical order. The param values are sorted too(not recursively though) """ key_vals = OrderedDict() - for param_dep in sort_by_path(params): + for param_dep in sorted(params, key=attrgetter("def_path")): dump = param_dep.dumpd() path, params = dump[PARAM_PATH], dump[PARAM_PARAMS] if isinstance(params, dict): @@ -119,9 +123,9 @@ def _serialize_params_values(params: List[ParamsDependency]): def to_pipeline_file(stage: "PipelineStage"): wdir = resolve_wdir(stage.wdir, stage.path) - params, deps = split_params_deps(stage) - deps = sorted(d.def_path for d in deps) - params = _serialize_params_keys(params) + param_objs, deps_objs = split_params_deps(stage) + deps = sorted(d.def_path for d in deps_objs) + params = _serialize_params_keys(param_objs) outs, metrics, plots = _serialize_outs(stage.outs) @@ -172,7 +176,7 @@ def to_single_stage_lockfile(stage: "Stage", **kwargs) -> dict: res = OrderedDict([("cmd", stage.cmd)]) params, deps = split_params_deps(stage) deps, outs = ( - [_dumpd(item) for item in sort_by_path(items)] + [_dumpd(item) for item in sorted(items, key=attrgetter("def_path"))] for items in [deps, stage.outs] ) params = _serialize_params_values(params) diff --git a/dvc/testing/fixtures.py b/dvc/testing/fixtures.py index ee6b00344..d2cce3489 100644 --- a/dvc/testing/fixtures.py +++ b/dvc/testing/fixtures.py @@ -46,6 +46,8 @@ def make_tmp_dir(tmp_path_factory, request, worker_id): cache_dir = tmp_path_factory.mktemp("dvc-test-cache" + worker_id) TmpDir(cache_dir).init(scm=scm, dvc=dvc, subdir=subdir) CACHE[(scm, dvc, subdir)] = cache = os.fspath(cache_dir) + + assert cache path = tmp_path_factory.mktemp(name) if isinstance(name, str) else name # ignore sqlite files from .dvc/tmp. We might not be closing the cache diff --git a/dvc/testing/path_info.py b/dvc/testing/path_info.py index 93f698a73..d3a4291dc 100644 --- a/dvc/testing/path_info.py +++ b/dvc/testing/path_info.py @@ -12,13 +12,14 @@ from dvc.utils.objects import cached_property class _BasePath: def overlaps(self, other): if isinstance(other, (str, bytes)): - other = self.__class__(other) + other = self.__class__(other) # type: ignore[call-arg] elif self.__class__ != other.__class__: return False return self.isin_or_eq(other) or other.isin(self) def isin_or_eq(self, other): - return self == other or self.isin(other) # pylint: disable=no-member + # pylint: disable-next=no-member + return self == other or self.isin(other) # type: ignore[attr-defined] class PathInfo(pathlib.PurePath, _BasePath): @@ -35,10 +36,11 @@ class PathInfo(pathlib.PurePath, _BasePath): WindowsPathInfo if os.name == "nt" else PosixPathInfo ) - return cls._from_parts(args) + return cls._from_parts(args) # type: ignore[attr-defined] def as_posix(self): - f = self._flavour # pylint: disable=no-member + # pylint: disable-next=no-member + f = self._flavour # type: ignore[attr-defined] # Unlike original implementation [1] that uses `str()` we actually need # to use `fspath`, because we've overridden `__str__` method to return # relative paths, which will break original `as_posix`. @@ -75,7 +77,10 @@ class PathInfo(pathlib.PurePath, _BasePath): return False # Use cached casefolded parts to compare paths n = len(other._cparts) - return len(self._cparts) > n and self._cparts[:n] == other._cparts + return ( + len(self._cparts) > n # type: ignore[attr-defined] + and self._cparts[:n] == other._cparts # type: ignore[attr-defined] + ) def relative_to(self, other): # pylint: disable=arguments-differ # pathlib relative_to raises exception when one path is not a direct @@ -153,7 +158,7 @@ class URLInfo(_BasePath): self._path = path else: if path and path[0] != "/": - path = "/" + path + path = "/" + path # type: ignore[operator] self._spath = path @property @@ -165,7 +170,9 @@ class URLInfo(_BasePath): return self._base_parts + self._path.parts def replace(self, path=None): - return self.from_parts(*self._base_parts, path=path) + return self.from_parts( # type: ignore[misc] + *self._base_parts, path=path + ) @cached_property def url(self) -> str: @@ -271,7 +278,7 @@ class HTTPURLInfo(URLInfo): self.fragment = p.fragment def replace(self, path=None): - return self.from_parts( + return self.from_parts( # type: ignore[misc] *self._base_parts, params=self.params, query=self.query, diff --git a/dvc/testing/tmp_dir.py b/dvc/testing/tmp_dir.py index eca2a19e0..490c5abe3 100644 --- a/dvc/testing/tmp_dir.py +++ b/dvc/testing/tmp_dir.py @@ -76,9 +76,8 @@ class TmpDir(pathlib.Path): ) # init parameter and `_init` method has been removed in Python 3.10. kw = {"init": False} if sys.version_info < (3, 10) else {} - self = cls._from_parts( # pylint: disable=unexpected-keyword-arg - args, **kw - ) + # pylint: disable-next=unexpected-keyword-arg + self = cls._from_parts(args, **kw) # type: ignore[attr-defined] if not self._flavour.is_supported: raise NotImplementedError( f"cannot instantiate {cls.__name__!r} on your system" @@ -236,7 +235,8 @@ class TmpDir(pathlib.Path): path.name: path.read_text(*args, **kwargs) for path in self.iterdir() } - return super().read_text(*args, encoding="utf-8", **kwargs) + kwargs.setdefault("encoding", "utf-8") # type: ignore[call-overload] + return super().read_text(*args, **kwargs) def oid_to_path(self, hash_): return str(self / hash_[0:2] / hash_[2:]) diff --git a/dvc/testing/workspace_tests.py b/dvc/testing/workspace_tests.py index ac81074ff..a16b3b7c7 100644 --- a/dvc/testing/workspace_tests.py +++ b/dvc/testing/workspace_tests.py @@ -1,4 +1,5 @@ import os +from typing import Dict, Union import pytest @@ -74,7 +75,7 @@ class TestImport: # of directories. So instead we create an empty file that ends with a # trailing slash in order to actually support this operation if is_object_storage: - contents = "" + contents: Union[str, Dict[str, str]] = "" else: contents = {} diff --git a/dvc/utils/collections.py b/dvc/utils/collections.py index f22af3b55..23350813f 100644 --- a/dvc/utils/collections.py +++ b/dvc/utils/collections.py @@ -1,9 +1,18 @@ import inspect from collections.abc import Mapping from functools import wraps -from typing import Callable, Dict, Iterable, List, TypeVar, Union +from typing import ( + Callable, + Dict, + Iterable, + List, + TypeVar, + Union, + no_type_check, +) +@no_type_check def apply_diff(src, dest): # noqa: C901 """Recursively apply changes from src to dest. @@ -155,7 +164,9 @@ def validate(*validators: Callable, post: bool = False): def inner(*args, **kwargs): ba = sig.bind(*args, **kwargs) ba.apply_defaults() - ba.arguments = _NamespacedDict(ba.arguments) + ba.arguments = _NamespacedDict( # type: ignore[assignment] + ba.arguments + ) if not post: for validator in validators: diff --git a/dvc/utils/diff.py b/dvc/utils/diff.py index 48f246a5c..269bf31a7 100644 --- a/dvc/utils/diff.py +++ b/dvc/utils/diff.py @@ -1,5 +1,6 @@ import json from collections import defaultdict +from typing import Dict from .flatten import flatten @@ -46,7 +47,7 @@ def _diff_dicts(old_dict, new_dict, with_unchanged): new = _flatten(new_dict) old = _flatten(old_dict) - res = defaultdict(dict) + res: Dict[str, Dict] = defaultdict(dict) xpaths = set(old.keys()) xpaths.update(set(new.keys())) @@ -77,7 +78,7 @@ def diff(old, new, with_unchanged=False): paths = set(old.keys()) paths.update(set(new.keys())) - res = defaultdict(dict) + res: Dict[str, Dict] = defaultdict(dict) for path in paths: path_diff = _diff( old.get(path, {}).get("data", {}), diff --git a/dvc/utils/serialize/_common.py b/dvc/utils/serialize/_common.py index 349d74593..e955c3bb0 100644 --- a/dvc/utils/serialize/_common.py +++ b/dvc/utils/serialize/_common.py @@ -9,6 +9,7 @@ from typing import ( Dict, Optional, Protocol, + TextIO, Union, ) @@ -28,6 +29,11 @@ class DumperFn(Protocol): ... +class DumpersFn(Protocol): + def __call__(self, data: Any, stream: TextIO) -> Any: + ... + + class ModifierFn(Protocol): def __call__( self, path: "StrPath", fs: Optional["FileSystem"] = None @@ -78,7 +84,7 @@ def _load_data( def _dump_data( path, data: Any, - dumper: DumperFn, + dumper: DumpersFn, fs: Optional["FileSystem"] = None, **dumper_args, ): @@ -91,11 +97,11 @@ def _dump_data( def _modify_data( path: "StrPath", parser: ParserFn, - dumper: DumperFn, + dumper: DumpersFn, fs: Optional["FileSystem"] = None, ): exists_fn = fs.exists if fs else os.path.exists file_exists = exists_fn(path) data = _load_data(path, parser=parser, fs=fs) if file_exists else {} yield data - dumper(path, data, fs=fs) + _dump_data(path, data, dumper=dumper, fs=fs) diff --git a/dvc/utils/serialize/_json.py b/dvc/utils/serialize/_json.py index ee2a9ce49..c6e9ac231 100644 --- a/dvc/utils/serialize/_json.py +++ b/dvc/utils/serialize/_json.py @@ -19,13 +19,17 @@ def parse_json(text, path, **kwargs): return json.loads(text, **kwargs) or {} +def _dump_json(data, stream, **kwargs): + return json.dump(data, stream, **kwargs) + + def dump_json(path, data, fs=None, **kwargs): - return _dump_data(path, data, dumper=json.dump, fs=fs, **kwargs) + return _dump_data(path, data, dumper=_dump_json, fs=fs, **kwargs) @contextmanager def modify_json(path, fs=None): - with _modify_data(path, parse_json, dump_json, fs=fs) as d: + with _modify_data(path, parse_json, _dump_json, fs=fs) as d: yield d diff --git a/dvc/utils/serialize/_py.py b/dvc/utils/serialize/_py.py index ea6d9f1ca..945410d8e 100644 --- a/dvc/utils/serialize/_py.py +++ b/dvc/utils/serialize/_py.py @@ -1,5 +1,6 @@ import ast from contextlib import contextmanager +from typing import Any from funcy import reraise @@ -81,7 +82,7 @@ def dump_py(path, data, fs=None): @contextmanager def modify_py(path, fs=None): - with _modify_data(path, parse_py_for_update, dump_py, fs=fs) as d: + with _modify_data(path, parse_py_for_update, _dump, fs=fs) as d: yield d @@ -129,16 +130,18 @@ def _ast_assign_to_dict(assign, only_self_params=False, lineno=False): else: raise AttributeError + value: Any if isinstance(assign.value, ast.Dict): value = {} for key, val in zip(assign.value.keys, assign.value.values): if lineno: - value[ast.literal_eval(key)] = { + value[ast.literal_eval(key)] = { # type: ignore[arg-type] "lineno": assign.lineno - 1, "value": ast.literal_eval(val), } else: - value[ast.literal_eval(key)] = ast.literal_eval(val) + v = ast.literal_eval(val) + value[ast.literal_eval(key)] = v # type: ignore[arg-type] elif isinstance(assign.value, ast.List): value = [ast.literal_eval(val) for val in assign.value.elts] elif isinstance(assign.value, ast.Set): diff --git a/dvc/utils/serialize/_toml.py b/dvc/utils/serialize/_toml.py index 31313896a..1d21ae4ff 100644 --- a/dvc/utils/serialize/_toml.py +++ b/dvc/utils/serialize/_toml.py @@ -47,5 +47,5 @@ def dump_toml(path, data, fs=None, **kwargs): @contextmanager def modify_toml(path, fs=None): - with _modify_data(path, parse_toml_for_update, dump_toml, fs=fs) as d: + with _modify_data(path, parse_toml_for_update, _dump, fs=fs) as d: yield d diff --git a/dvc/utils/serialize/_yaml.py b/dvc/utils/serialize/_yaml.py index 12e0b70a4..b8c4a55e0 100644 --- a/dvc/utils/serialize/_yaml.py +++ b/dvc/utils/serialize/_yaml.py @@ -1,6 +1,7 @@ import io from collections import OrderedDict from contextlib import contextmanager +from typing import Any, TextIO from funcy import reraise @@ -53,7 +54,7 @@ def _get_yaml(): return yaml -def _dump(data, stream): +def _dump(data: Any, stream: TextIO) -> Any: yaml = _get_yaml() return yaml.dump(data, stream) @@ -76,5 +77,5 @@ def dumps_yaml(d): @contextmanager def modify_yaml(path, fs=None): - with _modify_data(path, parse_yaml_for_update, dump_yaml, fs=fs) as d: + with _modify_data(path, parse_yaml_for_update, _dump, fs=fs) as d: yield d diff --git a/dvc/utils/threadpool.py b/dvc/utils/threadpool.py index 97de574d8..e621c05a5 100644 --- a/dvc/utils/threadpool.py +++ b/dvc/utils/threadpool.py @@ -45,29 +45,29 @@ class ThreadPoolExecutor(futures.ThreadPoolExecutor): tasks.update(create_taskset(len(done))) def shutdown(self, wait=True, *, cancel_futures=False): - if sys.version_info > (3, 9): + 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: + with self._shutdown_lock: + self._shutdown = True + if cancel_futures: + # Drain all work items from the queue, and then cancel + # their associated futures. + while True: + try: + work_item = self._work_queue.get_nowait() + except queue.Empty: + break + if work_item is not None: + work_item.future.cancel() - with self._shutdown_lock: - self._shutdown = True - if cancel_futures: - # Drain all work items from the queue, and then cancel their - # associated futures. - while True: - try: - work_item = self._work_queue.get_nowait() - except queue.Empty: - break - if work_item is not None: - work_item.future.cancel() - - # Send a wake-up to prevent threads calling - # _work_queue.get(block=True) from permanently blocking. - self._work_queue.put(None) - if wait: - for t in self._threads: - t.join() + # Send a wake-up to prevent threads calling + # _work_queue.get(block=True) from permanently blocking. + self._work_queue.put(None) # type: ignore[arg-type] + if wait: + for t in self._threads: + t.join() def __exit__(self, exc_type, exc_val, exc_tb): if self._cancel_on_error: diff --git a/pyproject.toml b/pyproject.toml index 873e862ce..567ff1ed7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -226,7 +226,7 @@ show_error_context = true show_traceback = true pretty = true no_implicit_optional = true -check_untyped_defs = false +check_untyped_defs = true strict_equality = true strict_concatenate = true # Warnings