Function collect_granular has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def collect_granular(
        self,
        target: Optional[str] = None,
        with_deps: bool = False,
        recursive: bool = False,

• Read up
  Read up

Function parse_target has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

def parse_target(
    target: str, default: Optional[str] = None, isa_glob: bool = False
) -> tuple[Optional[str], Optional[str]]:
    from dvc.dvcfile import LOCK_FILE, PROJECT_FILE, is_valid_filename
    from dvc.exceptions import DvcException

• Read up
  Read up

Function run has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def run(self):
        from dvc.exceptions import DvcException

        try:
            diff = self.repo.diff(self.args.a_rev, self.args.b_rev, self.args.targets)

• Read up
  Read up

Function __init__ has 15 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(  # noqa: PLR0913

Function test has 46 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def test(self, tmp_dir, dvc, remote):
        (stage,) = tmp_dir.dvc_gen("foo", "foo")
        out = stage.outs[0]
        cache = out.cache_path
        foo_hash = out.hash_info

Function get_remote has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def get_remote(
        self,
        name: Optional[str] = None,
        command: str = "<command>",
    ) -> "Remote":

• Read up
  Read up

Function _infer_x_y has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def _infer_x_y(self):
        x = self.properties.get("x", None)
        y = self.properties.get("y", None)

        inferred_properties: dict = {}

• Read up
  Read up

Function fetch_running_exp_from_temp_dir has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def fetch_running_exp_from_temp_dir(
    queue: "BaseStashQueue", rev: str, fetch_refs: bool
) -> dict[str, dict]:
    """Fetch status of running exps out of current working directory

• Read up
  Read up

Function _get_repo_dirs has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_repo_dirs(
        self,
        root_dir: Optional[str] = None,
        fs: Optional["FileSystem"] = None,
        uninitialized: bool = False,

• Read up
  Read up

Function build_outs_trie has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def build_outs_trie(stages):
    outs = Trie()

    for stage in stages:
        for out in stage.outs:

• Read up
  Read up

Function _exp_range_rows has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _exp_range_rows(
    exp_range: "ExpRange",
    *,
    all_headers: Iterable[str],
    fill_value: Optional[str],

• Read up
  Read up

Function _collect_names has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _collect_names(exp_states: Iterable["ExpState"]) -> _DataNames:
    result = _DataNames(defaultdict(dict), defaultdict(dict), set())

    def _collect_d(result_d: dict[str, dict[str, Any]], data_d: dict[str, Any]):
        for path, item in data_d.items():

• Read up
  Read up

Function iter_revs has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def iter_revs(
    scm: "Git",
    revs: Optional[list[str]] = None,
    num: int = 1,
    all_branches: bool = False,

• Read up
  Read up

Function make_dvc_bin has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def make_dvc_bin(
    dvc_rev,
    dvc_venvs,
    make_dvc_venv,
    dvc_repo,

• Read up
  Read up

Function _postprocess has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _postprocess(results):
    processed: dict[str, dict] = {}
    for rev, rev_data in results.items():
        if not rev_data:
            continue

• Read up
  Read up

Function _show_markdown has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _show_markdown(diff, show_hash=False, hide_missing=False):
    headers = ["Status", "Hash", "Path"] if show_hash else ["Status", "Path"]
    rows = []
    statuses = ["added", "deleted", "renamed", "modified"]
    if not hide_missing:

• Read up
  Read up

Function branch has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def branch(repo, exp_rev, branch_name=None, **kwargs):
    from dvc.scm import resolve_rev

    try:
        rev = resolve_rev(repo.scm, exp_rev)

• Read up
  Read up

Function _repro_dvc has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def _repro_dvc(
        cls,
        info: "ExecutorInfo",
        infofile: Optional[str] = None,
        log_errors: bool = True,

• Read up
  Read up

Function init has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):  # noqa: C901
    """
    Creates an empty repo on the given directory -- basically a
    `.dvc` directory with subdirectories for configuration and cache.

• Read up
  Read up

Function _collect_vars has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def _collect_vars(repo, params, stages=None) -> dict:
    vars_params: dict[str, dict] = defaultdict(dict)

    for stage in repo.index.stages:
        if isinstance(stage, PipelineStage) and stage.tracked_vars:

• Read up
  Read up