File celery.py has 531 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import hashlib
import locale
import logging
import os
from collections import defaultdict

LocalCeleryQueue has 39 functions (exceeds 20 allowed). Consider refactoring.

Open

class LocalCeleryQueue(BaseStashQueue):
    """DVC experiment queue.

    Maps queued experiments to (Git) stash reflog entries.
    """

Function _get_done_result has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_done_result(
        self, entry: QueueEntry, timeout: Optional[float] = None
    ) -> Optional["ExecutorResult"]:
        from celery.exceptions import TimeoutError as _CeleryTimeout

• Read up
  Read up

Function _mark_inactive_tasks_failure has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _mark_inactive_tasks_failure(
        self, remained_entries: dict[QueueEntry, str]
    ) -> None:
        remained_revs: list[str] = []
        running_ids = self._get_running_task_ids()

• Read up
  Read up

Function collect_active_data has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def collect_active_data(
        self,
        baseline_revs: Optional[Collection[str]],
        fetch_refs: bool = False,
        **kwargs,

• Read up
  Read up

Function iter_done has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def iter_done(self) -> Generator[QueueDoneResult, None, None]:
        for result, entry in self._iter_done_tasks():
            try:
                exp_result = self.get_result(entry)
            except FileNotFoundError:

• Read up
  Read up

Function collect_success_executors has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def collect_success_executors(
        self,
        baseline_revs: Optional[Collection[str]],
        **kwargs,
    ) -> dict[str, "ExpExecutor"]:

• Read up
  Read up

Function logs has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def logs(self, rev: str, encoding: Optional[str] = None, follow: bool = False):
        queue_entry: Optional[QueueEntry] = self.match_queue_entry_by_name(
            {rev}, self.iter_active(), self.iter_done()
        ).get(rev)
        if queue_entry is None:

• 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 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 _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 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 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