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):

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

Open

    def format_owner(self, diff: "ItemDiff", spec: Literal["owner", "user", "group"] = "owner"):
        if spec == "user":
            change = diff.user()
            return f"[{change.diff_data['item1']} -> {change.diff_data['item2']}]" if change else ""
        if spec == "group":

• Read up
  Read up

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

Open

    def get_item_data(self, item: "ItemDiff", jsonline=False) -> dict:
        diff_data = {}
        for key in self.used_call_keys:
            diff_data[key] = self.call_keys[key](item)

• Read up
  Read up

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

Open

def safe_unlink(path):
    """
    Safely unlink (delete) *path*.

    If we run out of space while deleting the file, we try truncating it first.

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    while True:
                        char = text.peek(1)
                        if char == "`":
                            text.read(1)
                            break

Avoid deeply nested control flow statements.

Open

                    if directive == "nanorst: inline-fill":
                        inline_mode = "fill"
                    elif directive == "nanorst: inline-replace":
                        inline_mode = "replace"
                    continue

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

Open

def unpack(stream, *, raw=RAW, unicode_errors=UNICODE_ERRORS, strict_map_key=False, **kwargs):

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

Open

    def process_file_chunks(self, item, cache, stats, show_progress, chunk_iter, chunk_processor=None):