Function test_chunkify_sparse has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def test_chunkify_sparse(tmpdir, fname, sparse_map, header_size, sparse):

Function do_rename has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def do_rename(self, args, repository, manifest, cache, archive):

Function set_acl has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def set_acl(path, access=None, default=None, nfs4=None, numeric_ids=False):

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

Open

    def __init__(self, path, exclusive=False, sleep=None, timeout=None, id=None):

Function pop_option has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

        def pop_option(options, key, present, not_present, wanted_type, int_base=0):

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

Open

def get_cache_dir(*, legacy=False, create=True):
    """Determine where to repository keys and cache"""

    if legacy:
        # Get cache home path

• Read up
  Read up

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

Open

    def __init__(
        self,
        path_or_location,
        create=False,
        exclusive=False,

• Read up
  Read up

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

Open

def format_line(format, data):
    for _, key, _, conversion in Formatter().parse(format):
        if not key:
            continue
        if conversion or key not in data:

• Read up
  Read up

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

Open

    def format_content(self, diff: "ItemDiff"):
        change = diff.content()
        if change:
            if change.diff_type == "added":
                return "{}: {:>20}".format(change.diff_type, format_file_size(change.diff_data["added"]))

• Read up
  Read up

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

Open

def repo_lister(repository, *, limit=None):
    marker = None
    finished = False
    while not finished:
        result = repository.list(limit=limit, marker=marker)

• Read up
  Read up

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

Open

    def validator(text):
        assert isinstance(text, str)
        if len(text) < min_length:
            raise argparse.ArgumentTypeError(f'Invalid {name}: "{text}" [length < {min_length}]')
        if len(text) > max_length:

• Read up
  Read up

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

Open

def eratosthenes():
    """Yields the sequence of prime numbers via the Sieve of Eratosthenes."""
    D = {}  # map each composite integer to its first-found prime factor
    q = 2  # q gets 2, 3, 4, 5, ... ad infinitum
    while True:

• Read up
  Read up

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

Open

def getfqdn(name=""):
    """Get fully qualified domain name from name.

    An empty argument is interpreted as meaning the local host.
    """

• Read up
  Read up

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

Open

def check_extension_modules():
    from .. import platform, compress, crypto, item, chunker, hashindex

    msg = """The Borg binary extension modules do not seem to be properly installed."""
    if hashindex.API_VERSION != "1.2_01":

• Read up
  Read up

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

Open

def classify_ec(ec):
    if not isinstance(ec, int):
        raise TypeError("ec must be of type int")
    if EXIT_SIGNAL_BASE <= ec <= 255:
        return "signal"

• Read up
  Read up

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

Open

    def process_items(self, archive, target):
        matcher = self.matcher

        for item in archive.iter_items():
            if not matcher.match(item.path):

• Read up
  Read up

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

Open

    def do_list(self, args, repository, manifest):
        """List archive contents"""
        matcher = build_matcher(args.patterns, args.paths)
        if args.format is not None:
            format = args.format

• Read up
  Read up

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

Open

def reopen(repository, exclusive: Optional[bool] = True, create=False):
    if isinstance(repository, LegacyRepository):
        if repository.io is not None or repository.lock is not None:
            raise RuntimeError("Repo must be closed before a reopen. Cannot support nested repository contexts.")
        return LegacyRepository(repository.path, exclusive=exclusive, create=create)

• Read up
  Read up

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

Open

    def parse(
        self, id: bytes, cdata: bytes, decompress: bool = True, want_compressed: bool = False, ro_type: str = None
    ) -> tuple[dict, bytes]:
        assert not (not decompress and not want_compressed), "invalid parameter combination!"
        assert isinstance(id, bytes)

• Read up
  Read up

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

Open

def process_chunks(repository, repo_objs, stats, recompress_ids, olevel):
    """process some chunks (usually: recompress)"""
    compr_keys = stats["compr_keys"]
    if compr_keys == 0:  # work around defaultdict(int)
        compr_keys = stats["compr_keys"] = set()

• Read up
  Read up