def __init__()

in cqlsh-expansion/pylib/cqlshlib/copyutil.py [0:0]

• 32 lines of code
• 4 McCabe index (conditional complexity)

    def __init__(self, params, target):
        mp.Process.__init__(self, target=target)
        self.inpipe = params['inpipe']
        self.outpipe = params['outpipe']
        self.inmsg = None  # must be initialized after fork on Windows
        self.outmsg = None  # must be initialized after fork on Windows
        self.ks = params['ks']
        self.table = params['table']
        self.local_dc = params['local_dc']
        self.columns = params['columns']
        self.debug = params['debug']
        self.port = params['port']
        self.hostname = params['hostname']
        self.connect_timeout = params['connect_timeout']
        self.cql_version = params['cql_version']
        self.auth_provider = params['auth_provider']
        self.parent_cluster = params['parent_cluster']
        self.ssl = params['ssl']
        self.protocol_version = params['protocol_version']
        self.config_file = params['config_file']

        options = params['options']
        self.date_time_format = options.copy['dtformats']
        self.consistency_level = options.copy['consistencylevel']
        self.decimal_sep = options.copy['decimalsep']
        self.thousands_sep = options.copy['thousandssep']
        self.boolean_styles = options.copy['boolstyle']
        self.max_attempts = options.copy['maxattempts']
        self.encoding = options.copy['encoding']
        # Here we inject some failures for testing purposes, only if this environment variable is set
        if os.environ.get('CQLSH_COPY_TEST_FAILURES', ''):
            self.test_failures = json.loads(os.environ.get('CQLSH_COPY_TEST_FAILURES', ''))
        else:
            self.test_failures = None