Function dir_is_tagged has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def dir_is_tagged(path, exclude_caches, exclude_if_present):
    """Determines whether the specified path is excluded by being a cache
    directory or containing user-specified tag files/directories. Returns a
    list of the names of the tag files/directories (either CACHEDIR.TAG or the
    matching user-specified files/directories).

• Read up
  Read up

Function inject_exception has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def inject_exception(self, kind):
        s1 = "test string"
        s2 = "test string2"
        if kind == "DoesNotExist":
            raise self.RepoCls.DoesNotExist(s1)

• Read up
  Read up

Function _mkstemp_inner has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def _mkstemp_inner(dir, pre, suf, flags, output_type, mode=0o600):

Function ids has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def ids(self):
        # yield the binary IDs of all archives
        if not self.legacy:
            try:
                infos = list(self.repository.store_list("archives"))

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if lr_dict is not None:
                            # Re-emit remote log messages locally.
                            _logger = logging.getLogger(lr_dict["name"])
                            if _logger.isEnabledFor(lr_dict["level"]):
                                _logger.handle(logging.LogRecord(**lr_dict))

Avoid deeply nested control flow statements.

Open

                        if is_preloaded:
                            assert cmd == "get", "is_preload is only supported for 'get'"
                            if calls[0]["id"] in self.chunkid_to_msgids:
                                waiting_for.append(pop_preload_msgid(calls.pop(0)["id"]))
                        else:

Function load has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def load(cls, repository, operations, key=None, *, ro_cls=RepoObj):

Function _log_warning has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def _log_warning(message, category, filename, lineno, file=None, line=None):

Avoid deeply nested control flow statements.

Open

                        while time.monotonic() - started < self.timeout:
                            locks = self._find_locks(only_exclusive=False)
                            if len(locks) == 1 and locks[0]["key"] == key:
                                logger.debug("LOCK-ACQUIRE: success! no non-exclusive locks are left!")
                                return self

Avoid deeply nested control flow statements.

Open

                        if not isinstance(unpacked, dict):
                            raise UnexpectedRPCDataFormatFromServer(data)

Function export_paperkey has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def export_paperkey(self, path):
        if path is None:
            path = "-"

        def grouped(s):

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if option.metavar:
                            option_fmt = "``%s " + option.metavar + "``"
                        else:
                            option_fmt = "``%s``"
                        option_str = ", ".join(option_fmt % s for s in option.option_strings)

Function call_many has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def call_many(self, cmd, calls, wait=True, is_preloaded=False, async_wait=True):

Function exists_name_and_ts has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def exists_name_and_ts(self, name, ts):
        # check if an archive with this name AND timestamp exists
        assert isinstance(name, str)
        assert isinstance(ts, datetime)
        if not self.legacy:

• Read up
  Read up

Function _lookup_name has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def _lookup_name(self, name, raw=False):
        assert isinstance(name, str)
        assert not self.legacy
        for archive_info in self._infos():
            if archive_info["exists"] and archive_info["name"] == name:

• Read up
  Read up

Function issue has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

        def issue(name, rawtext, text, lineno, inliner, options={}, content=[]):

Avoid deeply nested control flow statements.

Open

                        if msgid in self.ignore_responses:
                            self.ignore_responses.remove(msgid)
                            # async methods never return values, but may raise exceptions.
                            if "exception_class" in unpacked:
                                self.async_responses[msgid] = unpacked

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

Open

    def __init__(self, store, exclusive=False, sleep=None, timeout=1.0, stale=30 * 60, id=None):

Avoid deeply nested control flow statements.

Open

                            if self.repository is not None:
                                self.repository.close()
                            raise UnexpectedRPCDataFormatFromClient(__version__)

Function os_open has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def os_open(*, flags, path=None, parent_fd=None, name=None, noatime=False):