Avoid too many return statements within this function.

Open

        return self.fs.relpath(self.fs_path, self.repo.root_dir)

Avoid too many return statements within this function.

Open

                return 0

Avoid too many return statements within this function.

Open

            return 0

Avoid too many return statements within this function.

Open

        return _val

Function wait has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def wait(self, revs: Collection[str], **kwargs) -> None:
        """Block until the specified tasks have completed."""
        revs = [revs] if isinstance(revs, str) else revs
        results = self.match_queue_entry_by_name(
            revs, self.iter_queued(), self.iter_done(), self.iter_failed()

• Read up
  Read up

Function shutdown has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def shutdown(self, kill: bool = False):
        self.celery.control.shutdown()
        if kill:
            to_kill: dict[QueueEntry, str] = {}
            for entry in self.iter_active():

• Read up
  Read up

Function changed_outs has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def changed_outs(self, allow_missing: bool = False) -> bool:
        for out in self.outs:
            status = out.status()
            if status:
                if allow_missing and status[str(out)] in ["not in cache", "deleted"]:

• Read up
  Read up

Function celery_remove has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def celery_remove(self: "LocalCeleryQueue", revs: Collection[str]) -> list[str]:
    """Remove the specified entries from the queue.

    Arguments:
        revs: Stash revisions or queued exp names to be removed.

• Read up
  Read up

Function write_json has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def write_json(  # noqa: PLR0913
        self,
        data: Any,
        indent: Optional[int] = None,
        highlight: Optional[bool] = None,

• Read up
  Read up

Function prepare_default_pager has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def prepare_default_pager(
    clear_screen: bool = False,
    quit_if_one_screen: bool = True,
    ansi_escapes: bool = True,
    chop_long_lines: bool = True,

• Read up
  Read up

Function table has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def table(
        self,
        data: "TableData",
        headers: Optional["Headers"] = None,
        markdown: bool = False,

• Read up
  Read up

Function init_git has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def init_git(
        self,
        repo: "Repo",
        scm: "Git",
        stash_rev: str,

• Read up
  Read up

Function merge has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def merge(self, ancestor, other, allowed=None) -> None:
        assert other

        if not other.outs:
            return

• Read up
  Read up

Function _get_flags has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def _get_flags(out):
    annot = out.annot.to_dict()
    yield from annot.items()

    if not out.use_cache:

• Read up
  Read up

Function _try_to_kill_tasks has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _try_to_kill_tasks(
        self, to_kill: dict[QueueEntry, str], force: bool
    ) -> dict[QueueEntry, str]:
        fail_to_kill_entries: dict[QueueEntry, str] = {}
        for queue_entry, rev in to_kill.items():

• Read up
  Read up

Function kill has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def kill(self, revs: Collection[str], force: bool = False) -> None:
        name_dict: dict[str, Optional[QueueEntry]] = self.match_queue_entry_by_name(
            set(revs), self.iter_active()
        )

• Read up
  Read up

Function save_deps has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def save_deps(self, allow_missing=False):
        from dvc.dependency.base import DependencyDoesNotExistError

        for dep in self.deps:
            try:

• Read up
  Read up

Function _infer_y_from_data has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _infer_y_from_data(self):
        if self.plot_id in self.data:
            for lst in _lists(self.data[self.plot_id]):
                if all(isinstance(item, dict) for item in lst):
                    datapoint = first(lst)

• Read up
  Read up

Function collect_failed_data has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def collect_failed_data(
        self,
        baseline_revs: Optional[Collection[str]],
        **kwargs,
    ) -> dict[str, list["ExpRange"]]:

• Read up
  Read up

Function get_definition_data has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def get_definition_data(self, target_files, rev):
        result = {}
        for definition_file in target_files:
            if os.name == "nt":
                source_file = _normpath(definition_file).replace("\\", "/")

• Read up
  Read up