# Copyright 2013-2016 DataStax, Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# 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.

"""

This module houses the main classes you will interact with,

:class:`.Cluster` and :class:`.Session`.

"""

from __future__ import absolute_import

import atexit

from collections import defaultdict, Mapping

from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait as wait_futures

from copy import copy

from functools import partial, wraps

from itertools import groupby, count

import logging

from random import random

import six

from six.moves import filter, range, queue as Queue

import socket

import sys

import time

from threading import Lock, RLock, Thread, Event

import weakref

from weakref import WeakValueDictionary

try:

from weakref import WeakSet

except ImportError:

from cassandra.util import WeakSet # NOQA

from cassandra import (ConsistencyLevel, AuthenticationFailed,

OperationTimedOut, UnsupportedOperation,

SchemaTargetType, DriverException, ProtocolVersion)

from cassandra.connection import (ConnectionException, ConnectionShutdown,

ConnectionHeartbeat, ProtocolVersionUnsupported)

from cassandra.cqltypes import UserType

from cassandra.encoder import Encoder

from cassandra.protocol import (QueryMessage, ResultMessage,

ErrorMessage, ReadTimeoutErrorMessage,

WriteTimeoutErrorMessage,

UnavailableErrorMessage,

OverloadedErrorMessage,

PrepareMessage, ExecuteMessage,

PreparedQueryNotFound,

IsBootstrappingErrorMessage,

BatchMessage, RESULT_KIND_PREPARED,

RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS,

RESULT_KIND_SCHEMA_CHANGE, ProtocolHandler)

from cassandra.metadata import Metadata, protect_name, murmur3

from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy,

ExponentialReconnectionPolicy, HostDistance,

RetryPolicy, IdentityTranslator, NoSpeculativeExecutionPlan,

NoSpeculativeExecutionPolicy)

from cassandra.pool import (Host, _ReconnectionHandler, _HostReconnectionHandler,

HostConnectionPool, HostConnection,

NoConnectionsAvailable)

from cassandra.query import (SimpleStatement, PreparedStatement, BoundStatement,

BatchStatement, bind_params, QueryTrace,

named_tuple_factory, dict_factory, tuple_factory, FETCH_SIZE_UNSET)

def _is_eventlet_monkey_patched():

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

return False

import eventlet.patcher

return eventlet.patcher.is_monkey_patched('socket')

def _is_gevent_monkey_patched():

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

return False

import gevent.socket

return socket.socket is gevent.socket.socket

# default to gevent when we are monkey patched with gevent, eventlet when

# monkey patched with eventlet, otherwise if libev is available, use that as

# the default because it's fastest. Otherwise, use asyncore.

if _is_gevent_monkey_patched():

from cassandra.io.geventreactor import GeventConnection as DefaultConnection

elif _is_eventlet_monkey_patched():

from cassandra.io.eventletreactor import EventletConnection as DefaultConnection

else:

try:

from cassandra.io.libevreactor import LibevConnection as DefaultConnection # NOQA

except ImportError:

from cassandra.io.asyncorereactor import AsyncoreConnection as DefaultConnection # NOQA

# Forces load of utf8 encoding module to avoid deadlock that occurs

# if code that is being imported tries to import the module in a seperate

# thread.

# See http://bugs.python.org/issue10923

"".encode('utf8')

log = logging.getLogger(__name__)

DEFAULT_MIN_REQUESTS = 5

DEFAULT_MAX_REQUESTS = 100

DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST = 2

DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST = 8

DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST = 1

DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST = 2

_NOT_SET = object()

class NoHostAvailable(Exception):

"""

Raised when an operation is attempted but all connections are

busy, defunct, closed, or resulted in errors when used.

"""

errors = None

"""

A map of the form ``{ip: exception}`` which details the particular

Exception that was caught for each host the operation was attempted

against.

"""

def __init__(self, message, errors):

Exception.__init__(self, message, errors)

self.errors = errors

def _future_completed(future):

""" Helper for run_in_executor() """

exc = future.exception()

if exc:

log.debug("Failed to run task on executor", exc_info=exc)

def run_in_executor(f):

"""

A decorator to run the given method in the ThreadPoolExecutor.

"""

@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")

return new_f

_clusters_for_shutdown = set()

def _register_cluster_shutdown(cluster):

_clusters_for_shutdown.add(cluster)

def _discard_cluster_shutdown(cluster):

_clusters_for_shutdown.discard(cluster)

def _shutdown_clusters():

clusters = _clusters_for_shutdown.copy() # copy because shutdown modifies the global set "discard"

for cluster in clusters:

cluster.shutdown()

atexit.register(_shutdown_clusters)

def default_lbp_factory():

if murmur3 is not None:

return TokenAwarePolicy(DCAwareRoundRobinPolicy())

return DCAwareRoundRobinPolicy()

class ExecutionProfile(object):

load_balancing_policy = None

"""

An instance of :class:`.policies.LoadBalancingPolicy` or one of its subclasses.

Used in determining host distance for establishing connections, and routing requests.

Defaults to ``TokenAwarePolicy(DCAwareRoundRobinPolicy())`` if not specified

"""

retry_policy = None

"""

An instance of :class:`.policies.RetryPolicy` instance used when :class:`.Statement` objects do not have a

:attr:`~.Statement.retry_policy` explicitly set.

Defaults to :class:`.RetryPolicy` if not specified

"""

consistency_level = ConsistencyLevel.LOCAL_ONE

"""

:class:`.ConsistencyLevel` used when not specified on a :class:`.Statement`.

"""

serial_consistency_level = None

"""

Serial :class:`.ConsistencyLevel` used when not specified on a :class:`.Statement` (for LWT conditional statements).

"""

request_timeout = 10.0

"""

Request timeout used when not overridden in :meth:`.Session.execute`

"""

row_factory = staticmethod(tuple_factory)

"""

A callable to format results, accepting ``(colnames, rows)`` where ``colnames`` is a list of column names, and

``rows`` is a list of tuples, with each tuple representing a row of parsed values.

Some example implementations:

- :func:`cassandra.query.tuple_factory` - return a result row as a tuple

- :func:`cassandra.query.named_tuple_factory` - return a result row as a named tuple

- :func:`cassandra.query.dict_factory` - return a result row as a dict

- :func:`cassandra.query.ordered_dict_factory` - return a result row as an OrderedDict

"""

speculative_execution_policy = None

"""

An instance of :class:`.policies.SpeculativeExecutionPolicy`

Defaults to :class:`.NoSpeculativeExecutionPolicy` if not specified

"""

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

self.speculative_execution_policy = speculative_execution_policy or NoSpeculativeExecutionPolicy()

class ProfileManager(object):

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):

"""

internal-only; no checks are done because this entry is populated on cluster init

"""

return self.profiles[EXEC_PROFILE_DEFAULT]

EXEC_PROFILE_DEFAULT = object()

"""

Key for the ``Cluster`` default execution profile, used when no other profile is selected in

``Session.execute(execution_profile)``.

Use this as the key in ``Cluster(execution_profiles)`` to override the default profile.

"""

class _ConfigMode(object):

UNCOMMITTED = 0

LEGACY = 1

PROFILES = 2

class Cluster(object):

"""

The main class to use when interacting with a Cassandra cluster.

Typically, one instance of this class will be created for each

separate Cassandra cluster that your application interacts with.

Example usage::

>>> from cassandra.cluster import Cluster

>>> cluster = Cluster(['192.168.1.1', '192.168.1.2'])

>>> session = cluster.connect()

>>> session.execute("CREATE KEYSPACE ...")

>>> ...

>>> cluster.shutdown()

``Cluster`` and ``Session`` also provide context management functions

which implicitly handle shutdown when leaving scope.

"""

contact_points = ['127.0.0.1']

"""

The list of contact points to try connecting for cluster discovery.

Defaults to loopback interface.

Note: When using :class:`.DCAwareLoadBalancingPolicy` with no explicit

local_dc set (as is the default), the DC is chosen from an arbitrary

host in contact_points. In this case, contact_points should contain

only nodes from a single, local DC.

"""

port = 9042

"""

The server-side port to open connections to. Defaults to 9042.

"""

cql_version = None

"""

If a specific version of CQL should be used, this may be set to that

string version. Otherwise, the highest CQL version supported by the

server will be automatically used.

"""

protocol_version = ProtocolVersion.V4

"""

The maximum version of the native protocol to use.

See :class:`.ProtocolVersion` for more information about versions.

If not set in the constructor, the driver will automatically downgrade

version based on a negotiation with the server, but it is most efficient

to set this to the maximum supported by your version of Cassandra.

Setting this will also prevent conflicting versions negotiated if your

cluster is upgraded.

"""

allow_beta_protocol_version = False

"""

Setting true injects a flag in all messages that makes the server accept and use "beta" protocol version.

Used for testing new protocol features incrementally before the new version is complete.

"""

compression = True

"""

Controls compression for communications between the driver and Cassandra.

If left as the default of :const:`True`, either lz4 or snappy compression

may be used, depending on what is supported by both the driver

and Cassandra. If both are fully supported, lz4 will be preferred.

You may also set this to 'snappy' or 'lz4' to request that specific

compression type.

Setting this to :const:`False` disables compression.

"""

_auth_provider = None

_auth_provider_callable = None

@property

def auth_provider(self):

"""

When :attr:`~.Cluster.protocol_version` is 2 or higher, this should

be an instance of a subclass of :class:`~cassandra.auth.AuthProvider`,

such as :class:`~.PlainTextAuthProvider`.

When :attr:`~.Cluster.protocol_version` is 1, this should be

a function that accepts one argument, the IP address of a node,

and returns a dict of credentials for that node.

When not using authentication, this should be left as :const:`None`.

"""

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:

if self.protocol_version > 1:

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

"interface when protocol_version >= 2")

elif not callable(value):

raise TypeError("auth_provider must be callable when protocol_version == 1")

self._auth_provider_callable = value

self._auth_provider = value

_load_balancing_policy = None

@property

def load_balancing_policy(self):

"""

An instance of :class:`.policies.LoadBalancingPolicy` or

one of its subclasses.

.. versionchanged:: 2.6.0

Defaults to :class:`~.TokenAwarePolicy` (:class:`~.DCAwareRoundRobinPolicy`).

when using CPython (where the murmur3 extension is available). :class:`~.DCAwareRoundRobinPolicy`

otherwise. Default local DC will be chosen from contact points.

**Please see** :class:`~.DCAwareRoundRobinPolicy` **for a discussion on default behavior with respect to

DC locality and remote nodes.**

"""

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)

"""

An instance of :class:`.policies.ReconnectionPolicy`. Defaults to an instance

of :class:`.ExponentialReconnectionPolicy` with a base delay of one second and

a max delay of ten minutes.

"""

_default_retry_policy = RetryPolicy()

@property

def default_retry_policy(self):

"""

A default :class:`.policies.RetryPolicy` instance to use for all

:class:`.Statement` objects which do not have a :attr:`~.Statement.retry_policy`

explicitly set.

"""

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

"""

A factory function which creates instances of

:class:`.policies.ConvictionPolicy`. Defaults to

:class:`.policies.SimpleConvictionPolicy`.

"""

address_translator = IdentityTranslator()

"""

:class:`.policies.AddressTranslator` instance to be used in translating server node addresses

to driver connection addresses.

"""

connect_to_remote_hosts = True

"""

If left as :const:`True`, hosts that are considered :attr:`~.HostDistance.REMOTE`

by the :attr:`~.Cluster.load_balancing_policy` will have a connection

opened to them. Otherwise, they will not have a connection opened to them.

Note that the default load balancing policy ignores remote hosts by default.

.. versionadded:: 2.1.0

"""

metrics_enabled = False

"""

Whether or not metric collection is enabled. If enabled, :attr:`.metrics`

will be an instance of :class:`~cassandra.metrics.Metrics`.

"""

metrics = None

"""

An instance of :class:`cassandra.metrics.Metrics` if :attr:`.metrics_enabled` is

:const:`True`, else :const:`None`.

"""

ssl_options = None

"""

A optional dict which will be used as kwargs for ``ssl.wrap_socket()``

when new sockets are created. This should be used when client encryption

is enabled in Cassandra.

By default, a ``ca_certs`` value should be supplied (the value should be

a string pointing to the location of the CA certs file), and you probably

want to specify ``ssl_version`` as ``ssl.PROTOCOL_TLSv1`` to match

Cassandra's default protocol.

.. versionchanged:: 3.3.0

In addition to ``wrap_socket`` kwargs, clients may also specify ``'check_hostname': True`` to verify the cert hostname

as outlined in RFC 2818 and RFC 6125. Note that this requires the certificate to be transferred, so

should almost always require the option ``'cert_reqs': ssl.CERT_REQUIRED``. Note also that this functionality was not built into

Python standard library until (2.7.9, 3.2). To enable this mechanism in earlier versions, patch ``ssl.match_hostname``

with a custom or `back-ported function <https://pypi.python.org/pypi/backports.ssl_match_hostname>`_.

"""

sockopts = None

"""

An optional list of tuples which will be used as arguments to

``socket.setsockopt()`` for all created sockets.

Note: some drivers find setting TCPNODELAY beneficial in the context of

their execution model. It was not found generally beneficial for this driver.

To try with your own workload, set ``sockopts = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``

"""

max_schema_agreement_wait = 10

"""

The maximum duration (in seconds) that the driver will wait for schema

agreement across the cluster. Defaults to ten seconds.

If set <= 0, the driver will bypass schema agreement waits altogether.

"""

metadata = None

"""

An instance of :class:`cassandra.metadata.Metadata`.

"""

connection_class = DefaultConnection

"""

This determines what event loop system will be used for managing

I/O with Cassandra. These are the current options:

* :class:`cassandra.io.asyncorereactor.AsyncoreConnection`

* :class:`cassandra.io.libevreactor.LibevConnection`

* :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)

* :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)

* :class:`cassandra.io.twistedreactor.TwistedConnection`

By default, ``AsyncoreConnection`` will be used, which uses

the ``asyncore`` module in the Python standard library.

If ``libev`` is installed, ``LibevConnection`` will be used instead.

If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding

connection class will be used automatically.

"""

control_connection_timeout = 2.0

"""

A timeout, in seconds, for queries made by the control connection, such

as querying the current schema and information about nodes in the cluster.

If set to :const:`None`, there will be no timeout for these queries.

"""

idle_heartbeat_interval = 30

"""

Interval, in seconds, on which to heartbeat idle connections. This helps

keep connections open through network devices that expire idle connections.

It also helps discover bad connections early in low-traffic scenarios.

Setting to zero disables heartbeats.

"""

schema_event_refresh_window = 2

"""

Window, in seconds, within which a schema component will be refreshed after

receiving a schema_change event.

The driver delays a random amount of time in the range [0.0, window)

before executing the refresh. This serves two purposes:

1.) Spread the refresh for deployments with large fanout from C* to client tier,

preventing a 'thundering herd' problem with many clients refreshing simultaneously.

2.) Remove redundant refreshes. Redundant events arriving within the delay period

are discarded, and only one refresh is executed.

Setting this to zero will execute refreshes immediately.

Setting this negative will disable schema refreshes in response to push events

(refreshes will still occur in response to schema change responses to DDL statements

executed by Sessions of this Cluster).

"""

topology_event_refresh_window = 10

"""

Window, in seconds, within which the node and token list will be refreshed after

receiving a topology_change event.

Setting this to zero will execute refreshes immediately.

Setting this negative will disable node refreshes in response to push events.

See :attr:`.schema_event_refresh_window` for discussion of rationale

"""

status_event_refresh_window = 2

"""

Window, in seconds, within which the driver will start the reconnect after

receiving a status_change event.

Setting this to zero will connect immediately.

This is primarily used to avoid 'thundering herd' in deployments with large fanout from cluster to clients.

When nodes come up, clients attempt to reprepare prepared statements (depending on :attr:`.reprepare_on_up`), and

establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor.

"""

prepare_on_all_hosts = True

"""

Specifies whether statements should be prepared on all hosts, or just one.

This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup,

where a randomized initial condition of the load balancing policy can be expected to distribute prepares from

different clients across the cluster.

"""

reprepare_on_up = True

"""

Specifies whether all known prepared statements should be prepared on a node when it comes up.

May be used to avoid overwhelming a node on return, or if it is supposed that the node was only marked down due to

network. If statements are not reprepared, they are prepared on the first execution, causing

an extra roundtrip for one or more client requests.

"""

connect_timeout = 5

"""

Timeout, in seconds, for creating new connections.

This timeout covers the entire connection negotiation, including TCP

establishment, options passing, and authentication.

"""

@property

def schema_metadata_enabled(self):

"""

Flag indicating whether internal schema metadata is updated.

When disabled, the driver does not populate Cluster.metadata.keyspaces on connect, or on schema change events. This

can be used to speed initial connection, and reduce load on client and server during operation. Turning this off

gives away token aware request routing, and programmatic inspection of the metadata model.

"""

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):

"""

Flag indicating whether internal token metadata is updated.

When disabled, the driver does not query node token information on connect, or on topology change events. This

can be used to speed initial connection, and reduce load on client and server during operation. It is most useful

in large clusters using vnodes, where the token map can be expensive to compute. Turning this off

gives away token aware request routing, and programmatic inspection of the token ring.

"""

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

"""

A map of {keyspace: {type_name: UserType}}

"""

_listeners = None

_listener_lock = None

def __init__(self,

contact_points=["127.0.0.1"],

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):

"""

``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as

extablishing connection pools or refreshing metadata.

Any of the mutable Cluster attributes may be set as keyword arguments to the constructor.

"""

if contact_points is not None:

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

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.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._min_requests_per_connection = {

HostDistance.LOCAL: DEFAULT_MIN_REQUESTS,

HostDistance.REMOTE: DEFAULT_MIN_REQUESTS

}

self._max_requests_per_connection = {

HostDistance.LOCAL: DEFAULT_MAX_REQUESTS,

HostDistance.REMOTE: DEFAULT_MAX_REQUESTS

}

self._core_connections_per_host = {

HostDistance.LOCAL: DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST,

HostDistance.REMOTE: DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST

}

self._max_connections_per_host = {

HostDistance.LOCAL: DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST,

HostDistance.REMOTE: DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST

}

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):

"""

Registers a class to use to represent a particular user-defined type.

Query parameters for this user-defined type will be assumed to be

instances of `klass`. Result sets for this user-defined type will

be instances of `klass`. If no class is registered for a user-defined

type, a namedtuple will be used for result sets, and non-prepared

statements may not encode parameters for this type correctly.

`keyspace` is the name of the keyspace that the UDT is defined in.

`user_type` is the string name of the UDT to register the mapping

for.

`klass` should be a class with attributes whose names match the

fields of the user-defined type. The constructor must accepts kwargs

for each of the fields in the UDT.

This method should only be called after the type has been created

within Cassandra.

Example::

cluster = Cluster(protocol_version=3)

session = cluster.connect()

session.set_keyspace('mykeyspace')

session.execute("CREATE TYPE address (street text, zipcode int)")

session.execute("CREATE TABLE users (id int PRIMARY KEY, location address)")

# create a class to map to the "address" UDT

class Address(object):

def __init__(self, street, zipcode):

self.street = street

self.zipcode = zipcode

cluster.register_user_type('mykeyspace', 'address', Address)

# insert a row using an instance of Address

session.execute("INSERT INTO users (id, location) VALUES (%s, %s)",

(0, Address("123 Main St.", 78723)))

# results will include Address instances

results = session.execute("SELECT * FROM users")

row = results[0]

print row.id, row.location.street, row.location.zipcode

"""

if self.protocol_version < 3:

log.warning("User Type serialization is only supported in native protocol version 3+ (%d in use). "

"CQL encoding for simple statements will still work, but named tuples will "

"be returned when reading type %s.%s.", self.protocol_version, keyspace, user_type)

self._user_types[keyspace][user_type] = klass

for session in 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):

"""

Adds an :class:`.ExecutionProfile` to the cluster. This makes it available for use by ``name`` in :meth:`.Session.execute`

and :meth:`.Session.execute_async`. This method will raise if the profile already exists.

Normally profiles will be injected at cluster initialization via ``Cluster(execution_profiles)``. This method

provides a way of adding them dynamically.

Adding a new profile updates the connection pools according to the specified ``load_balancing_policy``. By default,

this method will wait up to five seconds for the pool creation to complete, so the profile can be used immediately

upon return. This behavior can be controlled using ``pool_wait_timeout`` (see

`concurrent.futures.wait <https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.wait>`_

for timeout semantics).

"""

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 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 get_min_requests_per_connection(self, host_distance):

return self._min_requests_per_connection[host_distance]

def set_min_requests_per_connection(self, host_distance, min_requests):

"""

Sets a threshold for concurrent requests per connection, below which

connections will be considered for disposal (down to core connections;

see :meth:`~Cluster.set_core_connections_per_host`).

Pertains to connection pool management in protocol versions {1,2}.

"""

if self.protocol_version >= 3:

raise UnsupportedOperation(

"Cluster.set_min_requests_per_connection() only has an effect "

"when using protocol_version 1 or 2.")

if min_requests < 0 or min_requests > 126 or \

min_requests >= self._max_requests_per_connection[host_distance]:

raise ValueError("min_requests must be 0-126 and less than the max_requests for this host_distance (%d)" %

(self._min_requests_per_connection[host_distance],))

self._min_requests_per_connection[host_distance] = min_requests

def get_max_requests_per_connection(self, host_distance):

return self._max_requests_per_connection[host_distance]

def set_max_requests_per_connection(self, host_distance, max_requests):