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

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

Open

    def process_symlink(self, *, path, parent_fd, name, st, strip_prefix):

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

Open

def valid_msgpacked_dict(d, keys_serialized):
    """check if the data <d> looks like a msgpacked dict"""
    d_len = len(d)
    if d_len == 0:
        return False

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if pi:
                            pi.show(increase=len(data), info=[remove_surrogates(item.path)])
                        with backup_io("write"):

Avoid deeply nested control flow statements.

Open

                        with backup_io("write"):
                            if sparse and zeros.startswith(data):
                                # all-zero chunk: create a hole in a sparse file
                                fd.seek(len(data), 1)
                            else:

Avoid deeply nested control flow statements.

Open

                        if not is_special_file and not changed_while_backup:
                            # we must not memorize special files, because the contents of e.g. a
                            # block or char device will change without its mtime/size/inode changing.
                            # also, we must not memorize a potentially inconsistent/corrupt file that
                            # changed while we backed it up.

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

Avoid deeply nested control flow statements.

Open

                        with backup_io("read"):
                            self.process_file_chunks(
                                item,
                                cache,
                                self.stats,

Avoid deeply nested control flow statements.

Open

                        if not self.shadow_index[key]:
                            # shadowed segments list is empty -> remove it
                            del self.shadow_index[key]
            assert segments[segment] == 0, "Corrupted segment reference count - corrupted index or hints"

Avoid deeply nested control flow statements.

Open

                            if tag == TAG_PUT2:
                                c_offset = self.HEADER_ID_SIZE + self.ENTRY_HASH_SIZE
                                # skip if header is invalid
                                if crc32(d[4:c_offset]) & 0xFFFFFFFF != crc:
                                    d = d[1:]

Avoid deeply nested control flow statements.

Open

                            for archive_info in manifest.archives.list(sort_by=["ts"]):
                                msg.append(format_archive(archive_info))
                        else:

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

Open

def cache_if_remote(repository, *, decrypted_cache=False, pack=None, unpack=None, transform=None, force_cache=False):
    """
    Return a Repository(No)Cache for *repository*.

    If *decrypted_cache* is a repo_objs object, then get and get_many will return a tuple

• Read up
  Read up

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

Open

    def process_dir(self, *, path, parent_fd, name, st, strip_prefix):

Avoid deeply nested control flow statements.

Open

                        if "size" in item:
                            item_size = item.size
                            if item_size != item_chunks_size:
                                raise BackupError(
                                    "Size inconsistency detected: size {}, chunks size {}".format(