if 'eventlet.patcher' not in sys.modules:

return False

import eventlet.patcher

if 'gevent.monkey' not in sys.modules:

return False

import gevent.socket

"""

errors = None

"""

def __init__(self, message, errors):

Exception.__init__(self, message, errors)

""" Helper for run_in_executor() """

exc = future.exception()

if exc:

"""

@wraps(f)

def new_f(self, *args, **kwargs):

if self.is_shutdown:

return

try:

future = self.executor.submit(f, self, *args, **kwargs)

future.add_done_callback(_future_completed)

except Exception:

log.exception("Failed to submit task to executor")

load_balancing_policy = None

"""

retry_policy = None

"""

consistency_level = ConsistencyLevel.LOCAL_ONE

"""

serial_consistency_level = None

"""

request_timeout = 10.0

"""

row_factory = staticmethod(tuple_factory)

"""

speculative_execution_policy = None

"""

def __init__(self, load_balancing_policy=None, retry_policy=None,

consistency_level=ConsistencyLevel.LOCAL_ONE, serial_consistency_level=None,

request_timeout=10.0, row_factory=named_tuple_factory, speculative_execution_policy=None):

self.load_balancing_policy = load_balancing_policy or default_lbp_factory()

self.retry_policy = retry_policy or RetryPolicy()

self.consistency_level = consistency_level

self.serial_consistency_level = serial_consistency_level

self.request_timeout = request_timeout

self.row_factory = row_factory

def __init__(self):

self.profiles = dict()

def distance(self, host):

distances = set(p.load_balancing_policy.distance(host) for p in self.profiles.values())

return HostDistance.LOCAL if HostDistance.LOCAL in distances else \

HostDistance.REMOTE if HostDistance.REMOTE in distances else \

HostDistance.IGNORED

def populate(self, cluster, hosts):

for p in self.profiles.values():

p.load_balancing_policy.populate(cluster, hosts)

def check_supported(self):

for p in self.profiles.values():

p.load_balancing_policy.check_supported()

def on_up(self, host):

for p in self.profiles.values():

p.load_balancing_policy.on_up(host)

def on_down(self, host):

for p in self.profiles.values():

p.load_balancing_policy.on_down(host)

def on_add(self, host):

for p in self.profiles.values():

p.load_balancing_policy.on_add(host)

def on_remove(self, host):

for p in self.profiles.values():

p.load_balancing_policy.on_remove(host)

@property

def default(self):

"""

"""

contact_points = ['127.0.0.1']

"""

port = 9042

"""

cql_version = None

"""

protocol_version = ProtocolVersion.V4

"""

allow_beta_protocol_version = False

"""

compression = True

"""

_auth_provider = None

_auth_provider_callable = None

@property

def auth_provider(self):

"""

return self._auth_provider

@auth_provider.setter # noqa

def auth_provider(self, value):

if not value:

self._auth_provider = value

return

try:

self._auth_provider_callable = value.new_authenticator

except AttributeError:

raise TypeError("auth_provider must implement the cassandra.auth.AuthProvider interface")

self._auth_provider = value

_load_balancing_policy = None

@property

def load_balancing_policy(self):

"""

return self._load_balancing_policy

@load_balancing_policy.setter

def load_balancing_policy(self, lbp):

if self._config_mode == _ConfigMode.PROFILES:

raise ValueError("Cannot set Cluster.load_balancing_policy while using Configuration Profiles. Set this in a profile instead.")

self._load_balancing_policy = lbp

self._config_mode = _ConfigMode.LEGACY

@property

def _default_load_balancing_policy(self):

return self.profile_manager.default.load_balancing_policy

reconnection_policy = ExponentialReconnectionPolicy(1.0, 600.0)

"""

_default_retry_policy = RetryPolicy()

@property

def default_retry_policy(self):

"""

return self._default_retry_policy

@default_retry_policy.setter

def default_retry_policy(self, policy):

if self._config_mode == _ConfigMode.PROFILES:

raise ValueError("Cannot set Cluster.default_retry_policy while using Configuration Profiles. Set this in a profile instead.")

self._default_retry_policy = policy

self._config_mode = _ConfigMode.LEGACY

conviction_policy_factory = SimpleConvictionPolicy

"""

address_translator = IdentityTranslator()

"""

connect_to_remote_hosts = True

"""

metrics_enabled = False

"""

metrics = None

"""

ssl_options = None

"""

sockopts = None

"""

max_schema_agreement_wait = 10

"""

metadata = None

"""

connection_class = DefaultConnection

"""

control_connection_timeout = 2.0

"""

idle_heartbeat_interval = 30

"""

idle_heartbeat_timeout = 30

"""

schema_event_refresh_window = 2

"""

topology_event_refresh_window = 10

"""

status_event_refresh_window = 2

"""

prepare_on_all_hosts = True

"""

reprepare_on_up = True

"""

connect_timeout = 5

"""

timestamp_generator = None

"""

@property

def schema_metadata_enabled(self):

"""

return self.control_connection._schema_meta_enabled

@schema_metadata_enabled.setter

def schema_metadata_enabled(self, enabled):

self.control_connection._schema_meta_enabled = bool(enabled)

@property

def token_metadata_enabled(self):

"""

return self.control_connection._token_meta_enabled

@token_metadata_enabled.setter

def token_metadata_enabled(self, enabled):

self.control_connection._token_meta_enabled = bool(enabled)

profile_manager = None

_config_mode = _ConfigMode.UNCOMMITTED

sessions = None

control_connection = None

scheduler = None

executor = None

is_shutdown = False

_is_setup = False

_prepared_statements = None

_prepared_statement_lock = None

_idle_heartbeat = None

_protocol_version_explicit = False

_discount_down_events = True

_user_types = None

"""

_listeners = None

_listener_lock = None

def __init__(self,

contact_points=_UNSET_ARG,

port=9042,

compression=True,

auth_provider=None,

load_balancing_policy=None,

reconnection_policy=None,

default_retry_policy=None,

conviction_policy_factory=None,

metrics_enabled=False,

connection_class=None,

ssl_options=None,

sockopts=None,

cql_version=None,

protocol_version=_NOT_SET,

executor_threads=2,

max_schema_agreement_wait=10,

control_connection_timeout=2.0,

idle_heartbeat_interval=30,

schema_event_refresh_window=2,

topology_event_refresh_window=10,

connect_timeout=5,

schema_metadata_enabled=True,

token_metadata_enabled=True,

address_translator=None,

status_event_refresh_window=2,

prepare_on_all_hosts=True,

reprepare_on_up=True,

execution_profiles=None,

allow_beta_protocol_version=False,

timestamp_generator=None,

idle_heartbeat_timeout=30):

"""

if contact_points is not _UNSET_ARG and load_balancing_policy is None:

log.warn('Cluster.__init__ called with contact_points specified, '

'but no load_balancing_policy. In the next major '

'version, this will raise an error; please specify a '

'load balancing policy. '

'(contact_points = {cp}, lbp = {lbp}'

''.format(cp=contact_points, lbp=load_balancing_policy))

if contact_points is not None:

if contact_points is _UNSET_ARG:

contact_points = ['127.0.0.1']

if isinstance(contact_points, six.string_types):

raise TypeError("contact_points should not be a string, it should be a sequence (e.g. list) of strings")

if None in contact_points:

raise ValueError("contact_points should not contain None (it can resolve to localhost)")

self.contact_points = contact_points

self.port = port

self.contact_points_resolved = [endpoint[4][0] for a in self.contact_points

for endpoint in socket.getaddrinfo(a, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM)]

self.compression = compression

if protocol_version is not _NOT_SET:

self.protocol_version = protocol_version

self._protocol_version_explicit = True

self.allow_beta_protocol_version = allow_beta_protocol_version

self.auth_provider = auth_provider

if load_balancing_policy is not None:

if isinstance(load_balancing_policy, type):

raise TypeError("load_balancing_policy should not be a class, it should be an instance of that class")

self.load_balancing_policy = load_balancing_policy

else:

self._load_balancing_policy = default_lbp_factory() # set internal attribute to avoid committing to legacy config mode

if reconnection_policy is not None:

if isinstance(reconnection_policy, type):

raise TypeError("reconnection_policy should not be a class, it should be an instance of that class")

self.reconnection_policy = reconnection_policy

if default_retry_policy is not None:

if isinstance(default_retry_policy, type):

raise TypeError("default_retry_policy should not be a class, it should be an instance of that class")

self.default_retry_policy = default_retry_policy

if conviction_policy_factory is not None:

if not callable(conviction_policy_factory):

raise ValueError("conviction_policy_factory must be callable")

self.conviction_policy_factory = conviction_policy_factory

if address_translator is not None:

if isinstance(address_translator, type):

raise TypeError("address_translator should not be a class, it should be an instance of that class")

self.address_translator = address_translator

if connection_class is not None:

self.connection_class = connection_class

if timestamp_generator is not None:

if not callable(timestamp_generator):

raise ValueError("timestamp_generator must be callable")

self.timestamp_generator = timestamp_generator

else:

self.timestamp_generator = MonotonicTimestampGenerator()

self.profile_manager = ProfileManager()

self.profile_manager.profiles[EXEC_PROFILE_DEFAULT] = ExecutionProfile(self.load_balancing_policy,

self.default_retry_policy,

Session._default_consistency_level,

Session._default_serial_consistency_level,

Session._default_timeout,

Session._row_factory)

# legacy mode if either of these is not default

if load_balancing_policy or default_retry_policy:

if execution_profiles:

raise ValueError("Clusters constructed with execution_profiles should not specify legacy parameters "

"load_balancing_policy or default_retry_policy. Configure this in a profile instead.")

self._config_mode = _ConfigMode.LEGACY

else:

if execution_profiles:

self.profile_manager.profiles.update(execution_profiles)

self._config_mode = _ConfigMode.PROFILES

self.metrics_enabled = metrics_enabled

self.ssl_options = ssl_options

self.sockopts = sockopts

self.cql_version = cql_version

self.max_schema_agreement_wait = max_schema_agreement_wait

self.control_connection_timeout = control_connection_timeout

self.idle_heartbeat_interval = idle_heartbeat_interval

self.idle_heartbeat_timeout = idle_heartbeat_timeout

self.schema_event_refresh_window = schema_event_refresh_window

self.topology_event_refresh_window = topology_event_refresh_window

self.status_event_refresh_window = status_event_refresh_window

self.connect_timeout = connect_timeout

self.prepare_on_all_hosts = prepare_on_all_hosts

self.reprepare_on_up = reprepare_on_up

self._listeners = set()

self._listener_lock = Lock()

# let Session objects be GC'ed (and shutdown) when the user no longer

# holds a reference.

self.sessions = WeakSet()

self.metadata = Metadata()

self.control_connection = None

self._prepared_statements = WeakValueDictionary()

self._prepared_statement_lock = Lock()

self._user_types = defaultdict(dict)

self.executor = ThreadPoolExecutor(max_workers=executor_threads)

self.scheduler = _Scheduler(self.executor)

self._lock = RLock()

if self.metrics_enabled:

from cassandra.metrics import Metrics

self.metrics = Metrics(weakref.proxy(self))

self.control_connection = ControlConnection(

self, self.control_connection_timeout,

self.schema_event_refresh_window, self.topology_event_refresh_window,

self.status_event_refresh_window,

schema_metadata_enabled, token_metadata_enabled)

def register_user_type(self, keyspace, user_type, klass):

"""

self._user_types[keyspace][user_type] = klass

for session in tuple(self.sessions):

session.user_type_registered(keyspace, user_type, klass)

UserType.evict_udt_class(keyspace, user_type)

def add_execution_profile(self, name, profile, pool_wait_timeout=5):

"""

if not isinstance(profile, ExecutionProfile):

raise TypeError("profile must be an instance of ExecutionProfile")

if self._config_mode == _ConfigMode.LEGACY:

raise ValueError("Cannot add execution profiles when legacy parameters are set explicitly.")

if name in self.profile_manager.profiles:

raise ValueError("Profile %s already exists")

self.profile_manager.profiles[name] = profile

profile.load_balancing_policy.populate(self, self.metadata.all_hosts())

# on_up after populate allows things like DCA LBP to choose default local dc

for host in filter(lambda h: h.is_up, self.metadata.all_hosts()):

profile.load_balancing_policy.on_up(host)

futures = set()

for session in tuple(self.sessions):

futures.update(session.update_created_pools())

_, not_done = wait_futures(futures, pool_wait_timeout)

if not_done:

raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout.")

def connection_factory(self, address, *args, **kwargs):

"""

kwargs = self._make_connection_kwargs(address, kwargs)

return self.connection_class.factory(address, self.connect_timeout, *args, **kwargs)

def _make_connection_factory(self, host, *args, **kwargs):

kwargs = self._make_connection_kwargs(host.address, kwargs)

return partial(self.connection_class.factory, host.address, self.connect_timeout, *args, **kwargs)

def _make_connection_kwargs(self, address, kwargs_dict):

if self._auth_provider_callable:

kwargs_dict.setdefault('authenticator', self._auth_provider_callable(address))

kwargs_dict.setdefault('port', self.port)

kwargs_dict.setdefault('compression', self.compression)

kwargs_dict.setdefault('sockopts', self.sockopts)

kwargs_dict.setdefault('ssl_options', self.ssl_options)

kwargs_dict.setdefault('cql_version', self.cql_version)

kwargs_dict.setdefault('protocol_version', self.protocol_version)

kwargs_dict.setdefault('user_type_map', self._user_types)

kwargs_dict.setdefault('allow_beta_protocol_version', self.allow_beta_protocol_version)

return kwargs_dict

def protocol_downgrade(self, host_addr, previous_version):

if self._protocol_version_explicit:

raise DriverException("ProtocolError returned from server while using explicitly set client protocol_version %d" % (previous_version,))