Avoid deeply nested control flow statements.

Open

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

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

Open

    def save(self, name=None, comment=None, timestamp=None, stats=None, additional_metadata=None):

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

Open

    def extract_helper(self, item, path, hlm, *, dry_run=False):

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 save has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def save(self, name=None, comment=None, timestamp=None, stats=None, additional_metadata=None):
        name = name or self.name
        self.items_buffer.flush(flush=True)
        item_ptrs = archive_put_items(
            self.items_buffer.chunks, repo_objs=self.repo_objs, cache=self.cache, stats=self.stats

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                            if size_invalid or tag > MAX_TAG_ID:
                                d = d[1:]
                                continue
                            if tag == TAG_PUT2:

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

Open

    def do_benchmark_cpu(self, args):
        """Benchmark CPU bound operations."""
        from timeit import timeit

        random_10M = os.urandom(10 * 1000 * 1000)

• Read up
  Read up

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

Open

    def save(self, archive, target, comment=None, delete_original=True):
        if self.dry_run:
            return
        if comment is None:
            comment = archive.metadata.get("comment", "")

• Read up
  Read up

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

Open

def pack(o, stream, *, use_bin_type=USE_BIN_TYPE, unicode_errors=UNICODE_ERRORS, **kwargs):

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

Open

def unpackb(packed, *, raw=RAW, unicode_errors=UNICODE_ERRORS, strict_map_key=False, **kwargs):

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

Open

def archive_put_items(chunk_ids, *, repo_objs, cache=None, stats=None, add_reference=None):

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

Open

def process(input):
    unpacker = RobustUnpacker(validator=_validator, item_keys=ITEM_KEYS)
    result = []
    for should_sync, chunks in input:
        if should_sync:

• Read up
  Read up

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

Open

    def do_repo_list(self, args, repository, manifest):
        """List the archives contained in a repository"""
        if args.format is not None:
            format = args.format
        elif args.short:

• Read up
  Read up

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

Open

    def analyze_change(self, base, new):
        """for each directory path, sum up the changed (removed or added) chunks' sizes between base and new."""

        def analyze_path_change(path):
            base_chunks = base[path]

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if not dry_run:
                            item.chunks = chunks  # TODO: overwrite? IDs and sizes are same.
                            archive.stats.nfiles += 1
                    if not dry_run:

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

Open

def get_repository(
    location, *, create, exclusive, lock_wait, lock, append_only, make_parent_dirs, storage_quota, args, v1_or_v2
):
    if location.proto in ("ssh", "socket"):
        RemoteRepoCls = LegacyRemoteRepository if v1_or_v2 else RemoteRepository

• Read up
  Read up

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

Open

def define_archive_filters_group(subparser, *, sort_by=True, first_last=True, oldest_newest=True, older_newer=True):

Avoid deeply nested control flow statements.

Open

                    if not is_win32:
                        # The S_IFBLK macro is broken on MINGW
                        assert stat.S_ISBLK(item.mode)
                    # looks like we can't use os.major/minor with data coming from another platform,
                    # thus we only do a rather rough check here:

Avoid deeply nested control flow statements.

Open

                            if not recurse_excluded_dir:
                                if keep_exclude_tags:
                                    if not dry_run:
                                        fso.process_dir_with_fd(
                                            path=path, fd=child_fd, st=st, strip_prefix=strip_prefix