From 2349b0f42954b0f2ed5025d666cccdd42ca3876c Mon Sep 17 00:00:00 2001 From: Brad Schoening Date: Tue, 14 Jul 2026 16:44:23 -0400 Subject: [PATCH] PEP-8 code style changes --- cassandra/__init__.py | 14 ++-- cassandra/auth.py | 1 + cassandra/cluster.py | 47 +++++++------ cassandra/concurrent.py | 5 +- cassandra/connection.py | 51 +++++++------- cassandra/cqltypes.py | 37 +++++++--- cassandra/encoder.py | 7 +- cassandra/marshal.py | 12 ++-- cassandra/metadata.py | 61 ++++++++-------- cassandra/metrics.py | 34 ++++----- cassandra/murmur3.py | 11 +-- cassandra/policies.py | 6 +- cassandra/pool.py | 13 ++-- cassandra/protocol.py | 5 +- cassandra/query.py | 21 +++--- cassandra/timestamps.py | 7 +- cassandra/util.py | 103 ++++++++++++++++------------ tests/__init__.py | 2 +- tests/integration/__init__.py | 32 +++++---- tests/integration/datatype_utils.py | 1 + tests/util.py | 4 +- 21 files changed, 268 insertions(+), 206 deletions(-) diff --git a/cassandra/__init__.py b/cassandra/__init__.py index c732708605..f0d47c5602 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 @@ def __init__(self, summary_message, consistency=None, required_replicas=None, al 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 RequestValidationException(DriverException): class ConfigurationException(RequestValidationException): """ - Server indicated request errro due to current configuration + Server indicated request error due to current configuration """ pass @@ -729,6 +732,7 @@ class UnresolvableContactPoints(DriverException): """ pass + class DependencyException(Exception): """ Specific exception class for handling issues with driver dependencies diff --git a/cassandra/auth.py b/cassandra/auth.py index 86759afe4d..9aa4d2c359 100644 --- a/cassandra/auth.py +++ b/cassandra/auth.py @@ -215,6 +215,7 @@ def initial_response(self): 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 9ffefe99b5..24e6965bb0 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -98,21 +98,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: @@ -122,10 +125,11 @@ def _connection_reduce_fn(val,import_fn): 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 @@ -383,8 +387,8 @@ def __init__(self, load_balancing_policy=_NOT_SET, retry_policy=None, 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.") @@ -675,6 +679,7 @@ def auth_provider(self, value): self._auth_provider = value _load_balancing_policy = None + @property def load_balancing_policy(self): """ @@ -711,6 +716,7 @@ def _default_load_balancing_policy(self): """ _default_retry_policy = RetryPolicy() + @property def default_retry_policy(self): """ @@ -768,9 +774,8 @@ def default_retry_policy(self, policy): 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. @@ -2289,6 +2294,7 @@ class Session(object): _monitor_reporter = None _row_factory = staticmethod(named_tuple_factory) + @property def row_factory(self): """ @@ -2354,8 +2360,8 @@ def default_consistency_level(self, cl): *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 @@ -2373,8 +2379,8 @@ def default_serial_consistency_level(self): @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.") @@ -3161,7 +3167,7 @@ def prepare_on_all_hosts(self, query, excluded_host, keyspace=None): 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 @@ -3198,7 +3204,7 @@ def shutdown(self): 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() @@ -3298,8 +3304,7 @@ def update_created_pools(self): 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. """ @@ -3907,9 +3912,9 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, @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: @@ -4717,7 +4722,7 @@ def _set_result(self, host, connection, pool, response): 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 012f52f954..cdd84fd013 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 @@ def _results(self): 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 48aa709821..d9bca62718 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -197,8 +197,8 @@ def resolve(self): 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 @@ class DefaultEndPointFactory(EndPointFactory): 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 @@ def _resolve_proxy_addresses(self): 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 @@ def resolve(self): 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 __init__(self, version, flags, stream, opcode, body_offset, end_pos): 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 @@ def is_checksumming_enabled(self): @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 @@ class Connection(object): # 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 @@ def _build_ssl_context_from_options(self): # 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 @@ def _wrap_socket_from_context(self): # 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 @@ def _handle_options_response(self, options_response): 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 @@ def _handle_options_response(self, options_response): # 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 7cde6765c0..eef0146d45 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 @@ def parse_casstype_args(typestring): # 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 @@ def __str__(self): return "EMPTY" __repr__ = __str__ + EMPTY = EmptyValue() @@ -397,6 +399,7 @@ def cass_parameterized_type(cls, full=False): 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 serialize(uuid, protocol_version): def serial_size(cls): return 16 + class BooleanType(_CassandraType): typename = 'boolean' @@ -480,6 +484,7 @@ def serialize(truth, protocol_version): def serial_size(cls): return 1 + class ByteType(_CassandraType): typename = 'tinyint' @@ -523,6 +528,7 @@ def serialize(byts, protocol_version): def serial_size(cls): return 4 + class DoubleType(_CassandraType): typename = 'double' @@ -538,6 +544,7 @@ def serialize(byts, protocol_version): def serial_size(cls): return 8 + class LongType(_CassandraType): typename = 'bigint' @@ -553,6 +560,7 @@ def serialize(byts, protocol_version): def serial_size(cls): return 8 + class Int32Type(_CassandraType): typename = 'int' @@ -568,6 +576,7 @@ def serialize(byts, protocol_version): def serial_size(cls): return 4 + class IntegerType(_CassandraType): typename = 'varint' @@ -610,6 +619,7 @@ def serialize(addr, protocol_version): class CounterColumnType(LongType): typename = 'counter' + cql_timestamp_formats = ( '%Y-%m-%d %H:%M', '%Y-%m-%d %H:%M:%S', @@ -667,6 +677,7 @@ def serialize(v, protocol_version): def serial_size(cls): return 8 + class TimestampType(DateType): pass @@ -692,6 +703,7 @@ def serialize(timeuuid, protocol_version): def serial_size(cls): return 16 + class SimpleDateType(_CassandraType): typename = 'date' date_format = "%Y-%m-%d" @@ -731,13 +743,14 @@ def deserialize(byts, protocol_version): 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 @@ def make_udt_class(cls, keyspace, udt_name, field_names, field_types): '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 @@ def evict_udt_class(cls, keyspace, udt_name): @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 used to match 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 @@ def serialize(cls, v, protocol_version): return buf.getvalue() + class VectorType(_CassandraType): typename = 'org.apache.cassandra.db.marshal.VectorType' vector_size = 0 @@ -1454,7 +1471,7 @@ def deserialize(cls, byts, protocol_version): 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 @@ def deserialize(cls, byts, protocol_version): 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 @@ def serialize(cls, v, protocol_version): 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 94093e85b6..3298ccf9de 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 @@ def cql_encode_sequence(self, val): 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 @@ def cql_encode_ipaddress(self, val): 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 e8733f0544..001d10f3f0 100644 --- a/cassandra/marshal.py +++ b/cassandra/marshal.py @@ -23,6 +23,7 @@ def _make_packer(format_string): 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 @@ def vints_unpack(term): # noqa return tuple(values) + def vints_pack(values): revbytes = bytearray() values = [int(v) for v in values[::-1]] @@ -127,7 +129,7 @@ def vints_pack(values): # 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 @@ def vints_pack(values): 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 @@ def uvint_pack(val): # 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 4c1be285b8..1f706996a0 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(self, endpoint_or_address, port=None): 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 @@ def __new__(metacls, name, bases, dct): return cls - class _ReplicationStrategy(object, metaclass=ReplicationStrategyTypeType): options_map = None @@ -436,9 +435,9 @@ def __init__(self, name, options_map): 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 @@ def make_token_replica_map(self, token_to_host_owner, ring): 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 @@ def export_as_string(self): 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 @@ def is_cql_compatible(self): 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 @@ class MD5Token(HashToken): 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): @@ -2188,8 +2187,8 @@ def _build_table_metadata(self, row, col_rows=None, trigger_rows=None): 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 @@ -2525,7 +2524,7 @@ def get_all_keyspaces(self): 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) @@ -2752,8 +2751,7 @@ class SchemaParserDSE60(SchemaParserV3): """ For DSE 6.0+ """ - recognized_table_options = (SchemaParserV3.recognized_table_options + - ("nodesync",)) + recognized_table_options = (SchemaParserV3.recognized_table_options + ("nodesync",)) class SchemaParserV4(SchemaParserV3): @@ -2899,8 +2897,7 @@ class SchemaParserDSE67(SchemaParserV4): """ For DSE 6.7+ """ - recognized_table_options = (SchemaParserV4.recognized_table_options + - ("nodesync",)) + recognized_table_options = (SchemaParserV4.recognized_table_options + ("nodesync",)) class SchemaParserDSE68(SchemaParserDSE67): @@ -2926,7 +2923,7 @@ def get_all_keyspaces(self): 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) @@ -3319,6 +3316,8 @@ def after_table_cql(cls, table_meta, ext_key, ext_blob): 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 a1eadc1fc4..a585000266 100644 --- a/cassandra/metrics.py +++ b/cassandra/metrics.py @@ -123,22 +123,24 @@ def __init__(self, cluster_proxy): 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'), - - # 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))) + 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))) # 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 282c43578d..cb80f27f38 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 @@ def _murmur3(data): return truncate_int64(h1) + try: from cassandra.cmurmur3 import murmur3 except ImportError: diff --git a/cassandra/policies.py b/cassandra/policies.py index d6f7063e7a..bc1861ab32 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 @@ def _rethrow(self, *args, **kwargs): 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 37fdaee96b..d060eb23e4 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -87,7 +87,7 @@ class Host(object): 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 @@ def get_state(self): 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 @@ def _wait_for_conn(self, timeout): 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_connections(self): 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 69340a805d..b1c4183cf4 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -54,6 +54,7 @@ class NotSupportedError(Exception): class InternalError(Exception): pass + ColumnMetadata = namedtuple("ColumnMetadata", ['keyspace_name', 'table_name', 'name', 'type']) HEADER_DIRECTION_TO_CLIENT = 0x80 @@ -1168,8 +1169,8 @@ def decode_message(cls, protocol_version, user_type_map, stream_id, flags, opcod :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 04c515d648..d359ccb6ff 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -85,6 +85,7 @@ def tuple_factory(colnames, rows): """ return rows + def named_tuple_factory(colnames, rows): """ Returns each row as a `namedtuple `_. @@ -245,8 +246,8 @@ def __init__(self, retry_policy=None, consistency_level=None, routing_key=None, 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 @@ def _get_serial_consistency_level(self): 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 @@ class PreparedStatement(object): A note about * in prepared statements """ - 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 @@ class HostTargetingStatement(object): 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 e2a2c1ea4c..6a958023fd 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 @@ def _maybe_warn(self, now): 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 408211ed05..8d7f72d3d7 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 @@ def min_uuid_from_time(timestamp): 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 @@ def uuid_from_time(time_arg, node=None, clock_seq=None): 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 @@ def _addrinfo_or_none(contact_point, port): """ 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 @@ def _remove(item, selfref=ref(self)): 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 _apply(self, other, method): def difference(self, other): return self._apply(other, self.data.difference) + __sub__ = difference def difference_update(self, other): @@ -368,6 +370,7 @@ def __isub__(self, other): def intersection(self, other): return self._apply(other, self.data.intersection) + __and__ = intersection def intersection_update(self, other): @@ -383,6 +386,7 @@ def __iand__(self, other): 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 __le__(self, other): 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 __eq__(self, other): 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 __ixor__(self, other): def union(self, other): return self._apply(other, self.data.union) + __or__ = union def isdisjoint(self, other): @@ -495,6 +502,7 @@ def __gt__(self, other): def __and__(self, other): return self._intersect(other) + __rand__ = __and__ def __iand__(self, other): @@ -504,6 +512,7 @@ def __iand__(self, other): def __or__(self, other): return self.union(other) + __ror__ = __or__ def __ior__(self, other): @@ -524,6 +533,7 @@ def __isub__(self, other): def __xor__(self, other): return self.symmetric_difference(other) + __rxor__ = __xor__ def __ixor__(self, other): @@ -636,8 +646,10 @@ def _find_insertion(self, x): 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 @@ def _find_insertion(self, x): 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 @@ class OrderedMap(Mapping): 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 @@ def _serialize_key(self, key): @total_ordering class Time(object): - ''' + """ Idealized time, independent of day. Up to nanosecond resolution - ''' + """ MICRO = 1000 MILLI = 1000 * MICRO @@ -861,9 +876,9 @@ def _from_timestring(self, s): 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 @@ def _from_timestring(self, s): 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 @@ def __str__(self): @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 @@ def _positional_rename_invalid_identifiers(field_names): 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 @@ class LineString(object): """ 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 @@ def from_wkt(s): 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 @@ def from_wkt(s): 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 @@ def __init__(self, months=0, days=0, nanoseconds=0): 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 __init__(self, value, precision): 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 validate(self): 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 @@ def __repr__(self): 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 + Cassandra's versions are assumed to correspond to major.minor.patch with an optional additional numeric build field as well as a string prerelease field. """ diff --git a/tests/__init__.py b/tests/__init__.py index 9b8dd89717..17a501fc94 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -39,7 +39,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 3b0103db31..9e72333a15 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 @@ def _get_dse_version_from_cass(cass_version): 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 @@ def get_supported_protocol_versions(): 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 @@ def _id_and_mark(f): 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') @@ -368,7 +370,8 @@ def _id_and_mark(f): 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'] @@ -421,10 +424,10 @@ def check_log_error(): 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(): @@ -860,15 +863,15 @@ def common_setup(cls, rf, keyspace_creation=True, create_class_table=False, **cl 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): @@ -894,7 +897,7 @@ def get_message_count(self, level, sub_string): 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): @@ -1047,6 +1050,7 @@ def __new__(cls, **kwargs): 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 @@ -1080,5 +1084,5 @@ def _get_config_val(self, k, v): 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 a4c4cdb4d8..ec7d279dfe 100644 --- a/tests/integration/datatype_utils.py +++ b/tests/integration/datatype_utils.py @@ -139,6 +139,7 @@ def get_sample_data(): return sample_data + SAMPLE_DATA = get_sample_data() diff --git a/tests/util.py b/tests/util.py index d44d6c91c8..5b4fd7f9fa 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 @@ def wrapped_condition(): return True, result attempt = 0 - while attempt < (max_attempts-1): + while attempt < (max_attempts - 1): attempt += 1 success, result = wrapped_condition() if success: