PEP-8 code style changes patch by Brad Schoening; reviewed by Brad Schoening and Bret McGuire reference: https://github.com/apache/cassandra-python-driver/pull/1300
diff --git a/cassandra/__init__.py b/cassandra/__init__.py index c732708..f0d47c5 100644 --- a/cassandra/__init__.py +++ b/cassandra/__init__.py
@@ -17,15 +17,18 @@ import logging import importlib.metadata + class NullHandler(logging.Handler): def emit(self, record): pass + logging.getLogger('cassandra').addHandler(NullHandler()) __version__ = importlib.metadata.version('cassandra-driver') + class ConsistencyLevel(object): """ Specifies how many replicas must respond for an operation to be considered @@ -417,10 +420,10 @@ self.consistency = consistency self.required_replicas = required_replicas self.alive_replicas = alive_replicas - Exception.__init__(self, summary_message + ' info=' + - repr({'consistency': consistency_value_to_name(consistency), - 'required_replicas': required_replicas, - 'alive_replicas': alive_replicas})) + Exception.__init__(self, summary_message + ' info=' + + repr({'consistency': consistency_value_to_name(consistency), + 'required_replicas': required_replicas, + 'alive_replicas': alive_replicas})) class Timeout(RequestExecutionException): @@ -632,7 +635,7 @@ class ConfigurationException(RequestValidationException): """ - Server indicated request errro due to current configuration + Server indicated request error due to current configuration """ pass @@ -729,6 +732,7 @@ """ pass + class DependencyException(Exception): """ Specific exception class for handling issues with driver dependencies
diff --git a/cassandra/auth.py b/cassandra/auth.py index 86759af..9aa4d2c 100644 --- a/cassandra/auth.py +++ b/cassandra/auth.py
@@ -215,6 +215,7 @@ def evaluate_challenge(self, challenge): return self.sasl.process(challenge) + # TODO remove me next major DSEPlainTextAuthProvider = PlainTextAuthProvider
diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 10ce3de..f2e1118 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py
@@ -93,21 +93,24 @@ except ImportError: from cassandra.util import WeakSet # NOQA + def _try_libev_import(): try: from cassandra.io.libevreactor import LibevConnection - return (LibevConnection,None) + return (LibevConnection, None) except DependencyException as e: return (None, e) + def _try_asyncore_import(): try: from cassandra.io.asyncorereactor import AsyncoreConnection - return (AsyncoreConnection,None) + return (AsyncoreConnection, None) except DependencyException as e: return (None, e) -def _connection_reduce_fn(val,import_fn): + +def _connection_reduce_fn(val, import_fn): (rv, excs) = val # If we've already found a workable Connection class return immediately if rv: @@ -117,10 +120,11 @@ excs.append(exc) return (rv or import_result, excs) + log = logging.getLogger(__name__) conn_fns = (_try_libev_import, _try_asyncore_import) -(conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None,[])) +(conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None, [])) if not conn_class: raise DependencyException("Unable to load a default connection class", excs) DefaultConnection = conn_class @@ -378,8 +382,8 @@ self.retry_policy = retry_policy or RetryPolicy() - if (serial_consistency_level is not None and - not ConsistencyLevel.is_serial(serial_consistency_level)): + if (serial_consistency_level is not None + and not ConsistencyLevel.is_serial(serial_consistency_level)): raise ValueError("serial_consistency_level must be either " "ConsistencyLevel.SERIAL " "or ConsistencyLevel.LOCAL_SERIAL.") @@ -670,6 +674,7 @@ self._auth_provider = value _load_balancing_policy = None + @property def load_balancing_policy(self): """ @@ -706,6 +711,7 @@ """ _default_retry_policy = RetryPolicy() + @property def default_retry_policy(self): """ @@ -763,9 +769,8 @@ Using ssl_options without ssl_context is deprecated and will be removed in the next major release. - An optional dict which will be used as kwargs for ``ssl.SSLContext.wrap_socket`` - when new sockets are created. This should be used when client encryption is enabled - in Cassandra. + An optional dict which will be used as kwargs for ``ssl.SSLContext.wrap_socket`` when new + sockets are created. This should be used when client encryption is enabled in Cassandra. The following documentation only applies when ssl_options is used without ssl_context. @@ -2240,6 +2245,7 @@ _monitor_reporter = None _row_factory = staticmethod(named_tuple_factory) + @property def row_factory(self): """ @@ -2305,8 +2311,8 @@ *Deprecated:* use execution profiles instead """ warn("Setting the consistency level at the session level will be removed in 4.0. Consider using " - "execution profiles and setting the desired consistency level to the EXEC_PROFILE_DEFAULT profile." - , DeprecationWarning) + "execution profiles and setting the desired consistency level to the EXEC_PROFILE_DEFAULT profile.", + DeprecationWarning) self._validate_set_legacy_config('default_consistency_level', cl) _default_serial_consistency_level = None @@ -2324,8 +2330,8 @@ @default_serial_consistency_level.setter def default_serial_consistency_level(self, cl): - if (cl is not None and - not ConsistencyLevel.is_serial(cl)): + if (cl is not None + and not ConsistencyLevel.is_serial(cl)): raise ValueError("default_serial_consistency_level must be either " "ConsistencyLevel.SERIAL " "or ConsistencyLevel.LOCAL_SERIAL.") @@ -3096,7 +3102,7 @@ for host in tuple(self._pools.keys()): if host != excluded_host and host.is_up: future = ResponseFuture(self, PrepareMessage(query=query, keyspace=keyspace), - None, self.default_timeout) + None, self.default_timeout) # we don't care about errors preparing against specific hosts, # since we can always prepare them as needed when the prepared @@ -3133,7 +3139,7 @@ self.is_shutdown = True # PYTHON-673. If shutdown was called shortly after session init, avoid - # a race by cancelling any initial connection attempts haven't started, + # a race by cancelling any initial connection attempts which haven't started, # then blocking on any that have. for future in self._initial_connect_futures: future.cancel() @@ -3233,8 +3239,7 @@ but also on other nodes (for instance, if a node dies, another previously ignored node may be now considered). - This method ensures that all hosts for which a pool should exist - have one, and hosts that shouldn't don't. + Ensures host pools exist only for eligible hosts. For internal use only. """ @@ -3842,9 +3847,9 @@ @staticmethod def _is_valid_peer(row): - return bool(_NodeInfo.get_broadcast_rpc_address(row) and row.get("host_id") and - row.get("data_center") and row.get("rack") and - ('tokens' not in row or row.get('tokens'))) + return bool(_NodeInfo.get_broadcast_rpc_address(row) and row.get("host_id") + and row.get("data_center") and row.get("rack") + and ('tokens' not in row or row.get('tokens'))) def _update_location_info(self, host, datacenter, rack): if host.datacenter == datacenter and host.rack == rack: @@ -4652,7 +4657,7 @@ current_keyspace = self._connection.keyspace prepared_keyspace = prepared_statement.keyspace if not ProtocolVersion.uses_keyspace_flag(self.session.cluster.protocol_version) \ - and prepared_keyspace and current_keyspace != prepared_keyspace: + and prepared_keyspace and current_keyspace != prepared_keyspace: self._set_final_exception( ValueError("The Session's current keyspace (%s) does " "not match the keyspace the statement was "
diff --git a/cassandra/concurrent.py b/cassandra/concurrent.py index 012f52f..cdd84fd 100644 --- a/cassandra/concurrent.py +++ b/cassandra/concurrent.py
@@ -29,7 +29,9 @@ ExecutionResult = namedtuple('ExecutionResult', ['success', 'result_or_exc']) -def execute_concurrent(session, statements_and_parameters, concurrency=100, raise_on_first_error=True, results_generator=False, execution_profile=EXEC_PROFILE_DEFAULT): + +def execute_concurrent(session, statements_and_parameters, concurrency=100, raise_on_first_error=True, + results_generator=False, execution_profile=EXEC_PROFILE_DEFAULT): """ See :meth:`.Session.execute_concurrent`. """ @@ -161,7 +163,6 @@ return [r[1] for r in sorted(self._results_queue)] - def execute_concurrent_with_args(session, statement, parameters, *args, **kwargs): """ See :meth:`.Session.execute_concurrent_with_args`.
diff --git a/cassandra/connection.py b/cassandra/connection.py index 48aa709..d9bca62 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py
@@ -197,8 +197,8 @@ return self._address, self._port def __eq__(self, other): - return isinstance(other, DefaultEndPoint) and \ - self.address == other.address and self.port == other.port + return isinstance(other, DefaultEndPoint)\ + and self.address == other.address and self.port == other.port def __hash__(self): return hash((self.address, self.port)) @@ -217,8 +217,7 @@ port = None """ - If no port is discovered in the row, this is the default port - used for endpoint creation. + If no port is discovered in the row, this is the default port used for endpoint creation. """ def __init__(self, port=None): @@ -282,16 +281,16 @@ socket.AF_UNSPEC, socket.SOCK_STREAM) def __eq__(self, other): - return (isinstance(other, SniEndPoint) and - self.address == other.address and self.port == other.port and - self._server_name == other._server_name) + return (isinstance(other, SniEndPoint) + and self.address == other.address and self.port == other.port + and self._server_name == other._server_name) def __hash__(self): return hash((self.address, self.port, self._server_name)) def __lt__(self, other): - return ((self.address, self.port, self._server_name) < - (other.address, other.port, self._server_name)) + return ((self.address, self.port, self._server_name) + < (other.address, other.port, self._server_name)) def __str__(self): return str("%s:%d:%s" % (self.address, self.port, self._server_name)) @@ -351,8 +350,8 @@ return self.address, None def __eq__(self, other): - return (isinstance(other, UnixSocketEndPoint) and - self._unix_socket_path == other._unix_socket_path) + return (isinstance(other, UnixSocketEndPoint) + and self._unix_socket_path == other._unix_socket_path) def __hash__(self): return hash(self._unix_socket_path) @@ -378,12 +377,12 @@ def __eq__(self, other): # facilitates testing if isinstance(other, _Frame): - return (self.version == other.version and - self.flags == other.flags and - self.stream == other.stream and - self.opcode == other.opcode and - self.body_offset == other.body_offset and - self.end_pos == other.end_pos) + return (self.version == other.version + and self.flags == other.flags + and self.stream == other.stream + and self.opcode == other.opcode + and self.body_offset == other.body_offset + and self.end_pos == other.end_pos) return NotImplemented def __str__(self): @@ -647,7 +646,7 @@ @property def has_consumed_segment(self): - return self._segment_consumed; + return self._segment_consumed def readable_io_bytes(self): return self.io_buffer.tell() @@ -721,7 +720,7 @@ # If the number of orphaned streams reaches this threshold, this connection # will become marked and will be replaced with a new connection by the # owning pool (currently, only HostConnection supports this) - orphaned_threshold = 3 * max_in_flight // 4 + orphaned_threshold = 3 * max_in_flight // 4 is_defunct = False is_closed = False @@ -869,7 +868,7 @@ # Extract a subset of names from self.ssl_options which apply to SSLContext creation ssl_context_opt_names = ['ssl_version', 'cert_reqs', 'check_hostname', 'keyfile', 'certfile', 'ca_certs', 'ciphers'] - opts = {k:self.ssl_options.get(k, None) for k in ssl_context_opt_names if k in self.ssl_options} + opts = {k: self.ssl_options.get(k, None) for k in ssl_context_opt_names if k in self.ssl_options} # Python >= 3.10 requires either PROTOCOL_TLS_CLIENT or PROTOCOL_TLS_SERVER, so we'll get ahead of things by always # being explicit @@ -897,11 +896,11 @@ # Extract a subset of names from self.ssl_options which apply to SSLContext.wrap_socket (or at least the parts # of it that don't involve building an SSLContext under the covers) wrap_socket_opt_names = ['server_side', 'do_handshake_on_connect', 'suppress_ragged_eofs', 'server_hostname'] - opts = {k:self.ssl_options.get(k, None) for k in wrap_socket_opt_names if k in self.ssl_options} + opts = {k: self.ssl_options.get(k, None) for k in wrap_socket_opt_names if k in self.ssl_options} # PYTHON-1186: set the server_hostname only if the SSLContext has # check_hostname enabled, and it is not already provided by the EndPoint ssl options - #opts['server_hostname'] = self.endpoint.address + # opts['server_hostname'] = self.endpoint.address if (self.ssl_context.check_hostname and 'server_hostname' not in opts): server_hostname = self.endpoint.address opts['server_hostname'] = server_hostname @@ -1355,8 +1354,8 @@ self._compressor = None compression_type = None if self.compression: - overlap = (set(locally_supported_compressions.keys()) & - set(remote_supported_compressions)) + overlap = (set(locally_supported_compressions.keys()) + & set(remote_supported_compressions)) if len(overlap) == 0: log.debug("No available compression types supported on both ends." " locally supported: %r. remotely supported: %r", @@ -1381,8 +1380,8 @@ # If snappy compression is selected with v5+checksumming, the connection # will fail with OTO. Only lz4 is supported - if (compression_type == 'snappy' and - ProtocolVersion.has_checksumming_support(self.protocol_version)): + if (compression_type == 'snappy' + and ProtocolVersion.has_checksumming_support(self.protocol_version)): log.debug("Snappy compression is not supported with protocol version %s and " "checksumming. Consider installing lz4. Disabling compression.", self.protocol_version) compression_type = None
diff --git a/cassandra/cqltypes.py b/cassandra/cqltypes.py index 7cde676..9c76955 100644 --- a/cassandra/cqltypes.py +++ b/cassandra/cqltypes.py
@@ -35,6 +35,7 @@ from collections import namedtuple from decimal import Decimal import io +import ipaddress from itertools import chain import logging import re @@ -53,7 +54,6 @@ from cassandra import util _little_endian_flag = 1 # we always serialize LE -import ipaddress apache_cassandra_type_prefix = 'org.apache.cassandra.db.marshal.' @@ -236,6 +236,7 @@ # return the first (outer) type, which will have all parameters applied return args[0][0][0] + def lookup_casstype(casstype): """ Given a Cassandra type as a string (possibly including parameters), hand @@ -267,6 +268,7 @@ return "EMPTY" __repr__ = __str__ + EMPTY = EmptyValue() @@ -397,6 +399,7 @@ def serial_size(cls): return None + # it's initially named with a _ to avoid registering it as a real type, but # client programs may want to use the name still for isinstance(), etc CassandraType = _CassandraType @@ -465,6 +468,7 @@ def serial_size(cls): return 16 + class BooleanType(_CassandraType): typename = 'boolean' @@ -480,6 +484,7 @@ def serial_size(cls): return 1 + class ByteType(_CassandraType): typename = 'tinyint' @@ -523,6 +528,7 @@ def serial_size(cls): return 4 + class DoubleType(_CassandraType): typename = 'double' @@ -538,6 +544,7 @@ def serial_size(cls): return 8 + class LongType(_CassandraType): typename = 'bigint' @@ -553,6 +560,7 @@ def serial_size(cls): return 8 + class Int32Type(_CassandraType): typename = 'int' @@ -568,6 +576,7 @@ def serial_size(cls): return 4 + class IntegerType(_CassandraType): typename = 'varint' @@ -610,6 +619,7 @@ class CounterColumnType(LongType): typename = 'counter' + cql_timestamp_formats = ( '%Y-%m-%d %H:%M', '%Y-%m-%d %H:%M:%S', @@ -667,6 +677,7 @@ def serial_size(cls): return 8 + class TimestampType(DateType): pass @@ -692,6 +703,7 @@ def serial_size(cls): return 16 + class SimpleDateType(_CassandraType): typename = 'date' date_format = "%Y-%m-%d" @@ -731,13 +743,14 @@ def serialize(byts, protocol_version): return int16_pack(byts) + class TimeType(_CassandraType): typename = 'time' # Time should be a fixed size 8 byte type but Cassandra 5.0 code marks it as # variable size... and we have to match what the server expects since the server # uses that specification to encode data of that type. - #@classmethod - #def serial_size(cls): + # @classmethod + # def serial_size(cls): # return 8 @staticmethod @@ -993,7 +1006,8 @@ 'fieldnames': field_names, 'keyspace': keyspace, 'mapped_class': None, - 'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, field_names)}) + 'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, + field_names)}) cls._cache[(keyspace, udt_name)] = instance return instance @@ -1006,9 +1020,11 @@ @classmethod def apply_parameters(cls, subtypes, names): - keyspace = subtypes[0].cass_parameterized_type() # when parsed from cassandra type, the keyspace is created as an unrecognized cass type; This gets the name back + # when parsed from cassandra type, the keyspace is created as an unrecognized cass type; This resolves the name + keyspace = subtypes[0].cass_parameterized_type() udt_name = _name_from_hex_string(subtypes[1].cassname) - field_names = tuple(_name_from_hex_string(encoded_name) for encoded_name in names[2:]) # using tuple here to match what comes into make_udt_class from other sources (for caching equality test) + # tuple matches what comes into make_udt_class from other sources (for caching equality test) + field_names = tuple(_name_from_hex_string(encoded_name) for encoded_name in names[2:]) return cls.make_udt_class(keyspace, udt_name, field_names, tuple(subtypes[2:])) @classmethod @@ -1430,6 +1446,7 @@ return buf.getvalue() + class VectorType(_CassandraType): typename = 'org.apache.cassandra.db.marshal.VectorType' vector_size = 0 @@ -1454,7 +1471,7 @@ expected_byte_size = serialized_size * cls.vector_size if len(byts) != expected_byte_size: raise ValueError( - "Expected vector of type {0} and dimension {1} to have serialized size {2}; observed serialized size of {3} instead"\ + "Expected vector of type {0} and dimension {1} to have serialized size {2}; observed serialized size of {3} instead" .format(cls.subtype.typename, cls.vector_size, expected_byte_size, len(byts))) indexes = (serialized_size * x for x in range(0, cls.vector_size)) return [cls.subtype.deserialize(byts[idx:idx + serialized_size], protocol_version) for idx in indexes] @@ -1468,8 +1485,8 @@ rv.append(cls.subtype.deserialize(byts[idx:idx + size], protocol_version)) idx += size except: - raise ValueError("Error reading additional data during vector deserialization after successfully adding {} elements"\ - .format(len(rv))) + raise ValueError("Error reading additional data during vector deserialization after successfully adding {} elements" + .format(len(rv))) # If we have any additional data in the serialized vector treat that as an error as well if idx < len(byts): @@ -1481,7 +1498,7 @@ v_length = len(v) if cls.vector_size != v_length: raise ValueError( - "Expected sequence of size {0} for vector of type {1} and dimension {0}, observed sequence of length {2}"\ + "Expected sequence of size {0} for vector of type {1} and dimension {0}, observed sequence of length {2}" .format(cls.vector_size, cls.subtype.typename, v_length)) serialized_size = cls.subtype.serial_size()
diff --git a/cassandra/encoder.py b/cassandra/encoder.py index 94093e8..3298ccf 100644 --- a/cassandra/encoder.py +++ b/cassandra/encoder.py
@@ -20,7 +20,6 @@ """ import logging -log = logging.getLogger(__name__) from binascii import hexlify from decimal import Decimal @@ -35,6 +34,8 @@ from cassandra.util import (OrderedDict, OrderedMap, OrderedMapSerializedKey, sortedset, Time, Date, Point, LineString, Polygon) +log = logging.getLogger(__name__) + def cql_quote(term): if isinstance(term, str): @@ -173,7 +174,7 @@ is suitable for ``IN`` value lists. """ return '(%s)' % ', '.join(self.mapping.get(type(v), self.cql_encode_object)(v) - for v in val) + for v in val) cql_encode_tuple = cql_encode_sequence """ @@ -223,4 +224,4 @@ return "'%s'" % val.compressed def cql_encode_decimal(self, val): - return self.cql_encode_float(float(val)) \ No newline at end of file + return self.cql_encode_float(float(val))
diff --git a/cassandra/marshal.py b/cassandra/marshal.py index e8733f0..001d10f 100644 --- a/cassandra/marshal.py +++ b/cassandra/marshal.py
@@ -23,6 +23,7 @@ unpack = lambda s: packer.unpack(s)[0] return pack, unpack + int64_pack, int64_unpack = _make_packer('>q') int32_pack, int32_unpack = _make_packer('>i') int16_pack, int16_unpack = _make_packer('>h') @@ -113,6 +114,7 @@ return tuple(values) + def vints_pack(values): revbytes = bytearray() values = [int(v) for v in values[::-1]] @@ -127,7 +129,7 @@ # i.e. with 1 extra byte, the first byte needs to be something like '10XXXXXX' # 2 bits reserved # i.e. with 8 extra bytes, the first byte needs to be '11111111' # 8 bits reserved reserved_bits = num_extra_bytes + 1 - while num_bits > (8-(reserved_bits)): + while num_bits > (8 - (reserved_bits)): num_extra_bytes += 1 num_bits -= 8 reserved_bits = min(num_extra_bytes + 1, 8) @@ -145,21 +147,23 @@ revbytes.reverse() return bytes(revbytes) + def uvint_unpack(bytes): first_byte = bytes[0] if (first_byte & 128) == 0: - return (first_byte,1) + return (first_byte, 1) num_extra_bytes = 8 - (~first_byte & 0xff).bit_length() rv = first_byte & (0xff >> num_extra_bytes) - for idx in range(1,num_extra_bytes + 1): + for idx in range(1, num_extra_bytes + 1): new_byte = bytes[idx] rv <<= 8 rv |= new_byte & 0xff return (rv, num_extra_bytes + 1) + def uvint_pack(val): rv = bytearray() if val < 128: @@ -172,7 +176,7 @@ # i.e. with 1 extra byte, the first byte needs to be something like '10XXXXXX' # 2 bits reserved # i.e. with 8 extra bytes, the first byte needs to be '11111111' # 8 bits reserved reserved_bits = num_extra_bytes + 1 - while num_bits > (8-(reserved_bits)): + while num_bits > (8 - (reserved_bits)): num_extra_bytes += 1 num_bits -= 8 reserved_bits = min(num_extra_bytes + 1, 8)
diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 8646441..94c7476 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py
@@ -28,12 +28,6 @@ import struct import random -murmur3 = None -try: - from cassandra.murmur3 import murmur3 -except ImportError as e: - pass - from cassandra import SignatureDescriptor, ConsistencyLevel, InvalidRequest, Unauthorized import cassandra.cqltypes as types from cassandra.encoder import Encoder @@ -44,6 +38,12 @@ from cassandra.pool import HostDistance from cassandra.connection import EndPoint +murmur3 = None +try: + from cassandra.murmur3 import murmur3 +except ImportError as e: + pass + log = logging.getLogger(__name__) cql_keywords = set(( @@ -351,8 +351,8 @@ def _get_host_by_address(self, address, port=None): for host in self._hosts.values(): - if (host.broadcast_rpc_address == address and - (port is None or host.broadcast_rpc_port is None or host.broadcast_rpc_port == port)): + if (host.broadcast_rpc_address == address + and (port is None or host.broadcast_rpc_port is None or host.broadcast_rpc_port == port)): return host return None @@ -386,7 +386,6 @@ return cls - class _ReplicationStrategy(object, metaclass=ReplicationStrategyTypeType): options_map = None @@ -436,9 +435,9 @@ self.options_map['class'] = self.name def __eq__(self, other): - return (isinstance(other, _UnknownStrategy) and - self.name == other.name and - self.options_map == other.options_map) + return (isinstance(other, _UnknownStrategy) + and self.name == other.name + and self.options_map == other.options_map) def export_for_schema(self): """ @@ -626,7 +625,7 @@ racks_this_dc = dc_racks[dc] hosts_this_dc = len(hosts_per_dc[dc]) - for token_offset_index in range(index, index+num_tokens): + for token_offset_index in range(index, index + num_tokens): if token_offset_index >= len(token_offsets): token_offset_index = token_offset_index - len(token_offsets) @@ -789,16 +788,16 @@ other_tables = [t for t in self.tables.values() if t not in tables_with_vertex] cql = "\n\n".join( - [self.as_cql_query() + ';'] + - self.user_type_strings() + - [f.export_as_string() for f in self.functions.values()] + - [a.export_as_string() for a in self.aggregates.values()] + - [t.export_as_string() for t in tables_with_vertex + other_tables]) + [self.as_cql_query() + ';'] + + self.user_type_strings() + + [f.export_as_string() for f in self.functions.values()] + + [a.export_as_string() for a in self.aggregates.values()] + + [t.export_as_string() for t in tables_with_vertex + other_tables]) if self._exc_info: import traceback ret = "/*\nWarning: Keyspace %s is incomplete because of an error processing metadata.\n" % \ - (self.name) + self.name for line in traceback.format_exception(*self._exc_info): ret += line ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % cql @@ -1272,9 +1271,9 @@ if comparator: # no compact storage with more than one column beyond PK if there # are clustering columns - incompatible = (self.is_compact_storage and - len(self.columns) > len(self.primary_key) + 1 and - len(self.clustering_key) >= 1) + incompatible = (self.is_compact_storage + and len(self.columns) > len(self.primary_key) + 1 + and len(self.clustering_key) >= 1) return not incompatible return True @@ -1864,7 +1863,7 @@ def hash_fn(cls, key): if isinstance(key, str): key = key.encode('UTF-8') - return abs(varint_unpack(md5(key,usedforsecurity=False).digest())) + return abs(varint_unpack(md5(key, usedforsecurity=False).digest())) class BytesToken(Token): @@ -2185,8 +2184,8 @@ is_compact = False has_value = False clustering_size = num_column_name_components - 2 - elif (len(column_aliases) == num_column_name_components - 1 and - issubclass(last_col, types.UTF8Type)): + elif (len(column_aliases) == num_column_name_components - 1 + and issubclass(last_col, types.UTF8Type)): # aliases? is_compact = False has_value = False @@ -2522,7 +2521,7 @@ def get_table(self, keyspaces, keyspace, table): cl = ConsistencyLevel.ONE - where_clause = bind_params(" WHERE keyspace_name = %%s AND %s = %%s" % (self._table_name_col), (keyspace, table), _encoder) + where_clause = bind_params(" WHERE keyspace_name = %%s AND %s = %%s" % self._table_name_col, (keyspace, table), _encoder) cf_query = QueryMessage(query=self._SELECT_TABLES + where_clause, consistency_level=cl) col_query = QueryMessage(query=self._SELECT_COLUMNS + where_clause, consistency_level=cl) indexes_query = QueryMessage(query=self._SELECT_INDEXES + where_clause, consistency_level=cl) @@ -2749,8 +2748,7 @@ """ For DSE 6.0+ """ - recognized_table_options = (SchemaParserV3.recognized_table_options + - ("nodesync",)) + recognized_table_options = (SchemaParserV3.recognized_table_options + ("nodesync",)) class SchemaParserV4(SchemaParserV3): @@ -2896,8 +2894,7 @@ """ For DSE 6.7+ """ - recognized_table_options = (SchemaParserV4.recognized_table_options + - ("nodesync",)) + recognized_table_options = (SchemaParserV4.recognized_table_options + ("nodesync",)) class SchemaParserDSE68(SchemaParserDSE67): @@ -2923,7 +2920,7 @@ def get_table(self, keyspaces, keyspace, table): table_meta = super(SchemaParserDSE68, self).get_table(keyspaces, keyspace, table) cl = ConsistencyLevel.ONE - where_clause = bind_params(" WHERE keyspace_name = %%s AND %s = %%s" % (self._table_name_col), (keyspace, table), _encoder) + where_clause = bind_params(" WHERE keyspace_name = %%s AND %s = %%s" % self._table_name_col, (keyspace, table), _encoder) vertices_query = QueryMessage(query=self._SELECT_VERTICES + where_clause, consistency_level=cl) edges_query = QueryMessage(query=self._SELECT_EDGES + where_clause, consistency_level=cl) @@ -3316,6 +3313,8 @@ return "RESTRICT ROWS ON %s.%s USING %s;" % (protect_name(table_meta.keyspace_name), protect_name(table_meta.name), protect_name(ext_blob.decode('utf-8'))) + + NO_VALID_REPLICA = object()
diff --git a/cassandra/metrics.py b/cassandra/metrics.py index a1eadc1..a585000 100644 --- a/cassandra/metrics.py +++ b/cassandra/metrics.py
@@ -123,22 +123,24 @@ self.stats_name = 'cassandra-{0}'.format(str(self._stats_counter)) Metrics._stats_counter += 1 self.stats = scales.collection(self.stats_name, - scales.PmfStat('request_timer'), - scales.IntStat('connection_errors'), - scales.IntStat('write_timeouts'), - scales.IntStat('read_timeouts'), - scales.IntStat('unavailables'), - scales.IntStat('other_errors'), - scales.IntStat('retries'), - scales.IntStat('ignores'), + scales.PmfStat('request_timer'), + scales.IntStat('connection_errors'), + scales.IntStat('write_timeouts'), + scales.IntStat('read_timeouts'), + scales.IntStat('unavailables'), + scales.IntStat('other_errors'), + scales.IntStat('retries'), + scales.IntStat('ignores'), - # gauges - scales.Stat('known_hosts', - lambda: len(cluster_proxy.metadata.all_hosts())), - scales.Stat('connected_to', - lambda: len(set(chain.from_iterable(s._pools.keys() for s in cluster_proxy.sessions)))), - scales.Stat('open_connections', - lambda: sum(sum(p.open_count for p in s._pools.values()) for s in cluster_proxy.sessions))) + # gauges + scales.Stat('known_hosts', + lambda: len(cluster_proxy.metadata.all_hosts())), + scales.Stat('connected_to', + lambda: len(set(chain.from_iterable( + s._pools.keys() for s in cluster_proxy.sessions)))), + scales.Stat('open_connections', + lambda: sum(sum(p.open_count for p in s._pools.values()) for s in + cluster_proxy.sessions))) # TODO, to be removed in 4.0 # /cassandra contains the metrics of the first cluster registered
diff --git a/cassandra/murmur3.py b/cassandra/murmur3.py index 282c435..cb80f27 100644 --- a/cassandra/murmur3.py +++ b/cassandra/murmur3.py
@@ -2,15 +2,15 @@ def body_and_tail(data): - l = len(data) - nblocks = l // 16 - tail = l % 16 + length = len(data) + nblocks = length // 16 + tail = length % 16 if nblocks: # we use '<', specifying little-endian byte order for data bigger than # a byte so behavior is the same on little- and big-endian platforms - return struct.unpack_from('<' + ('qq' * nblocks), data), struct.unpack_from('b' * tail, data, -tail), l + return struct.unpack_from('<' + ('qq' * nblocks), data), struct.unpack_from('b' * tail, data, -tail), length else: - return tuple(), struct.unpack_from('b' * tail, data, -tail), l + return tuple(), struct.unpack_from('b' * tail, data, -tail), length def rotl64(x, r): @@ -108,6 +108,7 @@ return truncate_int64(h1) + try: from cassandra.cmurmur3 import murmur3 except ImportError:
diff --git a/cassandra/policies.py b/cassandra/policies.py index d6f7063..bc1861a 100644 --- a/cassandra/policies.py +++ b/cassandra/policies.py
@@ -22,17 +22,16 @@ from threading import Lock import socket import warnings +from cassandra import WriteType as WT +from cassandra import ConsistencyLevel, OperationTimedOut log = logging.getLogger(__name__) -from cassandra import WriteType as WT - # This is done this way because WriteType was originally # defined here and in order not to break the API. # It may be removed in the next major. WriteType = WT -from cassandra import ConsistencyLevel, OperationTimedOut class HostDistance(object): """ @@ -1187,6 +1186,7 @@ ColDesc = namedtuple('ColDesc', ['ks', 'table', 'col']) + class ColumnEncryptionPolicy(object): """ A policy enabling (mostly) transparent encryption and decryption of data before it is
diff --git a/cassandra/pool.py b/cassandra/pool.py index 37fdaee..d060eb2 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py
@@ -87,7 +87,7 @@ broadcast_rpc_port = None """ The broadcast rpc port of the node, *if available*: - + 'system.local.rpc_port' or 'system.peers.native_transport_port' (DSE 6+) 'system.local.rpc_port' or 'system.peers_v2.native_port' (Cassandra 4) """ @@ -571,14 +571,15 @@ open_count = 1 if connection and not (connection.is_closed or connection.is_defunct) else 0 in_flights = [connection.in_flight] if connection else [] orphan_requests = [connection.orphaned_request_ids] if connection else [] - return {'shutdown': self.is_shutdown, 'open_count': open_count, \ - 'in_flights': in_flights, 'orphan_requests': orphan_requests} + return {'shutdown': self.is_shutdown, 'open_count': open_count, + 'in_flights': in_flights, 'orphan_requests': orphan_requests} @property def open_count(self): connection = self._connection return 1 if connection and not (connection.is_closed or connection.is_defunct) else 0 + _MAX_SIMULTANEOUS_CREATION = 1 _MIN_TRASH_INTERVAL = 10 @@ -752,7 +753,7 @@ while remaining > 0: # wait on our condition for the possibility that a connection - # is useable + # is usable self._await_available_conn(remaining) # self.shutdown() may trigger the above Condition @@ -931,5 +932,5 @@ def get_state(self): in_flights = [c.in_flight for c in self._connections] orphan_requests = [c.orphaned_request_ids for c in self._connections] - return {'shutdown': self.is_shutdown, 'open_count': self.open_count, \ - 'in_flights': in_flights, 'orphan_requests': orphan_requests} + return {'shutdown': self.is_shutdown, 'open_count': self.open_count, + 'in_flights': in_flights, 'orphan_requests': orphan_requests}
diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 69340a8..b1c4183 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py
@@ -54,6 +54,7 @@ class InternalError(Exception): pass + ColumnMetadata = namedtuple("ColumnMetadata", ['keyspace_name', 'table_name', 'name', 'type']) HEADER_DIRECTION_TO_CLIENT = 0x80 @@ -1168,8 +1169,8 @@ :param decompressor: optional decompression function to inflate the body :return: a message decoded from the body and frame attributes """ - if (not ProtocolVersion.has_checksumming_support(protocol_version) and - flags & COMPRESSED_FLAG): + if (not ProtocolVersion.has_checksumming_support(protocol_version) + and flags & COMPRESSED_FLAG): if decompressor is None: raise RuntimeError("No de-compressor available for compressed frame!") body = decompressor(body)
diff --git a/cassandra/query.py b/cassandra/query.py index 04c515d..d359ccb 100644 --- a/cassandra/query.py +++ b/cassandra/query.py
@@ -85,6 +85,7 @@ """ return rows + def named_tuple_factory(colnames, rows): """ Returns each row as a `namedtuple <https://docs.python.org/2/library/collections.html#collections.namedtuple>`_. @@ -245,8 +246,8 @@ def _key_parts_packed(self, parts): for p in parts: - l = len(p) - yield struct.pack(">H%dsB" % l, l, p, 0) + length = len(p) + yield struct.pack(">H%dsB" % length, length, p, 0) def _get_routing_key(self): return self._routing_key @@ -280,8 +281,8 @@ return self._serial_consistency_level def _set_serial_consistency_level(self, serial_consistency_level): - if (serial_consistency_level is not None and - not ConsistencyLevel.is_serial(serial_consistency_level)): + if (serial_consistency_level is not None + and not ConsistencyLevel.is_serial(serial_consistency_level)): raise ValueError( "serial_consistency_level must be either ConsistencyLevel.SERIAL " "or ConsistencyLevel.LOCAL_SERIAL") @@ -384,7 +385,7 @@ <b>A note about <code>*</code> in prepared statements</b> """ - column_metadata = None #TODO: make this bind_metadata in next major + column_metadata = None # TODO: make this bind_metadata in next major retry_policy = None consistency_level = None custom_payload = None @@ -1058,8 +1059,8 @@ it usable in a targeted LBP without modifying the user's statement. """ def __init__(self, inner_statement, target_host): - self.__class__ = type(inner_statement.__class__.__name__, - (self.__class__, inner_statement.__class__), - {}) - self.__dict__ = inner_statement.__dict__ - self.target_host = target_host + self.__class__ = type(inner_statement.__class__.__name__, + (self.__class__, inner_statement.__class__), + {}) + self.__dict__ = inner_statement.__dict__ + self.target_host = target_host
diff --git a/cassandra/timestamps.py b/cassandra/timestamps.py index e2a2c1e..6a95802 100644 --- a/cassandra/timestamps.py +++ b/cassandra/timestamps.py
@@ -25,6 +25,7 @@ log = logging.getLogger(__name__) + class MonotonicTimestampGenerator(object): """ An object that, when called, returns ``int(time.time() * 1e6)`` when @@ -98,9 +99,9 @@ diff = self.last - now since_last_warn = now - self._last_warn - warn = (self.warn_on_drift and - (diff >= self.warning_threshold * 1e6) and - (since_last_warn >= self.warning_interval * 1e6)) + warn = (self.warn_on_drift + and (diff >= self.warning_threshold * 1e6) + and (since_last_warn >= self.warning_interval * 1e6)) if warn: log.warning( "Clock skew detected: current tick ({now}) was {diff} "
diff --git a/cassandra/util.py b/cassandra/util.py index 408211e..f460396 100644 --- a/cassandra/util.py +++ b/cassandra/util.py
@@ -31,15 +31,14 @@ import time import uuid +from cassandra import DriverException + _HAS_GEOMET = True try: from geomet import wkt except: _HAS_GEOMET = False - -from cassandra import DriverException - DATETIME_EPOC = datetime.datetime(1970, 1, 1).replace(tzinfo=None) UTC_DATETIME_EPOC = datetime.datetime.fromtimestamp(0, tz=datetime.timezone.utc).replace(tzinfo=None) @@ -112,7 +111,8 @@ See :func:`uuid_from_time` for argument and return types. """ - return uuid_from_time(timestamp, 0x808080808080, 0x80) # Cassandra does byte-wise comparison; fill with min signed bytes (0x80 = -128) + # Pad with 0x80 (min signed byte) to ensure this UUID acts as the lower bound for the timestamp + return uuid_from_time(timestamp, 0x808080808080, 0x80) def max_uuid_from_time(timestamp): @@ -176,6 +176,7 @@ return uuid.UUID(fields=(time_low, time_mid, time_hi_version, clock_seq_hi_variant, clock_seq_low, node), version=1) + LOWEST_TIME_UUID = uuid.UUID('00000000-0000-1000-8080-808080808080') """ The lowest possible TimeUUID, as sorted by Cassandra. """ @@ -191,7 +192,7 @@ """ try: value = socket.getaddrinfo(contact_point, port, - socket.AF_UNSPEC, socket.SOCK_STREAM) + socket.AF_UNSPEC, socket.SOCK_STREAM) return value except socket.gaierror: log.debug('Could not resolve hostname "{}" ' @@ -264,10 +265,10 @@ self.update(data) def _commit_removals(self): - l = self._pending_removals + pending = self._pending_removals discard = self.data.discard - while l: - discard(l.pop()) + while pending: + discard(pending.pop()) def __iter__(self): with _IterationGuard(self): @@ -347,6 +348,7 @@ def difference(self, other): return self._apply(other, self.data.difference) + __sub__ = difference def difference_update(self, other): @@ -368,6 +370,7 @@ def intersection(self, other): return self._apply(other, self.data.intersection) + __and__ = intersection def intersection_update(self, other): @@ -383,6 +386,7 @@ def issubset(self, other): return self.data.issubset(ref(item) for item in other) + __lt__ = issubset def __le__(self, other): @@ -390,6 +394,7 @@ def issuperset(self, other): return self.data.issuperset(ref(item) for item in other) + __gt__ = issuperset def __ge__(self, other): @@ -402,6 +407,7 @@ def symmetric_difference(self, other): return self._apply(other, self.data.symmetric_difference) + __xor__ = symmetric_difference def symmetric_difference_update(self, other): @@ -423,6 +429,7 @@ def union(self, other): return self._apply(other, self.data.union) + __or__ = union def isdisjoint(self, other): @@ -495,6 +502,7 @@ def __and__(self, other): return self._intersect(other) + __rand__ = __and__ def __iand__(self, other): @@ -504,6 +512,7 @@ def __or__(self, other): return self.union(other) + __ror__ = __or__ def __ior__(self, other): @@ -524,6 +533,7 @@ def __xor__(self, other): return self.symmetric_difference(other) + __rxor__ = __xor__ def __ixor__(self, other): @@ -636,8 +646,10 @@ try: while lo < hi: mid = (lo + hi) // 2 - if a[mid] < x: lo = mid + 1 - else: hi = mid + if a[mid] < x: + lo = mid + 1 + else: + hi = mid except TypeError: # could not compare a[mid] with x # start scanning to find insertion point while swallowing type errors @@ -645,18 +657,21 @@ compared_one = False # flag is used to determine whether un-comparables are grouped at the front or back while lo < hi: try: - if a[lo] == x or a[lo] >= x: break + if a[lo] == x or a[lo] >= x: + break compared_one = True except TypeError: - if compared_one: break + if compared_one: + break lo += 1 return lo + sortedset = SortedSet # backwards-compatibility class OrderedMap(Mapping): - ''' + """ An ordered map that accepts non-hashable types for keys. It also maintains the insertion order of items, behaving as OrderedDict in that regard. These maps are constructed and read just as normal mapping types, except that they may @@ -680,7 +695,7 @@ This class derives from the (immutable) Mapping API. Objects in these maps are not intended be modified. - ''' + """ def __init__(self, *args, **kwargs): if len(args) > 1: @@ -783,11 +798,11 @@ @total_ordering class Time(object): - ''' + """ Idealized time, independent of day. Up to nanosecond resolution - ''' + """ MICRO = 1000 MILLI = 1000 * MICRO @@ -861,9 +876,9 @@ try: parts = s.split('.') base_time = time.strptime(parts[0], "%H:%M:%S") - self.nanosecond_time = (base_time.tm_hour * Time.HOUR + - base_time.tm_min * Time.MINUTE + - base_time.tm_sec * Time.SECOND) + self.nanosecond_time = (base_time.tm_hour * Time.HOUR + + base_time.tm_min * Time.MINUTE + + base_time.tm_sec * Time.SECOND) if len(parts) > 1: # right pad to 9 digits @@ -874,10 +889,10 @@ raise ValueError("can't interpret %r as a time" % (s,)) def _from_time(self, t): - self.nanosecond_time = (t.hour * Time.HOUR + - t.minute * Time.MINUTE + - t.second * Time.SECOND + - t.microsecond * Time.MICRO) + self.nanosecond_time = (t.hour * Time.HOUR + + t.minute * Time.MINUTE + + t.second * Time.SECOND + + t.microsecond * Time.MICRO) def __hash__(self): return self.nanosecond_time @@ -911,13 +926,13 @@ @total_ordering class Date(object): - ''' + """ Idealized date: year, month, day Offers wider year range than datetime.date. For Dates that cannot be represented as a datetime.date (because datetime.MINYEAR, datetime.MAXYEAR), this type falls back to printing days_from_epoch offset. - ''' + """ MINUTE = 60 HOUR = 60 * MINUTE @@ -1016,10 +1031,10 @@ names_out = list(field_names) for index, name in enumerate(field_names): if (not all(c.isalnum() or c == '_' for c in name) - or keyword.iskeyword(name) - or not name - or name[0].isdigit() - or name.startswith('_')): + or keyword.iskeyword(name) + or not name + or name[0].isdigit() + or name.startswith('_')): names_out[index] = 'field_%d_' % index return names_out @@ -1114,6 +1129,7 @@ """ Tuple of (x, y) coordinates in the linestring """ + def __init__(self, coords=tuple()): """ 'coords`: a sequence of (x, y) coordinates of points in the linestring @@ -1151,7 +1167,8 @@ raise ValueError("Invalid WKT geometry: '{0}'".format(s)) if geom['type'] != 'LineString': - raise ValueError("Invalid WKT geometry type. Expected 'LineString', got '{0}': '{1}'".format(geom['type'], s)) + raise ValueError( + "Invalid WKT geometry type. Expected 'LineString', got '{0}': '{1}'".format(geom['type'], s)) geom['coordinates'] = list_contents_to_tuple(geom['coordinates']) @@ -1244,7 +1261,8 @@ return Polygon(exterior=exterior, interiors=interiors) -_distance_wkt_pattern = re.compile("distance *\\( *\\( *([\\d\\.-]+) *([\\d+\\.-]+) *\\) *([\\d+\\.-]+) *\\) *$", re.IGNORECASE) +_distance_wkt_pattern = re.compile("distance *\\( *\\( *([\\d\\.-]+) *([\\d+\\.-]+) *\\) *([\\d+\\.-]+) *\\) *$", + re.IGNORECASE) class Distance(object): @@ -1320,7 +1338,8 @@ self.nanoseconds = nanoseconds def __eq__(self, other): - return isinstance(other, self.__class__) and self.months == other.months and self.days == other.days and self.nanoseconds == other.nanoseconds + return isinstance(other, + self.__class__) and self.months == other.months and self.days == other.days and self.nanoseconds == other.nanoseconds def __repr__(self): return "Duration({0}, {1}, {2})".format(self.months, self.days, self.nanoseconds) @@ -1471,12 +1490,10 @@ def __eq__(self, other): if not isinstance(other, self.__class__): return NotImplemented - return (self.milliseconds == other.milliseconds and - self.precision == other.precision) + return (self.milliseconds == other.milliseconds and self.precision == other.precision) def __lt__(self, other): - return ((str(self.milliseconds), str(self.precision)) < - (str(other.milliseconds), str(other.precision))) + return ((str(self.milliseconds), str(self.precision)) < (str(other.milliseconds), str(other.precision))) def datetime(self): """ @@ -1672,13 +1689,13 @@ def __eq__(self, other): if not isinstance(other, self.__class__): return NotImplemented - return (self.lower_bound == other.lower_bound and - self.upper_bound == other.upper_bound and - self.value == other.value) + return (self.lower_bound == other.lower_bound + and self.upper_bound == other.upper_bound + and self.value == other.value) def __lt__(self, other): - return ((str(self.lower_bound), str(self.upper_bound), str(self.value)) < - (str(other.lower_bound), str(other.upper_bound), str(other.value))) + return ((str(self.lower_bound), str(self.upper_bound), str(self.value)) + < (str(other.lower_bound), str(other.upper_bound), str(other.value))) def __str__(self): if self.value: @@ -1692,15 +1709,17 @@ self.lower_bound, self.upper_bound, self.value ) + VERSION_REGEX = re.compile("^(\\d+)\\.(\\d+)(\\.\\d+)?(\\.\\d+)?([~\\-]\\w[.\\w]*(?:-\\w[.\\w]*)*)?(\\+[.\\w]+)?$") + @total_ordering class Version(object): """ Representation of a Cassandra version. Mostly follows the implementation of the same logic in the Java driver; see https://github.com/apache/cassandra-java-driver/blob/4.19.2/core/src/main/java/com/datastax/oss/driver/api/core/Version.java. - Cassandra versions are assumed to correspond to major.minor.patch with an optional additional numeric build field as well as a + Versions correspond to major.minor.patch with an optional additional numeric build field and string prerelease field. """
diff --git a/tests/__init__.py b/tests/__init__.py index 047da82..e9ee854 100644 --- a/tests/__init__.py +++ b/tests/__init__.py
@@ -37,7 +37,7 @@ cython_env = os.getenv('VERIFY_CYTHON', "False") VERIFY_CYTHON = False -if(cython_env == 'True'): +if (cython_env == 'True'): VERIFY_CYTHON = True thread_pool_executor_class = ThreadPoolExecutor
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py index 68593f3..a9da910 100644 --- a/tests/integration/__init__.py +++ b/tests/integration/__init__.py
@@ -61,7 +61,7 @@ # # TODO: In the future we may want to make this configurable, but this should only apply # if a non-standard port were specified when starting up the cluster. -DEFAULT_SINGLE_INTERFACE_PORT=9046 +DEFAULT_SINGLE_INTERFACE_PORT = 9046 CCM_CLUSTER = None @@ -168,6 +168,7 @@ dse_ver = "2.1" return dse_ver + USE_CASS_EXTERNAL = bool(os.getenv('USE_CASS_EXTERNAL', False)) KEEP_TEST_CLUSTER = bool(os.getenv('KEEP_TEST_CLUSTER', False)) SIMULACRON_JAR = os.getenv('SIMULACRON_JAR', None) @@ -271,7 +272,7 @@ elif CASSANDRA_VERSION >= Version('3.0'): return (3, 4) elif CASSANDRA_VERSION >= Version('2.2'): - return (1,2, 3, 4) + return (1, 2, 3, 4) elif CASSANDRA_VERSION >= Version('2.1'): return (1, 2, 3) elif CASSANDRA_VERSION >= Version('2.0'): @@ -333,6 +334,7 @@ return _id_and_mark + local = local_decorator_creator() notprotocolv1 = unittest.skipUnless(PROTOCOL_VERSION > 1, 'Protocol v1 not supported') lessthenprotocolv4 = unittest.skipUnless(PROTOCOL_VERSION < 4, 'Protocol versions 4 or greater not supported') @@ -367,7 +369,8 @@ requiredse = unittest.skipUnless(DSE_VERSION, "DSE required") requirescloudproxy = unittest.skipIf(CLOUD_PROXY_PATH is None, "Cloud Proxy path hasn't been specified") -libevtest = unittest.skipUnless(EVENT_LOOP_MANAGER=="libev", "Test timing designed for libev loop") +libevtest = unittest.skipUnless(EVENT_LOOP_MANAGER == "libev", "Test timing designed for libev loop") + def wait_for_node_socket(node, timeout): binary_itf = node.network_interfaces['binary'] @@ -420,10 +423,10 @@ global CCM_CLUSTER log.debug("Checking log error of cluster {0}".format(CCM_CLUSTER.name)) for node in CCM_CLUSTER.nodelist(): - errors = node.grep_log_for_errors() - for error in errors: - for line in error: - print(line) + errors = node.grep_log_for_errors() + for error in errors: + for line in error: + print(line) def remove_cluster(): @@ -859,15 +862,15 @@ execute_until_pass(cls.session, ddl) def create_function_table(self): - ddl = ''' + ddl = ''' CREATE TABLE {0}.{1} ( k int PRIMARY KEY, v int )'''.format(self.keyspace_name, self.function_table_name) - execute_until_pass(self.session, ddl) + execute_until_pass(self.session, ddl) def drop_function_table(self): - ddl = "DROP TABLE {0}.{1} ".format(self.keyspace_name, self.function_table_name) - execute_until_pass(self.session, ddl) + ddl = "DROP TABLE {0}.{1} ".format(self.keyspace_name, self.function_table_name) + execute_until_pass(self.session, ddl) class MockLoggingHandler(logging.Handler): @@ -893,7 +896,7 @@ count = 0 for msg in self.messages.get(level): if sub_string in msg: - count+=1 + count += 1 return count def set_module_name(self, module_name): @@ -1046,6 +1049,7 @@ kwargs['allow_beta_protocol_version'] = cls.DEFAULT_ALLOW_BETA return Cluster(**kwargs) + # Subclass of CCMCluster (i.e. ccmlib.cluster.Cluster) which transparently performs # conversion of cassandra.yml directives into something matching the new syntax # introduced by CASSANDRA-15234 @@ -1079,5 +1083,5 @@ return v def set_configuration_options(self, values=None, *args, **kwargs): - new_values = {self._get_config_key(k, str(v)):self._get_config_val(k, str(v)) for (k,v) in values.items()} - super(Cassandra41CCMCluster, self).set_configuration_options(values=new_values, *args, **kwargs) \ No newline at end of file + new_values = {self._get_config_key(k, str(v)): self._get_config_val(k, str(v)) for (k, v) in values.items()} + super(Cassandra41CCMCluster, self).set_configuration_options(values=new_values, *args, **kwargs)
diff --git a/tests/integration/datatype_utils.py b/tests/integration/datatype_utils.py index a4c4cdb..ec7d279 100644 --- a/tests/integration/datatype_utils.py +++ b/tests/integration/datatype_utils.py
@@ -139,6 +139,7 @@ return sample_data + SAMPLE_DATA = get_sample_data()
diff --git a/tests/util.py b/tests/util.py index d44d6c9..5b4fd7f 100644 --- a/tests/util.py +++ b/tests/util.py
@@ -13,7 +13,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - + import time from functools import wraps @@ -54,7 +54,7 @@ return True, result attempt = 0 - while attempt < (max_attempts-1): + while attempt < (max_attempts - 1): attempt += 1 success, result = wrapped_condition() if success: