This is an automated email from the ASF dual-hosted git repository.
absurdfarce pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-python-driver.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8cd395bb PEP-8 code style changes
8cd395bb is described below
commit 8cd395bb0b816beacdec61da8938e9224fdeef4f
Author: Brad Schoening <[email protected]>
AuthorDate: Tue Jul 14 16:44:23 2026 -0400
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
---
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 c7327086..f0d47c56 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 @@ class Unavailable(RequestExecutionException):
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 86759afe..9aa4d2c3 100644
--- a/cassandra/auth.py
+++ b/cassandra/auth.py
@@ -215,6 +215,7 @@ class SaslAuthenticator(Authenticator):
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 10ce3de1..f2e11189 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -93,21 +93,24 @@ try:
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 @@ 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
@@ -378,8 +382,8 @@ class ExecutionProfile(object):
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 @@ class Cluster(object):
self._auth_provider = value
_load_balancing_policy = None
+
@property
def load_balancing_policy(self):
"""
@@ -706,6 +711,7 @@ class Cluster(object):
"""
_default_retry_policy = RetryPolicy()
+
@property
def default_retry_policy(self):
"""
@@ -763,9 +769,8 @@ class Cluster(object):
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 @@ class Session(object):
_monitor_reporter = None
_row_factory = staticmethod(named_tuple_factory)
+
@property
def row_factory(self):
"""
@@ -2305,8 +2311,8 @@ class Session(object):
*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 @@ class Session(object):
@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 @@ class Session(object):
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 @@ class Session(object):
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 @@ class Session(object):
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 @@ class ControlConnection(object):
@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 @@ class ResponseFuture(object):
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 012f52f9..cdd84fd0 100644
--- a/cassandra/concurrent.py
+++ b/cassandra/concurrent.py
@@ -29,7 +29,9 @@ log = logging.getLogger(__name__)
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 @@ class ConcurrentExecutorListResults(_ConcurrentExecutor):
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 48aa7098..d9bca627 100644
--- a/cassandra/connection.py
+++ b/cassandra/connection.py
@@ -197,8 +197,8 @@ class DefaultEndPoint(EndPoint):
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 @@ class SniEndPoint(EndPoint):
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 @@ class UnixSocketEndPoint(EndPoint):
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 @@ class _Frame(object):
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 @@ class _ConnectionIOBuffer(object):
@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 @@ class Connection(object):
# 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 @@ class Connection(object):
# 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 @@ class Connection(object):
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 @@ class Connection(object):
# 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 7cde6765..9c76955e 100644
--- a/cassandra/cqltypes.py
+++ b/cassandra/cqltypes.py
@@ -35,6 +35,7 @@ import calendar
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.marshal import (int8_pack, int8_unpack,
int16_pack, int16_unpack,
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 @@ class EmptyValue(object):
return "EMPTY"
__repr__ = __str__
+
EMPTY = EmptyValue()
@@ -397,6 +399,7 @@ class _CassandraType(object, metaclass=CassandraTypeType):
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 @@ class UUIDType(_CassandraType):
def serial_size(cls):
return 16
+
class BooleanType(_CassandraType):
typename = 'boolean'
@@ -480,6 +484,7 @@ class BooleanType(_CassandraType):
def serial_size(cls):
return 1
+
class ByteType(_CassandraType):
typename = 'tinyint'
@@ -523,6 +528,7 @@ class FloatType(_CassandraType):
def serial_size(cls):
return 4
+
class DoubleType(_CassandraType):
typename = 'double'
@@ -538,6 +544,7 @@ class DoubleType(_CassandraType):
def serial_size(cls):
return 8
+
class LongType(_CassandraType):
typename = 'bigint'
@@ -553,6 +560,7 @@ class LongType(_CassandraType):
def serial_size(cls):
return 8
+
class Int32Type(_CassandraType):
typename = 'int'
@@ -568,6 +576,7 @@ class Int32Type(_CassandraType):
def serial_size(cls):
return 4
+
class IntegerType(_CassandraType):
typename = 'varint'
@@ -610,6 +619,7 @@ class InetAddressType(_CassandraType):
class CounterColumnType(LongType):
typename = 'counter'
+
cql_timestamp_formats = (
'%Y-%m-%d %H:%M',
'%Y-%m-%d %H:%M:%S',
@@ -667,6 +677,7 @@ class DateType(_CassandraType):
def serial_size(cls):
return 8
+
class TimestampType(DateType):
pass
@@ -692,6 +703,7 @@ class TimeUUIDType(DateType):
def serial_size(cls):
return 16
+
class SimpleDateType(_CassandraType):
typename = 'date'
date_format = "%Y-%m-%d"
@@ -731,13 +743,14 @@ class ShortType(_CassandraType):
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 @@ class UserType(TupleType):
'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 @@ class UserType(TupleType):
@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 @@ class DateRangeType(CassandraType):
return buf.getvalue()
+
class VectorType(_CassandraType):
typename = 'org.apache.cassandra.db.marshal.VectorType'
vector_size = 0
@@ -1454,7 +1471,7 @@ class VectorType(_CassandraType):
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 @@ class VectorType(_CassandraType):
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 @@ class VectorType(_CassandraType):
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 94093e85..3298ccf9 100644
--- a/cassandra/encoder.py
+++ b/cassandra/encoder.py
@@ -20,7 +20,6 @@ called on each query parameter.
"""
import logging
-log = logging.getLogger(__name__)
from binascii import hexlify
from decimal import Decimal
@@ -35,6 +34,8 @@ import ipaddress
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 @@ class Encoder(object):
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 @@ class Encoder(object):
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 e8733f05..001d10f3 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 86464415..94c7476d 100644
--- a/cassandra/metadata.py
+++ b/cassandra/metadata.py
@@ -28,12 +28,6 @@ from threading import RLock
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.util import OrderedDict, Version
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 @@ class Metadata(object):
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 @@ class ReplicationStrategyTypeType(type):
return cls
-
class _ReplicationStrategy(object, metaclass=ReplicationStrategyTypeType):
options_map = None
@@ -436,9 +435,9 @@ class _UnknownStrategy(ReplicationStrategy):
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 @@ class NetworkTopologyStrategy(ReplicationStrategy):
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 @@ class KeyspaceMetadata(object):
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 @@ class TableMetadata(object):
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):
@@ -2185,8 +2184,8 @@ class SchemaParserV22(_SchemaParser):
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 @@ class SchemaParserV3(SchemaParserV22):
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 @@ 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):
@@ -2896,8 +2894,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):
@@ -2923,7 +2920,7 @@ class SchemaParserDSE68(SchemaParserDSE67):
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 @@ class RLACTableExtension(RegisteredTableExtension):
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 a1eadc1f..a5850002 100644
--- a/cassandra/metrics.py
+++ b/cassandra/metrics.py
@@ -123,22 +123,24 @@ class Metrics(object):
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 282c4357..cb80f27f 100644
--- a/cassandra/murmur3.py
+++ b/cassandra/murmur3.py
@@ -2,15 +2,15 @@ import struct
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 d6f7063e..bc1861ab 100644
--- a/cassandra/policies.py
+++ b/cassandra/policies.py
@@ -22,17 +22,16 @@ from random import randint, shuffle
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 @@ class NeverRetryPolicy(RetryPolicy):
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 37fdaee9..d060eb23 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 @@ class HostConnection(object):
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 @@ class HostConnectionPool(object):
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 @@ class HostConnectionPool(object):
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 69340a80..b1c4183c 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 @@ class _ProtocolHandler(object):
: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 04c515d6..d359ccb6 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
<https://docs.python.org/2/library/collections.html#collections.namedtuple>`_.
@@ -245,8 +246,8 @@ class Statement(object):
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 @@ class Statement(object):
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):
<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 @@ 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 e2a2c1ea..6a958023 100644
--- a/cassandra/timestamps.py
+++ b/cassandra/timestamps.py
@@ -25,6 +25,7 @@ from threading import Lock
log = logging.getLogger(__name__)
+
class MonotonicTimestampGenerator(object):
"""
An object that, when called, returns ``int(time.time() * 1e6)`` when
@@ -98,9 +99,9 @@ class MonotonicTimestampGenerator(object):
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 408211ed..f460396e 100644
--- a/cassandra/util.py
+++ b/cassandra/util.py
@@ -31,15 +31,14 @@ import sys
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 @@ class WeakSet(object):
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 @@ class WeakSet(object):
def difference(self, other):
return self._apply(other, self.data.difference)
+
__sub__ = difference
def difference_update(self, other):
@@ -368,6 +370,7 @@ class WeakSet(object):
def intersection(self, other):
return self._apply(other, self.data.intersection)
+
__and__ = intersection
def intersection_update(self, other):
@@ -383,6 +386,7 @@ class WeakSet(object):
def issubset(self, other):
return self.data.issubset(ref(item) for item in other)
+
__lt__ = issubset
def __le__(self, other):
@@ -390,6 +394,7 @@ class WeakSet(object):
def issuperset(self, other):
return self.data.issuperset(ref(item) for item in other)
+
__gt__ = issuperset
def __ge__(self, other):
@@ -402,6 +407,7 @@ class WeakSet(object):
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 @@ class WeakSet(object):
def union(self, other):
return self._apply(other, self.data.union)
+
__or__ = union
def isdisjoint(self, other):
@@ -495,6 +502,7 @@ class SortedSet(object):
def __and__(self, other):
return self._intersect(other)
+
__rand__ = __and__
def __iand__(self, other):
@@ -504,6 +512,7 @@ class SortedSet(object):
def __or__(self, other):
return self.union(other)
+
__ror__ = __or__
def __ior__(self, other):
@@ -524,6 +533,7 @@ class SortedSet(object):
def __xor__(self, other):
return self.symmetric_difference(other)
+
__rxor__ = __xor__
def __ixor__(self, other):
@@ -636,8 +646,10 @@ class SortedSet(object):
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 @@ class SortedSet(object):
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 @@ class OrderedMapSerializedKey(OrderedMap):
@total_ordering
class Time(object):
- '''
+ """
Idealized time, independent of day.
Up to nanosecond resolution
- '''
+ """
MICRO = 1000
MILLI = 1000 * MICRO
@@ -861,9 +876,9 @@ class Time(object):
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 @@ class Time(object):
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 @@ class Time(object):
@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 @@ class LineString(object):
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 @@ class Polygon(object):
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 @@ class Duration(object):
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 @@ class DateRangeBound(object):
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 @@ class DateRange(object):
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 @@ class DateRange(object):
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 047da82c..e9ee8546 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -37,7 +37,7 @@ EVENT_LOOP_MANAGER = os.getenv('EVENT_LOOP_MANAGER', "libev")
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 68593f3c..a9da9100 100644
--- a/tests/integration/__init__.py
+++ b/tests/integration/__init__.py
@@ -61,7 +61,7 @@ MULTIDC_CLUSTER_NAME = 'multidc_test_cluster'
#
# 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 local_decorator_creator():
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 @@ notdse = unittest.skipIf(DSE_VERSION, "DSE not supported")
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 @@ 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():
@@ -859,15 +862,15 @@ class BasicKeyspaceUnitTestCase(unittest.TestCase):
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 @@ class MockLoggingHandler(logging.Handler):
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 @@ class TestCluster(object):
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 @@ class Cassandra41CCMCluster(CCMCluster):
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 a4c4cdb4..ec7d279d 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 d44d6c91..5b4fd7f9 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 wait_until_not_raised(condition, delay, max_attempts):
return True, result
attempt = 0
- while attempt < (max_attempts-1):
+ while attempt < (max_attempts - 1):
attempt += 1
success, result = wrapped_condition()
if success:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
