File fs.py has 444 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import errno
import hashlib
import os
import posixpath
import re

File manifest.py has 435 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import enum
import re
from collections import namedtuple
from datetime import datetime, timedelta, timezone
from operator import attrgetter

File tar_cmds.py has 435 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import argparse
import base64
import logging
import os
import stat

Function api has a Cognitive Complexity of 42 (exceeds 5 allowed). Consider refactoring.

Open

def api(*, since, **kwargs_decorator):
    """Check version requirements and use self.call to do the remote method call.

    <since> specifies the version in which borg introduced this method.
    Calling this method when connected to an older version will fail without transmitting anything to the server.

• Read up
  Read up

Function api has a Cognitive Complexity of 42 (exceeds 5 allowed). Consider refactoring.

Open

def api(*, since, **kwargs_decorator):
    """Check version requirements and use self.call to do the remote method call.

    <since> specifies the version in which borg introduced this method.
    Calling this method when connected to an older version will fail without transmitting anything to the server.

• Read up
  Read up

File repository.py has 433 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import os
import time

from borgstore.store import Store
from borgstore.store import ObjectNotFound as StoreObjectNotFound

Function do_extract has a Cognitive Complexity of 38 (exceeds 5 allowed). Consider refactoring.

Open

    def do_extract(self, args, repository, manifest, archive):
        """Extract archive contents"""
        # be restrictive when restoring files, restore permissions later
        if sys.getfilesystemencoding() == "ascii":
            logger.warning(

• Read up
  Read up

File __init__.py has 381 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import errno
import filecmp
import io
import os
import re

Function write_options has a Cognitive Complexity of 34 (exceeds 5 allowed). Consider refactoring.

Open

    def write_options(self, parser, fp):
        def is_positional_group(group):
            return any(not o.option_strings for o in group._group_actions)

        # HTML output:

• Read up
  Read up

Function prepare_txn has a Cognitive Complexity of 34 (exceeds 5 allowed). Consider refactoring.

Open

    def prepare_txn(self, transaction_id, do_cleanup=True):
        self._active_txn = True
        if self.do_lock and not self.lock.got_exclusive_lock():
            if self.exclusive is not None:
                # self.exclusive is either True or False, thus a new client is active here.

• Read up
  Read up

File fslocking.py has 375 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import errno
import json
import os
import tempfile
import time

Function generate_level has a Cognitive Complexity of 33 (exceeds 5 allowed). Consider refactoring.

Open

    def generate_level(self, prefix, parser, Archiver, extra_choices=None):
        is_subcommand = False
        choices = {}
        for action in parser._actions:
            if action.choices is not None and "SubParsersAction" in str(action.__class__):

• Read up
  Read up

Function translate has a Cognitive Complexity of 33 (exceeds 5 allowed). Consider refactoring.

Open

def translate(pat, match_end=r"\Z"):
    """Translate a shell-style pattern to a regular expression.

    The pattern may include ``**<sep>`` (<sep> stands for the platform-specific path separator; "/" on POSIX systems)
    for matching zero or more directory levels and "*" for matching zero or more arbitrary characters except any path

• Read up
  Read up

File help_cmd.py has 362 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import collections
import functools
import textwrap

from ..constants import *  # NOQA

Function main has a Cognitive Complexity of 31 (exceeds 5 allowed). Consider refactoring.

Open

def main():  # pragma: no cover
    # Make sure stdout and stderr have errors='replace' to avoid unicode
    # issues when print()-ing unicode file names
    sys.stdout = ErrorIgnoringTextIOWrapper(sys.stdout.buffer, sys.stdout.encoding, "replace", line_buffering=True)
    sys.stderr = ErrorIgnoringTextIOWrapper(sys.stderr.buffer, sys.stderr.encoding, "replace", line_buffering=True)

• Read up
  Read up

Function do_repo_delete has a Cognitive Complexity of 30 (exceeds 5 allowed). Consider refactoring.

Open

    def do_repo_delete(self, args, repository):
        """Delete a repository"""
        self.output_list = args.output_list
        dry_run = args.dry_run
        keep_security_info = args.keep_security_info

• Read up
  Read up

RemoteRepository has 34 functions (exceeds 20 allowed). Consider refactoring.

Open

class RemoteRepository:
    extra_test_args = []  # type: ignore

    class RPCError(Exception):
        def __init__(self, unpacked):

Function acquire has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def acquire(self):
        # goal
        # for exclusive lock: there must be only 1 exclusive lock and no other (exclusive or non-exclusive) locks.
        # for non-exclusive lock: there can be multiple n-e locks, but there must not exist an exclusive lock.
        logger.debug(f"LOCK-ACQUIRE: trying to acquire a lock. exclusive: {self.is_exclusive}.")

• Read up
  Read up

Function show_progress has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def show_progress(self, item=None, final=False, stream=None, dt=None):
        now = time.monotonic()
        if dt is None or now - self.last_progress > dt:
            self.last_progress = now
            if self.output_json:

• Read up
  Read up

Function analyze_archives has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def analyze_archives(self) -> Tuple[Set, Set, int, int, int]:
        """Iterate over all items in all archives, create the dicts id -> size of all used/wanted chunks."""

        def use_it(id, *, wanted=False):
            entry = self.chunks.get(id)

• Read up
  Read up