<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import atexit <EOL> from collections import defaultdict , Mapping <EOL> from concurrent . futures import ThreadPoolExecutor <EOL> import logging <EOL> from random import random <EOL> import socket <EOL> import sys <EOL> import time <EOL> from threading import Lock , RLock , Thread , Event <EOL> import warnings <EOL> import six <EOL> from six . moves import range <EOL> from six . moves import queue as Queue <EOL> import weakref <EOL> from weakref import WeakValueDictionary <EOL> try : <EOL> from weakref import WeakSet <EOL> except ImportError : <EOL> from cassandra . util import WeakSet <EOL> from functools import partial , wraps <EOL> from itertools import groupby , count <EOL> from cassandra import ( ConsistencyLevel , AuthenticationFailed , <EOL> OperationTimedOut , UnsupportedOperation , <EOL> SchemaTargetType , DriverException ) <EOL> from cassandra . connection import ( ConnectionException , ConnectionShutdown , <EOL> ConnectionHeartbeat , ProtocolVersionUnsupported ) <EOL> from cassandra . cqltypes import UserType <EOL> from cassandra . encoder import Encoder <EOL> from cassandra . protocol import ( QueryMessage , ResultMessage , <EOL> ErrorMessage , ReadTimeoutErrorMessage , <EOL> WriteTimeoutErrorMessage , <EOL> UnavailableErrorMessage , <EOL> OverloadedErrorMessage , <EOL> PrepareMessage , ExecuteMessage , <EOL> PreparedQueryNotFound , <EOL> IsBootstrappingErrorMessage , <EOL> BatchMessage , RESULT_KIND_PREPARED , <EOL> RESULT_KIND_SET_KEYSPACE , RESULT_KIND_ROWS , <EOL> RESULT_KIND_SCHEMA_CHANGE , MIN_SUPPORTED_VERSION , <EOL> ProtocolHandler ) <EOL> from cassandra . metadata import Metadata , protect_name , murmur3 <EOL> from cassandra . policies import ( TokenAwarePolicy , DCAwareRoundRobinPolicy , SimpleConvictionPolicy , <EOL> ExponentialReconnectionPolicy , HostDistance , <EOL> RetryPolicy , IdentityTranslator ) <EOL> from cassandra . pool import ( Host , _ReconnectionHandler , _HostReconnectionHandler , <EOL> HostConnectionPool , HostConnection , <EOL> NoConnectionsAvailable ) <EOL> from cassandra . query import ( SimpleStatement , PreparedStatement , BoundStatement , <EOL> BatchStatement , bind_params , QueryTrace , <EOL> named_tuple_factory , dict_factory , tuple_factory , FETCH_SIZE_UNSET ) <EOL> def _is_eventlet_monkey_patched ( ) : <EOL> if '<STR_LIT>' not in sys . modules : <EOL> return False <EOL> import eventlet . patcher <EOL> return eventlet . patcher . is_monkey_patched ( '<STR_LIT>' ) <EOL> def _is_gevent_monkey_patched ( ) : <EOL> if '<STR_LIT>' not in sys . modules : <EOL> return False <EOL> import gevent . socket <EOL> return socket . socket is gevent . socket . socket <EOL> if _is_gevent_monkey_patched ( ) : <EOL> from cassandra . io . geventreactor import GeventConnection as DefaultConnection <EOL> elif _is_eventlet_monkey_patched ( ) : <EOL> from cassandra . io . eventletreactor import EventletConnection as DefaultConnection <EOL> else : <EOL> try : <EOL> from cassandra . io . libevreactor import LibevConnection as DefaultConnection <EOL> except ImportError : <EOL> from cassandra . io . asyncorereactor import AsyncoreConnection as DefaultConnection <EOL> "<STR_LIT>" . encode ( '<STR_LIT:utf8>' ) <EOL> log = logging . getLogger ( __name__ ) <EOL> DEFAULT_MIN_REQUESTS = <NUM_LIT:5> <EOL> DEFAULT_MAX_REQUESTS = <NUM_LIT:100> <EOL> DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST = <NUM_LIT:2> <EOL> DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST = <NUM_LIT:8> <EOL> DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST = <NUM_LIT:1> <EOL> DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST = <NUM_LIT:2> <EOL> _NOT_SET = object ( ) <EOL> class NoHostAvailable ( Exception ) : <EOL> """<STR_LIT>""" <EOL> errors = None <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message , errors ) : <EOL> Exception . __init__ ( self , message , errors ) <EOL> self . errors = errors <EOL> def _future_completed ( future ) : <EOL> """<STR_LIT>""" <EOL> exc = future . exception ( ) <EOL> if exc : <EOL> log . debug ( "<STR_LIT>" , exc_info = exc ) <EOL> def run_in_executor ( f ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( f ) <EOL> def new_f ( self , * args , ** kwargs ) : <EOL> if self . is_shutdown : <EOL> return <EOL> try : <EOL> future = self . executor . submit ( f , self , * args , ** kwargs ) <EOL> future . add_done_callback ( _future_completed ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> return new_f <EOL> def _shutdown_cluster ( cluster ) : <EOL> if cluster and not cluster . is_shutdown : <EOL> cluster . shutdown ( ) <EOL> import platform <EOL> if platform . python_implementation ( ) == '<STR_LIT>' : <EOL> def default_lbp_factory ( ) : <EOL> if murmur3 is not None : <EOL> return TokenAwarePolicy ( DCAwareRoundRobinPolicy ( ) ) <EOL> return DCAwareRoundRobinPolicy ( ) <EOL> else : <EOL> def default_lbp_factory ( ) : <EOL> return DCAwareRoundRobinPolicy ( ) <EOL> class Cluster ( object ) : <EOL> """<STR_LIT>""" <EOL> contact_points = [ '<STR_LIT:127.0.0.1>' ] <EOL> """<STR_LIT>""" <EOL> port = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> cql_version = None <EOL> """<STR_LIT>""" <EOL> protocol_version = <NUM_LIT:4> <EOL> """<STR_LIT>""" <EOL> compression = True <EOL> """<STR_LIT>""" <EOL> _auth_provider = None <EOL> _auth_provider_callable = None <EOL> @ property <EOL> def auth_provider ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _auth_provider <EOL> @ auth_provider . setter <EOL> def auth_provider ( self , value ) : <EOL> if not value : <EOL> self . _auth_provider = value <EOL> return <EOL> try : <EOL> self . _auth_provider_callable = value . new_authenticator <EOL> except AttributeError : <EOL> if self . protocol_version > <NUM_LIT:1> : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> elif not callable ( value ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . _auth_provider_callable = value <EOL> self . _auth_provider = value <EOL> load_balancing_policy = None <EOL> """<STR_LIT>""" <EOL> reconnection_policy = ExponentialReconnectionPolicy ( <NUM_LIT:1.0> , <NUM_LIT> ) <EOL> """<STR_LIT>""" <EOL> default_retry_policy = RetryPolicy ( ) <EOL> """<STR_LIT>""" <EOL> conviction_policy_factory = SimpleConvictionPolicy <EOL> """<STR_LIT>""" <EOL> address_translator = IdentityTranslator ( ) <EOL> """<STR_LIT>""" <EOL> connect_to_remote_hosts = True <EOL> """<STR_LIT>""" <EOL> metrics_enabled = False <EOL> """<STR_LIT>""" <EOL> metrics = None <EOL> """<STR_LIT>""" <EOL> ssl_options = None <EOL> """<STR_LIT>""" <EOL> sockopts = None <EOL> """<STR_LIT>""" <EOL> max_schema_agreement_wait = <NUM_LIT:10> <EOL> """<STR_LIT>""" <EOL> metadata = None <EOL> """<STR_LIT>""" <EOL> connection_class = DefaultConnection <EOL> """<STR_LIT>""" <EOL> control_connection_timeout = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> idle_heartbeat_interval = <NUM_LIT:30> <EOL> """<STR_LIT>""" <EOL> schema_event_refresh_window = <NUM_LIT:2> <EOL> """<STR_LIT>""" <EOL> topology_event_refresh_window = <NUM_LIT:10> <EOL> """<STR_LIT>""" <EOL> connect_timeout = <NUM_LIT:5> <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def schema_metadata_enabled ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . control_connection . _schema_meta_enabled <EOL> @ schema_metadata_enabled . setter <EOL> def schema_metadata_enabled ( self , enabled ) : <EOL> self . control_connection . _schema_meta_enabled = bool ( enabled ) <EOL> @ property <EOL> def token_metadata_enabled ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . control_connection . _token_meta_enabled <EOL> @ token_metadata_enabled . setter <EOL> def token_metadata_enabled ( self , enabled ) : <EOL> self . control_connection . _token_meta_enabled = bool ( enabled ) <EOL> sessions = None <EOL> control_connection = None <EOL> scheduler = None <EOL> executor = None <EOL> is_shutdown = False <EOL> _is_setup = False <EOL> _prepared_statements = None <EOL> _prepared_statement_lock = None <EOL> _idle_heartbeat = None <EOL> _user_types = None <EOL> """<STR_LIT>""" <EOL> _listeners = None <EOL> _listener_lock = None <EOL> def __init__ ( self , <EOL> contact_points = [ "<STR_LIT:127.0.0.1>" ] , <EOL> port = <NUM_LIT> , <EOL> compression = True , <EOL> auth_provider = None , <EOL> load_balancing_policy = None , <EOL> reconnection_policy = None , <EOL> default_retry_policy = None , <EOL> conviction_policy_factory = None , <EOL> metrics_enabled = False , <EOL> connection_class = None , <EOL> ssl_options = None , <EOL> sockopts = None , <EOL> cql_version = None , <EOL> protocol_version = <NUM_LIT:4> , <EOL> executor_threads = <NUM_LIT:2> , <EOL> max_schema_agreement_wait = <NUM_LIT:10> , <EOL> control_connection_timeout = <NUM_LIT> , <EOL> idle_heartbeat_interval = <NUM_LIT:30> , <EOL> schema_event_refresh_window = <NUM_LIT:2> , <EOL> topology_event_refresh_window = <NUM_LIT:10> , <EOL> connect_timeout = <NUM_LIT:5> , <EOL> schema_metadata_enabled = True , <EOL> token_metadata_enabled = True , <EOL> address_translator = None ) : <EOL> """<STR_LIT>""" <EOL> if contact_points is not None : <EOL> if isinstance ( contact_points , six . string_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if None in contact_points : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . contact_points = contact_points <EOL> self . port = port <EOL> self . contact_points_resolved = [ endpoint [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] for a in self . contact_points <EOL> for endpoint in socket . getaddrinfo ( a , self . port , socket . AF_UNSPEC , socket . SOCK_STREAM ) ] <EOL> self . compression = compression <EOL> self . protocol_version = protocol_version <EOL> self . auth_provider = auth_provider <EOL> if load_balancing_policy is not None : <EOL> if isinstance ( load_balancing_policy , type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . load_balancing_policy = load_balancing_policy <EOL> else : <EOL> self . load_balancing_policy = default_lbp_factory ( ) <EOL> if reconnection_policy is not None : <EOL> if isinstance ( reconnection_policy , type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . reconnection_policy = reconnection_policy <EOL> if default_retry_policy is not None : <EOL> if isinstance ( default_retry_policy , type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . default_retry_policy = default_retry_policy <EOL> if conviction_policy_factory is not None : <EOL> if not callable ( conviction_policy_factory ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . conviction_policy_factory = conviction_policy_factory <EOL> if address_translator is not None : <EOL> if isinstance ( address_translator , type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . address_translator = address_translator <EOL> if connection_class is not None : <EOL> self . connection_class = connection_class <EOL> self . metrics_enabled = metrics_enabled <EOL> self . ssl_options = ssl_options <EOL> self . sockopts = sockopts <EOL> self . cql_version = cql_version <EOL> self . max_schema_agreement_wait = max_schema_agreement_wait <EOL> self . control_connection_timeout = control_connection_timeout <EOL> self . idle_heartbeat_interval = idle_heartbeat_interval <EOL> self . schema_event_refresh_window = schema_event_refresh_window <EOL> self . topology_event_refresh_window = topology_event_refresh_window <EOL> self . connect_timeout = connect_timeout <EOL> self . _listeners = set ( ) <EOL> self . _listener_lock = Lock ( ) <EOL> self . sessions = WeakSet ( ) <EOL> self . metadata = Metadata ( ) <EOL> self . control_connection = None <EOL> self . _prepared_statements = WeakValueDictionary ( ) <EOL> self . _prepared_statement_lock = Lock ( ) <EOL> self . _user_types = defaultdict ( dict ) <EOL> self . _min_requests_per_connection = { <EOL> HostDistance . LOCAL : DEFAULT_MIN_REQUESTS , <EOL> HostDistance . REMOTE : DEFAULT_MIN_REQUESTS <EOL> } <EOL> self . _max_requests_per_connection = { <EOL> HostDistance . LOCAL : DEFAULT_MAX_REQUESTS , <EOL> HostDistance . REMOTE : DEFAULT_MAX_REQUESTS <EOL> } <EOL> self . _core_connections_per_host = { <EOL> HostDistance . LOCAL : DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST , <EOL> HostDistance . REMOTE : DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST <EOL> } <EOL> self . _max_connections_per_host = { <EOL> HostDistance . LOCAL : DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST , <EOL> HostDistance . REMOTE : DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST <EOL> } <EOL> self . executor = ThreadPoolExecutor ( max_workers = executor_threads ) <EOL> self . scheduler = _Scheduler ( self . executor ) <EOL> self . _lock = RLock ( ) <EOL> if self . metrics_enabled : <EOL> from cassandra . metrics import Metrics <EOL> self . metrics = Metrics ( weakref . proxy ( self ) ) <EOL> self . control_connection = ControlConnection ( <EOL> self , self . control_connection_timeout , <EOL> self . schema_event_refresh_window , self . topology_event_refresh_window , <EOL> schema_metadata_enabled , token_metadata_enabled ) <EOL> def register_user_type ( self , keyspace , user_type , klass ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol_version < <NUM_LIT:3> : <EOL> log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , self . protocol_version , keyspace , user_type ) <EOL> self . _user_types [ keyspace ] [ user_type ] = klass <EOL> for session in self . sessions : <EOL> session . user_type_registered ( keyspace , user_type , klass ) <EOL> UserType . evict_udt_class ( keyspace , user_type ) <EOL> def get_min_requests_per_connection ( self , host_distance ) : <EOL> return self . _min_requests_per_connection [ host_distance ] <EOL> def set_min_requests_per_connection ( self , host_distance , min_requests ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol_version >= <NUM_LIT:3> : <EOL> raise UnsupportedOperation ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . _min_requests_per_connection [ host_distance ] = min_requests <EOL> def get_max_requests_per_connection ( self , host_distance ) : <EOL> return self . _max_requests_per_connection [ host_distance ] <EOL> def set_max_requests_per_connection ( self , host_distance , max_requests ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol_version >= <NUM_LIT:3> : <EOL> raise UnsupportedOperation ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . _max_requests_per_connection [ host_distance ] = max_requests <EOL> def get_core_connections_per_host ( self , host_distance ) : <EOL> """<STR_LIT>""" <EOL> return self . _core_connections_per_host [ host_distance ] <EOL> def set_core_connections_per_host ( self , host_distance , core_connections ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol_version >= <NUM_LIT:3> : <EOL> raise UnsupportedOperation ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> old = self . _core_connections_per_host [ host_distance ] <EOL> self . _core_connections_per_host [ host_distance ] = core_connections <EOL> if old < core_connections : <EOL> self . _ensure_core_connections ( ) <EOL> def get_max_connections_per_host ( self , host_distance ) : <EOL> """<STR_LIT>""" <EOL> return self . _max_connections_per_host [ host_distance ] <EOL> def set_max_connections_per_host ( self , host_distance , max_connections ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol_version >= <NUM_LIT:3> : <EOL> raise UnsupportedOperation ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . _max_connections_per_host [ host_distance ] = max_connections <EOL> def connection_factory ( self , address , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs = self . _make_connection_kwargs ( address , kwargs ) <EOL> return self . connection_class . factory ( address , self . connect_timeout , * args , ** kwargs ) <EOL> def _make_connection_factory ( self , host , * args , ** kwargs ) : <EOL> kwargs = self . _make_connection_kwargs ( host . address , kwargs ) <EOL> return partial ( self . connection_class . factory , host . address , self . connect_timeout , * args , ** kwargs ) <EOL> def _make_connection_kwargs ( self , address , kwargs_dict ) : <EOL> if self . _auth_provider_callable : <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . _auth_provider_callable ( address ) ) <EOL> kwargs_dict . setdefault ( '<STR_LIT:port>' , self . port ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . compression ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . sockopts ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . ssl_options ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . cql_version ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . protocol_version ) <EOL> kwargs_dict . setdefault ( '<STR_LIT>' , self . _user_types ) <EOL> return kwargs_dict <EOL> def protocol_downgrade ( self , host_addr , previous_version ) : <EOL> new_version = previous_version - <NUM_LIT:1> <EOL> if new_version < self . protocol_version : <EOL> if new_version >= MIN_SUPPORTED_VERSION : <EOL> log . warning ( "<STR_LIT>" , self . protocol_version , new_version , host_addr ) <EOL> self . protocol_version = new_version <EOL> else : <EOL> raise DriverException ( "<STR_LIT>" % ( new_version , MIN_SUPPORTED_VERSION ) ) <EOL> def connect ( self , keyspace = None ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> if self . is_shutdown : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> if not self . _is_setup : <EOL> log . debug ( "<STR_LIT>" , <EOL> self . contact_points , self . protocol_version ) <EOL> self . connection_class . initialize_reactor ( ) <EOL> atexit . register ( partial ( _shutdown_cluster , self ) ) <EOL> for address in self . contact_points_resolved : <EOL> host , new = self . add_host ( address , signal = False ) <EOL> if new : <EOL> host . set_up ( ) <EOL> for listener in self . listeners : <EOL> listener . on_add ( host ) <EOL> self . load_balancing_policy . populate ( <EOL> weakref . proxy ( self ) , self . metadata . all_hosts ( ) ) <EOL> try : <EOL> self . control_connection . connect ( ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . shutdown ( ) <EOL> raise <EOL> self . load_balancing_policy . check_supported ( ) <EOL> if self . idle_heartbeat_interval : <EOL> self . _idle_heartbeat = ConnectionHeartbeat ( self . idle_heartbeat_interval , self . get_connection_holders ) <EOL> self . _is_setup = True <EOL> session = self . _new_session ( ) <EOL> if keyspace : <EOL> session . set_keyspace ( keyspace ) <EOL> return session <EOL> def get_connection_holders ( self ) : <EOL> holders = [ ] <EOL> for s in self . sessions : <EOL> holders . extend ( s . get_pools ( ) ) <EOL> holders . append ( self . control_connection ) <EOL> return holders <EOL> def shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> if self . is_shutdown : <EOL> return <EOL> else : <EOL> self . is_shutdown = True <EOL> if self . _idle_heartbeat : <EOL> self . _idle_heartbeat . stop ( ) <EOL> self . scheduler . shutdown ( ) <EOL> self . control_connection . shutdown ( ) <EOL> for session in self . sessions : <EOL> session . shutdown ( ) <EOL> self . executor . shutdown ( ) <EOL> def _new_session ( self ) : <EOL> session = Session ( self , self . metadata . all_hosts ( ) ) <EOL> self . _session_register_user_types ( session ) <EOL> self . sessions . add ( session ) <EOL> return session <EOL> def _session_register_user_types ( self , session ) : <EOL> for keyspace , type_map in six . iteritems ( self . _user_types ) : <EOL> for udt_name , klass in six . iteritems ( type_map ) : <EOL> session . user_type_registered ( keyspace , udt_name , klass ) <EOL> def _cleanup_failed_on_up_handling ( self , host ) : <EOL> self . load_balancing_policy . on_down ( host ) <EOL> self . control_connection . on_down ( host ) <EOL> for session in self . sessions : <EOL> session . remove_pool ( host ) <EOL> self . _start_reconnector ( host , is_host_addition = False ) <EOL> def _on_up_future_completed ( self , host , futures , results , lock , finished_future ) : <EOL> with lock : <EOL> futures . discard ( finished_future ) <EOL> try : <EOL> results . append ( finished_future . result ( ) ) <EOL> except Exception as exc : <EOL> results . append ( exc ) <EOL> if futures : <EOL> return <EOL> try : <EOL> for exc in [ f for f in results if isinstance ( f , Exception ) ] : <EOL> log . error ( "<STR_LIT>" , host , exc_info = exc ) <EOL> self . _cleanup_failed_on_up_handling ( host ) <EOL> return <EOL> if not all ( results ) : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> self . _cleanup_failed_on_up_handling ( host ) <EOL> return <EOL> log . info ( "<STR_LIT>" , host ) <EOL> host . set_up ( ) <EOL> for listener in self . listeners : <EOL> listener . on_up ( host ) <EOL> finally : <EOL> with host . lock : <EOL> host . _currently_handling_node_up = False <EOL> for session in self . sessions : <EOL> session . update_created_pools ( ) <EOL> def on_up ( self , host ) : <EOL> """<STR_LIT>""" <EOL> if self . is_shutdown : <EOL> return <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> with host . lock : <EOL> if host . _currently_handling_node_up : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> return <EOL> if host . is_up : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> return <EOL> host . _currently_handling_node_up = True <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> have_future = False <EOL> futures = set ( ) <EOL> try : <EOL> log . info ( "<STR_LIT>" , host ) <EOL> reconnector = host . get_and_set_reconnection_handler ( None ) <EOL> if reconnector : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> reconnector . cancel ( ) <EOL> self . _prepare_all_queries ( host ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> for session in self . sessions : <EOL> session . remove_pool ( host ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> self . load_balancing_policy . on_up ( host ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> self . control_connection . on_up ( host ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> futures_lock = Lock ( ) <EOL> futures_results = [ ] <EOL> callback = partial ( self . _on_up_future_completed , host , futures , futures_results , futures_lock ) <EOL> for session in self . sessions : <EOL> future = session . add_or_renew_pool ( host , is_host_addition = False ) <EOL> if future is not None : <EOL> have_future = True <EOL> future . add_done_callback ( callback ) <EOL> futures . add ( future ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" , host ) <EOL> for future in futures : <EOL> future . cancel ( ) <EOL> self . _cleanup_failed_on_up_handling ( host ) <EOL> with host . lock : <EOL> host . _currently_handling_node_up = False <EOL> raise <EOL> else : <EOL> if not have_future : <EOL> with host . lock : <EOL> host . _currently_handling_node_up = False <EOL> return futures <EOL> def _start_reconnector ( self , host , is_host_addition ) : <EOL> if self . load_balancing_policy . distance ( host ) == HostDistance . IGNORED : <EOL> return <EOL> schedule = self . reconnection_policy . new_schedule ( ) <EOL> conn_factory = self . _make_connection_factory ( host ) <EOL> reconnector = _HostReconnectionHandler ( <EOL> host , conn_factory , is_host_addition , self . on_add , self . on_up , <EOL> self . scheduler , schedule , host . get_and_set_reconnection_handler , <EOL> new_handler = None ) <EOL> old_reconnector = host . get_and_set_reconnection_handler ( reconnector ) <EOL> if old_reconnector : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> old_reconnector . cancel ( ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> reconnector . start ( ) <EOL> @ run_in_executor <EOL> def on_down ( self , host , is_host_addition , expect_host_to_be_down = False ) : <EOL> """<STR_LIT>""" <EOL> if self . is_shutdown : <EOL> return <EOL> with host . lock : <EOL> if ( not host . is_up and not expect_host_to_be_down ) or host . is_currently_reconnecting ( ) : <EOL> return <EOL> host . set_down ( ) <EOL> log . warning ( "<STR_LIT>" , host ) <EOL> self . load_balancing_policy . on_down ( host ) <EOL> self . control_connection . on_down ( host ) <EOL> for session in self . sessions : <EOL> session . on_down ( host ) <EOL> for listener in self . listeners : <EOL> listener . on_down ( host ) <EOL> self . _start_reconnector ( host , is_host_addition ) <EOL> def on_add ( self , host , refresh_nodes = True ) : <EOL> if self . is_shutdown : <EOL> return <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> distance = self . load_balancing_policy . distance ( host ) <EOL> if distance != HostDistance . IGNORED : <EOL> self . _prepare_all_queries ( host ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> self . load_balancing_policy . on_add ( host ) <EOL> self . control_connection . on_add ( host , refresh_nodes ) <EOL> if distance == HostDistance . IGNORED : <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , host ) <EOL> self . _finalize_add ( host ) <EOL> return <EOL> futures_lock = Lock ( ) <EOL> futures_results = [ ] <EOL> futures = set ( ) <EOL> def future_completed ( future ) : <EOL> with futures_lock : <EOL> futures . discard ( future ) <EOL> try : <EOL> futures_results . append ( future . result ( ) ) <EOL> except Exception as exc : <EOL> futures_results . append ( exc ) <EOL> if futures : <EOL> return <EOL> log . debug ( '<STR_LIT>' , host ) <EOL> for exc in [ f for f in futures_results if isinstance ( f , Exception ) ] : <EOL> log . error ( "<STR_LIT>" , host , exc_info = exc ) <EOL> return <EOL> if not all ( futures_results ) : <EOL> log . warning ( "<STR_LIT>" , host ) <EOL> return <EOL> self . _finalize_add ( host ) <EOL> have_future = False <EOL> for session in self . sessions : <EOL> future = session . add_or_renew_pool ( host , is_host_addition = True ) <EOL> if future is not None : <EOL> have_future = True <EOL> futures . add ( future ) <EOL> future . add_done_callback ( future_completed ) <EOL> if not have_future : <EOL> self . _finalize_add ( host ) <EOL> def _finalize_add ( self , host ) : <EOL> host . set_up ( ) <EOL> for listener in self . listeners : <EOL> listener . on_add ( host ) <EOL> for session in self . sessions : <EOL> session . update_created_pools ( ) <EOL> def on_remove ( self , host ) : <EOL> if self . is_shutdown : <EOL> return <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> host . set_down ( ) <EOL> self . load_balancing_policy . on_remove ( host ) <EOL> for session in self . sessions : <EOL> session . on_remove ( host ) <EOL> for listener in self . listeners : <EOL> listener . on_remove ( host ) <EOL> self . control_connection . on_remove ( host ) <EOL> def signal_connection_failure ( self , host , connection_exc , is_host_addition , expect_host_to_be_down = False ) : <EOL> is_down = host . signal_connection_failure ( connection_exc ) <EOL> if is_down : <EOL> self . on_down ( host , is_host_addition , expect_host_to_be_down ) <EOL> return is_down <EOL> def add_host ( self , address , datacenter = None , rack = None , signal = True , refresh_nodes = True ) : <EOL> """<STR_LIT>""" <EOL> host , new = self . metadata . add_or_return_host ( Host ( address , self . conviction_policy_factory , datacenter , rack ) ) <EOL> if new and signal : <EOL> log . info ( "<STR_LIT>" , host ) <EOL> self . on_add ( host , refresh_nodes ) <EOL> return host , new <EOL> def remove_host ( self , host ) : <EOL> """<STR_LIT>""" <EOL> if host and self . metadata . remove_host ( host ) : <EOL> log . info ( "<STR_LIT>" , host ) <EOL> self . on_remove ( host ) <EOL> def register_listener ( self , listener ) : <EOL> """<STR_LIT>""" <EOL> with self . _listener_lock : <EOL> self . _listeners . add ( listener ) <EOL> def unregister_listener ( self , listener ) : <EOL> """<STR_LIT>""" <EOL> with self . _listener_lock : <EOL> self . _listeners . remove ( listener ) <EOL> @ property <EOL> def listeners ( self ) : <EOL> with self . _listener_lock : <EOL> return self . _listeners . copy ( ) <EOL> def _ensure_core_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> for session in self . sessions : <EOL> for pool in session . _pools . values ( ) : <EOL> pool . ensure_core_connections ( ) <EOL> @ staticmethod <EOL> def _validate_refresh_schema ( keyspace , table , usertype , function , aggregate ) : <EOL> if any ( ( table , usertype , function , aggregate ) ) : <EOL> if not keyspace : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if sum ( <NUM_LIT:1> for e in ( table , usertype , function ) if e ) > <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _target_type_from_refresh_args ( keyspace , table , usertype , function , aggregate ) : <EOL> if aggregate : <EOL> return SchemaTargetType . AGGREGATE <EOL> elif function : <EOL> return SchemaTargetType . FUNCTION <EOL> elif usertype : <EOL> return SchemaTargetType . TYPE <EOL> elif table : <EOL> return SchemaTargetType . TABLE <EOL> elif keyspace : <EOL> return SchemaTargetType . KEYSPACE <EOL> return None <EOL> def refresh_schema_metadata ( self , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_keyspace_metadata ( self , keyspace , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . KEYSPACE , keyspace = keyspace , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_table_metadata ( self , keyspace , table , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . TABLE , keyspace = keyspace , table = table , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_materialized_view_metadata ( self , keyspace , view , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . TABLE , keyspace = keyspace , table = view , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_user_type_metadata ( self , keyspace , user_type , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . TYPE , keyspace = keyspace , type = user_type , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_user_function_metadata ( self , keyspace , function , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . FUNCTION , keyspace = keyspace , function = function , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_user_aggregate_metadata ( self , keyspace , aggregate , max_schema_agreement_wait = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_schema ( target_type = SchemaTargetType . AGGREGATE , keyspace = keyspace , aggregate = aggregate , <EOL> schema_agreement_wait = max_schema_agreement_wait , force = True ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def refresh_nodes ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . control_connection . refresh_node_list_and_token_map ( ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> def set_meta_refresh_enabled ( self , enabled ) : <EOL> """<STR_LIT>""" <EOL> self . schema_metadata_enabled = enabled <EOL> self . token_metadata_enabled = enabled <EOL> def _prepare_all_queries ( self , host ) : <EOL> if not self . _prepared_statements : <EOL> return <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> connection = None <EOL> try : <EOL> connection = self . connection_factory ( host . address ) <EOL> try : <EOL> self . control_connection . wait_for_schema_agreement ( connection ) <EOL> except Exception : <EOL> log . debug ( "<STR_LIT>" , host , exc_info = True ) <EOL> statements = self . _prepared_statements . values ( ) <EOL> for keyspace , ks_statements in groupby ( statements , lambda s : s . keyspace ) : <EOL> if keyspace is not None : <EOL> connection . set_keyspace_blocking ( keyspace ) <EOL> ks_statements = list ( ks_statements ) <EOL> chunks = [ ] <EOL> for i in range ( <NUM_LIT:0> , len ( ks_statements ) , <NUM_LIT:10> ) : <EOL> chunks . append ( ks_statements [ i : i + <NUM_LIT:10> ] ) <EOL> for ks_chunk in chunks : <EOL> messages = [ PrepareMessage ( query = s . query_string ) for s in ks_chunk ] <EOL> responses = connection . wait_for_responses ( * messages , timeout = <NUM_LIT> ) <EOL> for response in responses : <EOL> if ( not isinstance ( response , ResultMessage ) or <EOL> response . kind != RESULT_KIND_PREPARED ) : <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , host , response ) <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> except OperationTimedOut as timeout : <EOL> log . warning ( "<STR_LIT>" , host , timeout ) <EOL> except ( ConnectionException , socket . error ) as exc : <EOL> log . warning ( "<STR_LIT>" , host , exc ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" , host ) <EOL> finally : <EOL> if connection : <EOL> connection . close ( ) <EOL> def prepare_on_all_sessions ( self , query_id , prepared_statement , excluded_host ) : <EOL> with self . _prepared_statement_lock : <EOL> self . _prepared_statements [ query_id ] = prepared_statement <EOL> for session in self . sessions : <EOL> session . prepare_on_all_hosts ( prepared_statement . query_string , excluded_host ) <EOL> class Session ( object ) : <EOL> """<STR_LIT>""" <EOL> cluster = None <EOL> hosts = None <EOL> keyspace = None <EOL> is_shutdown = False <EOL> row_factory = staticmethod ( named_tuple_factory ) <EOL> """<STR_LIT>""" <EOL> default_timeout = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> default_consistency_level = ConsistencyLevel . LOCAL_ONE <EOL> """<STR_LIT>""" <EOL> default_serial_consistency_level = None <EOL> """<STR_LIT>""" <EOL> max_trace_wait = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> default_fetch_size = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> use_client_timestamp = True <EOL> """<STR_LIT>""" <EOL> encoder = None <EOL> """<STR_LIT>""" <EOL> client_protocol_handler = ProtocolHandler <EOL> """<STR_LIT>""" <EOL> _lock = None <EOL> _pools = None <EOL> _load_balancer = None <EOL> _metrics = None <EOL> def __init__ ( self , cluster , hosts ) : <EOL> self . cluster = cluster <EOL> self . hosts = hosts <EOL> self . _lock = RLock ( ) <EOL> self . _pools = { } <EOL> self . _load_balancer = cluster . load_balancing_policy <EOL> self . _metrics = cluster . metrics <EOL> self . _protocol_version = self . cluster . protocol_version <EOL> self . encoder = Encoder ( ) <EOL> futures = [ ] <EOL> for host in hosts : <EOL> future = self . add_or_renew_pool ( host , is_host_addition = False ) <EOL> if future is not None : <EOL> futures . append ( future ) <EOL> for future in futures : <EOL> future . result ( ) <EOL> def execute ( self , query , parameters = None , timeout = _NOT_SET , trace = False , custom_payload = None ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_async ( query , parameters , trace , custom_payload , timeout ) . result ( ) <EOL> def execute_async ( self , query , parameters = None , trace = False , custom_payload = None , timeout = _NOT_SET ) : <EOL> """<STR_LIT>""" <EOL> if timeout is _NOT_SET : <EOL> timeout = self . default_timeout <EOL> future = self . _create_response_future ( query , parameters , trace , custom_payload , timeout ) <EOL> future . _protocol_handler = self . client_protocol_handler <EOL> future . send_request ( ) <EOL> return future <EOL> def _create_response_future ( self , query , parameters , trace , custom_payload , timeout ) : <EOL> """<STR_LIT>""" <EOL> prepared_statement = None <EOL> if isinstance ( query , six . string_types ) : <EOL> query = SimpleStatement ( query ) <EOL> elif isinstance ( query , PreparedStatement ) : <EOL> query = query . bind ( parameters ) <EOL> cl = query . consistency_level if query . consistency_level is not None else self . default_consistency_level <EOL> serial_cl = query . serial_consistency_level if query . serial_consistency_level is not None else self . default_serial_consistency_level <EOL> fetch_size = query . fetch_size <EOL> if fetch_size is FETCH_SIZE_UNSET and self . _protocol_version >= <NUM_LIT:2> : <EOL> fetch_size = self . default_fetch_size <EOL> elif self . _protocol_version == <NUM_LIT:1> : <EOL> fetch_size = None <EOL> if self . _protocol_version >= <NUM_LIT:3> and self . use_client_timestamp : <EOL> timestamp = int ( time . time ( ) * <NUM_LIT> ) <EOL> else : <EOL> timestamp = None <EOL> if isinstance ( query , SimpleStatement ) : <EOL> query_string = query . query_string <EOL> if parameters : <EOL> query_string = bind_params ( query_string , parameters , self . encoder ) <EOL> message = QueryMessage ( <EOL> query_string , cl , serial_cl , <EOL> fetch_size , timestamp = timestamp ) <EOL> elif isinstance ( query , BoundStatement ) : <EOL> message = ExecuteMessage ( <EOL> query . prepared_statement . query_id , query . values , cl , <EOL> serial_cl , fetch_size , <EOL> timestamp = timestamp ) <EOL> prepared_statement = query . prepared_statement <EOL> elif isinstance ( query , BatchStatement ) : <EOL> if self . _protocol_version < <NUM_LIT:2> : <EOL> raise UnsupportedOperation ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> message = BatchMessage ( <EOL> query . batch_type , query . _statements_and_parameters , cl , <EOL> serial_cl , timestamp ) <EOL> message . tracing = trace <EOL> message . update_custom_payload ( query . custom_payload ) <EOL> message . update_custom_payload ( custom_payload ) <EOL> return ResponseFuture ( <EOL> self , message , query , timeout , metrics = self . _metrics , <EOL> prepared_statement = prepared_statement ) <EOL> def prepare ( self , query , custom_payload = None ) : <EOL> """<STR_LIT>""" <EOL> message = PrepareMessage ( query = query ) <EOL> future = ResponseFuture ( self , message , query = None , timeout = self . default_timeout ) <EOL> try : <EOL> future . send_request ( ) <EOL> query_id , column_metadata , pk_indexes = future . result ( ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> raise <EOL> prepared_statement = PreparedStatement . from_message ( <EOL> query_id , column_metadata , pk_indexes , self . cluster . metadata , query , self . keyspace , <EOL> self . _protocol_version ) <EOL> prepared_statement . custom_payload = future . custom_payload <EOL> host = future . _current_host <EOL> try : <EOL> self . cluster . prepare_on_all_sessions ( query_id , prepared_statement , host ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> return prepared_statement <EOL> def prepare_on_all_hosts ( self , query , excluded_host ) : <EOL> """<STR_LIT>""" <EOL> futures = [ ] <EOL> for host in self . _pools . keys ( ) : <EOL> if host != excluded_host and host . is_up : <EOL> future = ResponseFuture ( self , PrepareMessage ( query = query ) , None , self . default_timeout ) <EOL> try : <EOL> request_id = future . _query ( host ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" , host ) <EOL> continue <EOL> if request_id is None : <EOL> log . debug ( "<STR_LIT>" , <EOL> host , future . _errors . get ( host ) ) <EOL> continue <EOL> futures . append ( ( host , future ) ) <EOL> for host , future in futures : <EOL> try : <EOL> future . result ( ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" , host ) <EOL> def shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> if self . is_shutdown : <EOL> return <EOL> else : <EOL> self . is_shutdown = True <EOL> for pool in self . _pools . values ( ) : <EOL> pool . shutdown ( ) <EOL> def add_or_renew_pool ( self , host , is_host_addition ) : <EOL> """<STR_LIT>""" <EOL> distance = self . _load_balancer . distance ( host ) <EOL> if distance == HostDistance . IGNORED : <EOL> return None <EOL> def run_add_or_renew_pool ( ) : <EOL> try : <EOL> if self . _protocol_version >= <NUM_LIT:3> : <EOL> new_pool = HostConnection ( host , distance , self ) <EOL> else : <EOL> new_pool = HostConnectionPool ( host , distance , self ) <EOL> except AuthenticationFailed as auth_exc : <EOL> conn_exc = ConnectionException ( str ( auth_exc ) , host = host ) <EOL> self . cluster . signal_connection_failure ( host , conn_exc , is_host_addition ) <EOL> return False <EOL> except Exception as conn_exc : <EOL> log . warning ( "<STR_LIT>" , <EOL> host , exc_info = conn_exc ) <EOL> self . cluster . signal_connection_failure ( <EOL> host , conn_exc , is_host_addition , expect_host_to_be_down = True ) <EOL> return False <EOL> previous = self . _pools . get ( host ) <EOL> self . _pools [ host ] = new_pool <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> if previous : <EOL> previous . shutdown ( ) <EOL> return True <EOL> return self . submit ( run_add_or_renew_pool ) <EOL> def remove_pool ( self , host ) : <EOL> pool = self . _pools . pop ( host , None ) <EOL> if pool : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> return self . submit ( pool . shutdown ) <EOL> else : <EOL> return None <EOL> def update_created_pools ( self ) : <EOL> """<STR_LIT>""" <EOL> for host in self . cluster . metadata . all_hosts ( ) : <EOL> distance = self . _load_balancer . distance ( host ) <EOL> pool = self . _pools . get ( host ) <EOL> if not pool or pool . is_shutdown : <EOL> if distance != HostDistance . IGNORED and host . is_up : <EOL> self . add_or_renew_pool ( host , False ) <EOL> elif distance != pool . host_distance : <EOL> if distance == HostDistance . IGNORED : <EOL> self . remove_pool ( host ) <EOL> else : <EOL> pool . host_distance = distance <EOL> def on_down ( self , host ) : <EOL> """<STR_LIT>""" <EOL> future = self . remove_pool ( host ) <EOL> if future : <EOL> future . add_done_callback ( lambda f : self . update_created_pools ( ) ) <EOL> def on_remove ( self , host ) : <EOL> """<STR_LIT>""" <EOL> self . on_down ( host ) <EOL> def set_keyspace ( self , keyspace ) : <EOL> """<STR_LIT>""" <EOL> self . execute ( '<STR_LIT>' % ( protect_name ( keyspace ) , ) ) <EOL> def _set_keyspace_for_all_pools ( self , keyspace , callback ) : <EOL> """<STR_LIT>""" <EOL> self . keyspace = keyspace <EOL> remaining_callbacks = set ( self . _pools . values ( ) ) <EOL> errors = { } <EOL> if not remaining_callbacks : <EOL> callback ( errors ) <EOL> return <EOL> def pool_finished_setting_keyspace ( pool , host_errors ) : <EOL> remaining_callbacks . remove ( pool ) <EOL> if host_errors : <EOL> errors [ pool . host ] = host_errors <EOL> if not remaining_callbacks : <EOL> callback ( host_errors ) <EOL> for pool in self . _pools . values ( ) : <EOL> pool . _set_keyspace_for_all_conns ( keyspace , pool_finished_setting_keyspace ) <EOL> def user_type_registered ( self , keyspace , user_type , klass ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> ks_meta = self . cluster . metadata . keyspaces [ keyspace ] <EOL> except KeyError : <EOL> raise UserTypeDoesNotExist ( <EOL> '<STR_LIT>' % ( keyspace , ) ) <EOL> try : <EOL> type_meta = ks_meta . user_types [ user_type ] <EOL> except KeyError : <EOL> raise UserTypeDoesNotExist ( <EOL> '<STR_LIT>' % ( user_type , keyspace ) ) <EOL> field_names = type_meta . field_names <EOL> if six . PY2 : <EOL> field_names = [ fn . encode ( '<STR_LIT:utf-8>' ) for fn in field_names ] <EOL> def encode ( val ) : <EOL> return '<STR_LIT>' % '<STR_LIT>' . join ( '<STR_LIT>' % ( <EOL> field_name , <EOL> self . encoder . cql_encode_all_types ( getattr ( val , field_name , None ) ) <EOL> ) for field_name in field_names ) <EOL> self . encoder . mapping [ klass ] = encode <EOL> def submit ( self , fn , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_shutdown : <EOL> return self . cluster . executor . submit ( fn , * args , ** kwargs ) <EOL> def get_pool_state ( self ) : <EOL> return dict ( ( host , pool . get_state ( ) ) for host , pool in self . _pools . items ( ) ) <EOL> def get_pools ( self ) : <EOL> return self . _pools . values ( ) <EOL> class UserTypeDoesNotExist ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class _ControlReconnectionHandler ( _ReconnectionHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , control_connection , * args , ** kwargs ) : <EOL> _ReconnectionHandler . __init__ ( self , * args , ** kwargs ) <EOL> self . control_connection = weakref . proxy ( control_connection ) <EOL> def try_reconnect ( self ) : <EOL> return self . control_connection . _reconnect_internal ( ) <EOL> def on_reconnection ( self , connection ) : <EOL> self . control_connection . _set_new_connection ( connection ) <EOL> def on_exception ( self , exc , next_delay ) : <EOL> if isinstance ( exc , AuthenticationFailed ) : <EOL> return False <EOL> else : <EOL> log . debug ( "<STR_LIT>" , exc ) <EOL> return True <EOL> def _watch_callback ( obj_weakref , method_name , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> obj = obj_weakref ( ) <EOL> if obj is None : <EOL> return <EOL> getattr ( obj , method_name ) ( * args , ** kwargs ) <EOL> def _clear_watcher ( conn , expiring_weakref ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> conn . control_conn_disposed ( ) <EOL> except ReferenceError : <EOL> pass <EOL> class ControlConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> _SELECT_PEERS = "<STR_LIT>" <EOL> _SELECT_PEERS_NO_TOKENS = "<STR_LIT>" <EOL> _SELECT_LOCAL = "<STR_LIT>" <EOL> _SELECT_LOCAL_NO_TOKENS = "<STR_LIT>" <EOL> _SELECT_SCHEMA_PEERS = "<STR_LIT>" <EOL> _SELECT_SCHEMA_LOCAL = "<STR_LIT>" <EOL> _is_shutdown = False <EOL> _timeout = None <EOL> _protocol_version = None <EOL> _schema_event_refresh_window = None <EOL> _topology_event_refresh_window = None <EOL> _schema_meta_enabled = True <EOL> _token_meta_enabled = True <EOL> _time = time <EOL> def __init__ ( self , cluster , timeout , <EOL> schema_event_refresh_window , <EOL> topology_event_refresh_window , <EOL> schema_meta_enabled = True , <EOL> token_meta_enabled = True ) : <EOL> self . _cluster = weakref . proxy ( cluster ) <EOL> self . _connection = None <EOL> self . _timeout = timeout <EOL> self . _schema_event_refresh_window = schema_event_refresh_window <EOL> self . _topology_event_refresh_window = topology_event_refresh_window <EOL> self . _schema_meta_enabled = schema_meta_enabled <EOL> self . _token_meta_enabled = token_meta_enabled <EOL> self . _lock = RLock ( ) <EOL> self . _schema_agreement_lock = Lock ( ) <EOL> self . _reconnection_handler = None <EOL> self . _reconnection_lock = RLock ( ) <EOL> self . _event_schedule_times = { } <EOL> def connect ( self ) : <EOL> if self . _is_shutdown : <EOL> return <EOL> self . _protocol_version = self . _cluster . protocol_version <EOL> self . _set_new_connection ( self . _reconnect_internal ( ) ) <EOL> def _set_new_connection ( self , conn ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> old = self . _connection <EOL> self . _connection = conn <EOL> if old : <EOL> log . debug ( "<STR_LIT>" , old , conn ) <EOL> old . close ( ) <EOL> def _reconnect_internal ( self ) : <EOL> """<STR_LIT>""" <EOL> errors = { } <EOL> for host in self . _cluster . load_balancing_policy . make_query_plan ( ) : <EOL> try : <EOL> return self . _try_connect ( host ) <EOL> except ConnectionException as exc : <EOL> errors [ host . address ] = exc <EOL> log . warning ( "<STR_LIT>" , host , exc_info = True ) <EOL> self . _cluster . signal_connection_failure ( host , exc , is_host_addition = False ) <EOL> except Exception as exc : <EOL> errors [ host . address ] = exc <EOL> log . warning ( "<STR_LIT>" , host , exc_info = True ) <EOL> if self . _is_shutdown : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> raise NoHostAvailable ( "<STR_LIT>" , errors ) <EOL> def _try_connect ( self , host ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> while True : <EOL> try : <EOL> connection = self . _cluster . connection_factory ( host . address , is_control_connection = True ) <EOL> if self . _is_shutdown : <EOL> connection . close ( ) <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> break <EOL> except ProtocolVersionUnsupported as e : <EOL> self . _cluster . protocol_downgrade ( host . address , e . startup_version ) <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> connection ) <EOL> self_weakref = weakref . ref ( self , callback = partial ( _clear_watcher , weakref . proxy ( connection ) ) ) <EOL> try : <EOL> connection . register_watchers ( { <EOL> "<STR_LIT>" : partial ( _watch_callback , self_weakref , '<STR_LIT>' ) , <EOL> "<STR_LIT>" : partial ( _watch_callback , self_weakref , '<STR_LIT>' ) , <EOL> "<STR_LIT>" : partial ( _watch_callback , self_weakref , '<STR_LIT>' ) <EOL> } , register_timeout = self . _timeout ) <EOL> sel_peers = self . _SELECT_PEERS if self . _token_meta_enabled else self . _SELECT_PEERS_NO_TOKENS <EOL> sel_local = self . _SELECT_LOCAL if self . _token_meta_enabled else self . _SELECT_LOCAL_NO_TOKENS <EOL> peers_query = QueryMessage ( query = sel_peers , consistency_level = ConsistencyLevel . ONE ) <EOL> local_query = QueryMessage ( query = sel_local , consistency_level = ConsistencyLevel . ONE ) <EOL> shared_results = connection . wait_for_responses ( <EOL> peers_query , local_query , timeout = self . _timeout ) <EOL> self . _refresh_node_list_and_token_map ( connection , preloaded_results = shared_results ) <EOL> self . _refresh_schema ( connection , preloaded_results = shared_results , schema_agreement_wait = - <NUM_LIT:1> ) <EOL> except Exception : <EOL> connection . close ( ) <EOL> raise <EOL> return connection <EOL> def reconnect ( self ) : <EOL> if self . _is_shutdown : <EOL> return <EOL> self . _submit ( self . _reconnect ) <EOL> def _reconnect ( self ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> try : <EOL> self . _set_new_connection ( self . _reconnect_internal ( ) ) <EOL> except NoHostAvailable : <EOL> schedule = self . cluster . reconnection_policy . new_schedule ( ) <EOL> with self . _reconnection_lock : <EOL> if self . _reconnection_handler : <EOL> self . _reconnection_handler . cancel ( ) <EOL> self . _reconnection_handler = _ControlReconnectionHandler ( <EOL> self , self . _cluster . scheduler , schedule , <EOL> self . _get_and_set_reconnection_handler , <EOL> new_handler = None ) <EOL> self . _reconnection_handler . start ( ) <EOL> except Exception : <EOL> log . debug ( "<STR_LIT>" , exc_info = True ) <EOL> raise <EOL> def _get_and_set_reconnection_handler ( self , new_handler ) : <EOL> """<STR_LIT>""" <EOL> with self . _reconnection_lock : <EOL> old = self . _reconnection_handler <EOL> self . _reconnection_handler = new_handler <EOL> return old <EOL> def _submit ( self , * args , ** kwargs ) : <EOL> try : <EOL> if not self . _cluster . is_shutdown : <EOL> return self . _cluster . executor . submit ( * args , ** kwargs ) <EOL> except ReferenceError : <EOL> pass <EOL> return None <EOL> def shutdown ( self ) : <EOL> with self . _reconnection_lock : <EOL> if self . _reconnection_handler : <EOL> self . _reconnection_handler . cancel ( ) <EOL> with self . _lock : <EOL> if self . _is_shutdown : <EOL> return <EOL> else : <EOL> self . _is_shutdown = True <EOL> log . debug ( "<STR_LIT>" ) <EOL> if self . _connection : <EOL> self . _connection . close ( ) <EOL> self . _connection = None <EOL> def refresh_schema ( self , force = False , ** kwargs ) : <EOL> try : <EOL> if self . _connection : <EOL> return self . _refresh_schema ( self . _connection , force = force , ** kwargs ) <EOL> except ReferenceError : <EOL> pass <EOL> except Exception : <EOL> log . debug ( "<STR_LIT>" , exc_info = True ) <EOL> self . _signal_error ( ) <EOL> return False <EOL> def _refresh_schema ( self , connection , preloaded_results = None , schema_agreement_wait = None , force = False , ** kwargs ) : <EOL> if self . _cluster . is_shutdown : <EOL> return False <EOL> agreed = self . wait_for_schema_agreement ( connection , <EOL> preloaded_results = preloaded_results , <EOL> wait_time = schema_agreement_wait ) <EOL> if not self . _schema_meta_enabled and not force : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return False <EOL> if not agreed : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return False <EOL> self . _cluster . metadata . refresh ( connection , self . _timeout , ** kwargs ) <EOL> return True <EOL> def refresh_node_list_and_token_map ( self , force_token_rebuild = False ) : <EOL> try : <EOL> if self . _connection : <EOL> self . _refresh_node_list_and_token_map ( self . _connection , force_token_rebuild = force_token_rebuild ) <EOL> return True <EOL> except ReferenceError : <EOL> pass <EOL> except Exception : <EOL> log . debug ( "<STR_LIT>" , exc_info = True ) <EOL> self . _signal_error ( ) <EOL> return False <EOL> def _refresh_node_list_and_token_map ( self , connection , preloaded_results = None , <EOL> force_token_rebuild = False ) : <EOL> if preloaded_results : <EOL> log . debug ( "<STR_LIT>" ) <EOL> peers_result = preloaded_results [ <NUM_LIT:0> ] <EOL> local_result = preloaded_results [ <NUM_LIT:1> ] <EOL> else : <EOL> cl = ConsistencyLevel . ONE <EOL> if not self . _token_meta_enabled : <EOL> log . debug ( "<STR_LIT>" ) <EOL> sel_peers = self . _SELECT_PEERS_NO_TOKENS <EOL> sel_local = self . _SELECT_LOCAL_NO_TOKENS <EOL> else : <EOL> log . debug ( "<STR_LIT>" ) <EOL> sel_peers = self . _SELECT_PEERS <EOL> sel_local = self . _SELECT_LOCAL <EOL> peers_query = QueryMessage ( query = sel_peers , consistency_level = cl ) <EOL> local_query = QueryMessage ( query = sel_local , consistency_level = cl ) <EOL> peers_result , local_result = connection . wait_for_responses ( <EOL> peers_query , local_query , timeout = self . _timeout ) <EOL> peers_result = dict_factory ( * peers_result . results ) <EOL> partitioner = None <EOL> token_map = { } <EOL> if local_result . results : <EOL> local_rows = dict_factory ( * ( local_result . results ) ) <EOL> local_row = local_rows [ <NUM_LIT:0> ] <EOL> cluster_name = local_row [ "<STR_LIT>" ] <EOL> self . _cluster . metadata . cluster_name = cluster_name <EOL> host = self . _cluster . metadata . get_host ( connection . host ) <EOL> if host : <EOL> datacenter = local_row . get ( "<STR_LIT>" ) <EOL> rack = local_row . get ( "<STR_LIT>" ) <EOL> self . _update_location_info ( host , datacenter , rack ) <EOL> host . listen_address = local_row . get ( "<STR_LIT>" ) <EOL> host . broadcast_address = local_row . get ( "<STR_LIT>" ) <EOL> host . release_version = local_row . get ( "<STR_LIT>" ) <EOL> partitioner = local_row . get ( "<STR_LIT>" ) <EOL> tokens = local_row . get ( "<STR_LIT>" ) <EOL> if partitioner and tokens : <EOL> token_map [ host ] = tokens <EOL> should_rebuild_token_map = force_token_rebuild or self . _cluster . metadata . partitioner is None <EOL> found_hosts = set ( ) <EOL> for row in peers_result : <EOL> addr = self . _rpc_from_peer_row ( row ) <EOL> tokens = row . get ( "<STR_LIT>" , None ) <EOL> if '<STR_LIT>' in row and not tokens : <EOL> log . warning ( "<STR_LIT>" % ( addr , connection . host ) ) <EOL> continue <EOL> found_hosts . add ( addr ) <EOL> host = self . _cluster . metadata . get_host ( addr ) <EOL> datacenter = row . get ( "<STR_LIT>" ) <EOL> rack = row . get ( "<STR_LIT>" ) <EOL> if host is None : <EOL> log . debug ( "<STR_LIT>" , addr ) <EOL> host , _ = self . _cluster . add_host ( addr , datacenter , rack , signal = True , refresh_nodes = False ) <EOL> should_rebuild_token_map = True <EOL> else : <EOL> should_rebuild_token_map |= self . _update_location_info ( host , datacenter , rack ) <EOL> host . broadcast_address = row . get ( "<STR_LIT>" ) <EOL> host . release_version = row . get ( "<STR_LIT>" ) <EOL> if partitioner and tokens : <EOL> token_map [ host ] = tokens <EOL> for old_host in self . _cluster . metadata . all_hosts ( ) : <EOL> if old_host . address != connection . host and old_host . address not in found_hosts : <EOL> should_rebuild_token_map = True <EOL> if old_host . address not in self . _cluster . contact_points : <EOL> log . debug ( "<STR_LIT>" , old_host ) <EOL> self . _cluster . remove_host ( old_host ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> if partitioner and should_rebuild_token_map : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . _cluster . metadata . rebuild_token_map ( partitioner , token_map ) <EOL> def _update_location_info ( self , host , datacenter , rack ) : <EOL> if host . datacenter == datacenter and host . rack == rack : <EOL> return False <EOL> self . _cluster . load_balancing_policy . on_down ( host ) <EOL> host . set_location_info ( datacenter , rack ) <EOL> self . _cluster . load_balancing_policy . on_up ( host ) <EOL> return True <EOL> def _delay_for_event_type ( self , event_type , delay_window ) : <EOL> next_time = self . _event_schedule_times . get ( event_type , <NUM_LIT:0> ) <EOL> now = self . _time . time ( ) <EOL> if now <= next_time : <EOL> this_time = next_time + <NUM_LIT> <EOL> delay = this_time - now <EOL> else : <EOL> delay = random ( ) * delay_window <EOL> this_time = now + delay <EOL> self . _event_schedule_times [ event_type ] = this_time <EOL> return delay <EOL> def _handle_topology_change ( self , event ) : <EOL> change_type = event [ "<STR_LIT>" ] <EOL> addr = self . _translate_address ( event [ "<STR_LIT:address>" ] [ <NUM_LIT:0> ] ) <EOL> if change_type == "<STR_LIT>" or change_type == "<STR_LIT>" : <EOL> if self . _topology_event_refresh_window >= <NUM_LIT:0> : <EOL> delay = self . _delay_for_event_type ( '<STR_LIT>' , self . _topology_event_refresh_window ) <EOL> self . _cluster . scheduler . schedule_unique ( delay , self . refresh_node_list_and_token_map ) <EOL> elif change_type == "<STR_LIT>" : <EOL> host = self . _cluster . metadata . get_host ( addr ) <EOL> self . _cluster . scheduler . schedule_unique ( <NUM_LIT:0> , self . _cluster . remove_host , host ) <EOL> def _handle_status_change ( self , event ) : <EOL> change_type = event [ "<STR_LIT>" ] <EOL> addr = self . _translate_address ( event [ "<STR_LIT:address>" ] [ <NUM_LIT:0> ] ) <EOL> host = self . _cluster . metadata . get_host ( addr ) <EOL> if change_type == "<STR_LIT>" : <EOL> delay = <NUM_LIT:1> + self . _delay_for_event_type ( '<STR_LIT>' , <NUM_LIT:0.5> ) <EOL> if host is None : <EOL> self . _cluster . scheduler . schedule_unique ( delay , self . refresh_node_list_and_token_map ) <EOL> else : <EOL> self . _cluster . scheduler . schedule_unique ( delay , self . _cluster . on_up , host ) <EOL> elif change_type == "<STR_LIT>" : <EOL> if host is not None : <EOL> self . _cluster . on_down ( host , is_host_addition = False ) <EOL> def _translate_address ( self , addr ) : <EOL> return self . _cluster . address_translator . translate ( addr ) <EOL> def _handle_schema_change ( self , event ) : <EOL> if self . _schema_event_refresh_window < <NUM_LIT:0> : <EOL> return <EOL> delay = self . _delay_for_event_type ( '<STR_LIT>' , self . _schema_event_refresh_window ) <EOL> self . _cluster . scheduler . schedule_unique ( delay , self . refresh_schema , ** event ) <EOL> def wait_for_schema_agreement ( self , connection = None , preloaded_results = None , wait_time = None ) : <EOL> total_timeout = wait_time if wait_time is not None else self . _cluster . max_schema_agreement_wait <EOL> if total_timeout <= <NUM_LIT:0> : <EOL> return True <EOL> with self . _schema_agreement_lock : <EOL> if self . _is_shutdown : <EOL> return <EOL> if not connection : <EOL> connection = self . _connection <EOL> if preloaded_results : <EOL> log . debug ( "<STR_LIT>" ) <EOL> peers_result = preloaded_results [ <NUM_LIT:0> ] <EOL> local_result = preloaded_results [ <NUM_LIT:1> ] <EOL> schema_mismatches = self . _get_schema_mismatches ( peers_result , local_result , connection . host ) <EOL> if schema_mismatches is None : <EOL> return True <EOL> log . debug ( "<STR_LIT>" ) <EOL> start = self . _time . time ( ) <EOL> elapsed = <NUM_LIT:0> <EOL> cl = ConsistencyLevel . ONE <EOL> schema_mismatches = None <EOL> while elapsed < total_timeout : <EOL> peers_query = QueryMessage ( query = self . _SELECT_SCHEMA_PEERS , consistency_level = cl ) <EOL> local_query = QueryMessage ( query = self . _SELECT_SCHEMA_LOCAL , consistency_level = cl ) <EOL> try : <EOL> timeout = min ( self . _timeout , total_timeout - elapsed ) <EOL> peers_result , local_result = connection . wait_for_responses ( <EOL> peers_query , local_query , timeout = timeout ) <EOL> except OperationTimedOut as timeout : <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , timeout ) <EOL> elapsed = self . _time . time ( ) - start <EOL> continue <EOL> except ConnectionShutdown : <EOL> if self . _is_shutdown : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return None <EOL> else : <EOL> raise <EOL> schema_mismatches = self . _get_schema_mismatches ( peers_result , local_result , connection . host ) <EOL> if schema_mismatches is None : <EOL> return True <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . _time . sleep ( <NUM_LIT> ) <EOL> elapsed = self . _time . time ( ) - start <EOL> log . warning ( "<STR_LIT>" , <EOL> connection . host , schema_mismatches ) <EOL> return False <EOL> def _get_schema_mismatches ( self , peers_result , local_result , local_address ) : <EOL> peers_result = dict_factory ( * peers_result . results ) <EOL> versions = defaultdict ( set ) <EOL> if local_result . results : <EOL> local_row = dict_factory ( * local_result . results ) [ <NUM_LIT:0> ] <EOL> if local_row . get ( "<STR_LIT>" ) : <EOL> versions [ local_row . get ( "<STR_LIT>" ) ] . add ( local_address ) <EOL> for row in peers_result : <EOL> schema_ver = row . get ( '<STR_LIT>' ) <EOL> if not schema_ver : <EOL> continue <EOL> addr = self . _rpc_from_peer_row ( row ) <EOL> peer = self . _cluster . metadata . get_host ( addr ) <EOL> if peer and peer . is_up : <EOL> versions [ schema_ver ] . add ( addr ) <EOL> if len ( versions ) == <NUM_LIT:1> : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return None <EOL> return dict ( ( version , list ( nodes ) ) for version , nodes in six . iteritems ( versions ) ) <EOL> def _rpc_from_peer_row ( self , row ) : <EOL> addr = row . get ( "<STR_LIT>" ) <EOL> if not addr or addr in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> addr = row . get ( "<STR_LIT>" ) <EOL> return self . _translate_address ( addr ) <EOL> def _signal_error ( self ) : <EOL> with self . _lock : <EOL> if self . _is_shutdown : <EOL> return <EOL> if self . _connection and self . _connection . is_defunct : <EOL> host = self . _cluster . metadata . get_host ( self . _connection . host ) <EOL> if host : <EOL> self . _cluster . signal_connection_failure ( <EOL> host , self . _connection . last_error , is_host_addition = False ) <EOL> return <EOL> self . reconnect ( ) <EOL> def on_up ( self , host ) : <EOL> pass <EOL> def on_down ( self , host ) : <EOL> conn = self . _connection <EOL> if conn and conn . host == host . address and self . _reconnection_handler is None : <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , host ) <EOL> self . reconnect ( ) <EOL> def on_add ( self , host , refresh_nodes = True ) : <EOL> if refresh_nodes : <EOL> self . refresh_node_list_and_token_map ( force_token_rebuild = True ) <EOL> def on_remove ( self , host ) : <EOL> self . refresh_node_list_and_token_map ( force_token_rebuild = True ) <EOL> def get_connections ( self ) : <EOL> c = getattr ( self , '<STR_LIT>' , None ) <EOL> return [ c ] if c else [ ] <EOL> def return_connection ( self , connection ) : <EOL> if connection is self . _connection and ( connection . is_defunct or connection . is_closed ) : <EOL> self . reconnect ( ) <EOL> def _stop_scheduler ( scheduler , thread ) : <EOL> try : <EOL> if not scheduler . is_shutdown : <EOL> scheduler . shutdown ( ) <EOL> except ReferenceError : <EOL> pass <EOL> thread . join ( ) <EOL> class _Scheduler ( object ) : <EOL> _queue = None <EOL> _scheduled_tasks = None <EOL> _executor = None <EOL> is_shutdown = False <EOL> def __init__ ( self , executor ) : <EOL> self . _queue = Queue . PriorityQueue ( ) <EOL> self . _scheduled_tasks = set ( ) <EOL> self . _count = count ( ) <EOL> self . _executor = executor <EOL> t = Thread ( target = self . run , name = "<STR_LIT>" ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> atexit . register ( partial ( _stop_scheduler , weakref . proxy ( self ) , t ) ) <EOL> def shutdown ( self ) : <EOL> try : <EOL> log . debug ( "<STR_LIT>" ) <EOL> except AttributeError : <EOL> pass <EOL> self . is_shutdown = True <EOL> self . _queue . put_nowait ( ( <NUM_LIT:0> , <NUM_LIT:0> , None ) ) <EOL> def schedule ( self , delay , fn , * args , ** kwargs ) : <EOL> self . _insert_task ( delay , ( fn , args , tuple ( kwargs . items ( ) ) ) ) <EOL> def schedule_unique ( self , delay , fn , * args , ** kwargs ) : <EOL> task = ( fn , args , tuple ( kwargs . items ( ) ) ) <EOL> if task not in self . _scheduled_tasks : <EOL> self . _insert_task ( delay , task ) <EOL> else : <EOL> log . debug ( "<STR_LIT>" , task ) <EOL> def _insert_task ( self , delay , task ) : <EOL> if not self . is_shutdown : <EOL> run_at = time . time ( ) + delay <EOL> self . _scheduled_tasks . add ( task ) <EOL> self . _queue . put_nowait ( ( run_at , next ( self . _count ) , task ) ) <EOL> else : <EOL> log . debug ( "<STR_LIT>" , task ) <EOL> def run ( self ) : <EOL> while True : <EOL> if self . is_shutdown : <EOL> return <EOL> try : <EOL> while True : <EOL> run_at , i , task = self . _queue . get ( block = True , timeout = None ) <EOL> if self . is_shutdown : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return <EOL> if run_at <= time . time ( ) : <EOL> self . _scheduled_tasks . discard ( task ) <EOL> fn , args , kwargs = task <EOL> kwargs = dict ( kwargs ) <EOL> future = self . _executor . submit ( fn , * args , ** kwargs ) <EOL> future . add_done_callback ( self . _log_if_failed ) <EOL> else : <EOL> self . _queue . put_nowait ( ( run_at , i , task ) ) <EOL> break <EOL> except Queue . Empty : <EOL> pass <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def _log_if_failed ( self , future ) : <EOL> exc = future . exception ( ) <EOL> if exc : <EOL> log . warning ( <EOL> "<STR_LIT>" , <EOL> exc_info = exc ) <EOL> def refresh_schema_and_set_result ( control_conn , response_future , ** kwargs ) : <EOL> try : <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT:%s>" , kwargs ) <EOL> response_future . is_schema_agreed = control_conn . _refresh_schema ( response_future . _connection , ** kwargs ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> response_future . session . submit ( control_conn . refresh_schema , ** kwargs ) <EOL> finally : <EOL> response_future . _set_final_result ( None ) <EOL> class ResponseFuture ( object ) : <EOL> """<STR_LIT>""" <EOL> query = None <EOL> """<STR_LIT>""" <EOL> is_schema_agreed = True <EOL> """<STR_LIT>""" <EOL> session = None <EOL> row_factory = None <EOL> message = None <EOL> default_timeout = None <EOL> _req_id = None <EOL> _final_result = _NOT_SET <EOL> _col_names = None <EOL> _final_exception = None <EOL> _query_traces = None <EOL> _callbacks = None <EOL> _errbacks = None <EOL> _current_host = None <EOL> _current_pool = None <EOL> _connection = None <EOL> _query_retries = <NUM_LIT:0> <EOL> _start_time = None <EOL> _metrics = None <EOL> _paging_state = None <EOL> _custom_payload = None <EOL> _warnings = None <EOL> _timer = None <EOL> _protocol_handler = ProtocolHandler <EOL> _warned_timeout = False <EOL> def __init__ ( self , session , message , query , timeout , metrics = None , prepared_statement = None ) : <EOL> self . session = session <EOL> self . row_factory = session . row_factory <EOL> self . message = message <EOL> self . query = query <EOL> self . timeout = timeout <EOL> self . _metrics = metrics <EOL> self . prepared_statement = prepared_statement <EOL> self . _callback_lock = Lock ( ) <EOL> if metrics is not None : <EOL> self . _start_time = time . time ( ) <EOL> self . _make_query_plan ( ) <EOL> self . _event = Event ( ) <EOL> self . _errors = { } <EOL> self . _callbacks = [ ] <EOL> self . _errbacks = [ ] <EOL> def _start_timer ( self ) : <EOL> if self . timeout is not None : <EOL> self . _timer = self . session . cluster . connection_class . create_timer ( self . timeout , self . _on_timeout ) <EOL> def _cancel_timer ( self ) : <EOL> if self . _timer : <EOL> self . _timer . cancel ( ) <EOL> def _on_timeout ( self ) : <EOL> errors = self . _errors <EOL> if not errors : <EOL> if self . is_schema_agreed : <EOL> errors = { self . _current_host . address : "<STR_LIT>" } <EOL> else : <EOL> connection = getattr ( self . session . cluster . control_connection , '<STR_LIT>' ) <EOL> host = connection . host if connection else '<STR_LIT>' <EOL> errors = { host : "<STR_LIT>" } <EOL> self . _set_final_exception ( OperationTimedOut ( errors , self . _current_host ) ) <EOL> def _make_query_plan ( self ) : <EOL> self . query_plan = iter ( self . session . _load_balancer . make_query_plan ( <EOL> self . session . keyspace , self . query ) ) <EOL> def send_request ( self ) : <EOL> """<STR_LIT>""" <EOL> start = time . time ( ) <EOL> for host in self . query_plan : <EOL> req_id = self . _query ( host ) <EOL> if req_id is not None : <EOL> self . _req_id = req_id <EOL> if self . _timer is None : <EOL> self . _start_timer ( ) <EOL> return <EOL> if self . timeout is not None and time . time ( ) - start > self . timeout : <EOL> self . _on_timeout ( ) <EOL> return <EOL> self . _set_final_exception ( NoHostAvailable ( <EOL> "<STR_LIT>" , self . _errors ) ) <EOL> def _query ( self , host , message = None , cb = None ) : <EOL> if message is None : <EOL> message = self . message <EOL> if cb is None : <EOL> cb = self . _set_result <EOL> pool = self . session . _pools . get ( host ) <EOL> if not pool : <EOL> self . _errors [ host ] = ConnectionException ( "<STR_LIT>" ) <EOL> return None <EOL> elif pool . is_shutdown : <EOL> self . _errors [ host ] = ConnectionException ( "<STR_LIT>" ) <EOL> return None <EOL> self . _current_host = host <EOL> self . _current_pool = pool <EOL> connection = None <EOL> try : <EOL> connection , request_id = pool . borrow_connection ( timeout = <NUM_LIT> ) <EOL> self . _connection = connection <EOL> connection . send_msg ( message , request_id , cb = cb , encoder = self . _protocol_handler . encode_message , decoder = self . _protocol_handler . decode_message ) <EOL> return request_id <EOL> except NoConnectionsAvailable as exc : <EOL> log . debug ( "<STR_LIT>" , host ) <EOL> self . _errors [ host ] = exc <EOL> return None <EOL> except Exception as exc : <EOL> log . debug ( "<STR_LIT>" , host , exc_info = True ) <EOL> self . _errors [ host ] = exc <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_connection_error ( ) <EOL> if connection : <EOL> pool . return_connection ( connection ) <EOL> return None <EOL> @ property <EOL> def has_more_pages ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _paging_state is not None <EOL> @ property <EOL> def warnings ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _event . is_set ( ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> return self . _warnings <EOL> @ property <EOL> def custom_payload ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _event . is_set ( ) : <EOL> raise DriverException ( "<STR_LIT>" ) <EOL> return self . _custom_payload <EOL> def start_fetching_next_page ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _paging_state : <EOL> raise QueryExhausted ( ) <EOL> self . _make_query_plan ( ) <EOL> self . message . paging_state = self . _paging_state <EOL> self . _event . clear ( ) <EOL> self . _final_result = _NOT_SET <EOL> self . _final_exception = None <EOL> self . _timer = None <EOL> self . send_request ( ) <EOL> def _reprepare ( self , prepare_message ) : <EOL> cb = partial ( self . session . submit , self . _execute_after_prepare ) <EOL> request_id = self . _query ( self . _current_host , prepare_message , cb = cb ) <EOL> if request_id is None : <EOL> self . send_request ( ) <EOL> def _set_result ( self , response ) : <EOL> try : <EOL> if self . _current_pool and self . _connection : <EOL> self . _current_pool . return_connection ( self . _connection ) <EOL> trace_id = getattr ( response , '<STR_LIT>' , None ) <EOL> if trace_id : <EOL> if not self . _query_traces : <EOL> self . _query_traces = [ ] <EOL> self . _query_traces . append ( QueryTrace ( trace_id , self . session ) ) <EOL> self . _warnings = getattr ( response , '<STR_LIT>' , None ) <EOL> self . _custom_payload = getattr ( response , '<STR_LIT>' , None ) <EOL> if isinstance ( response , ResultMessage ) : <EOL> if response . kind == RESULT_KIND_SET_KEYSPACE : <EOL> session = getattr ( self , '<STR_LIT>' , None ) <EOL> if session : <EOL> session . _set_keyspace_for_all_pools ( <EOL> response . results , self . _set_keyspace_completed ) <EOL> elif response . kind == RESULT_KIND_SCHEMA_CHANGE : <EOL> self . is_schema_agreed = False <EOL> self . session . submit ( <EOL> refresh_schema_and_set_result , <EOL> self . session . cluster . control_connection , <EOL> self , ** response . results ) <EOL> else : <EOL> results = getattr ( response , '<STR_LIT>' , None ) <EOL> if results is not None and response . kind == RESULT_KIND_ROWS : <EOL> self . _paging_state = response . paging_state <EOL> self . _col_names = results [ <NUM_LIT:0> ] <EOL> results = self . row_factory ( * results ) <EOL> self . _set_final_result ( results ) <EOL> elif isinstance ( response , ErrorMessage ) : <EOL> retry_policy = None <EOL> if self . query : <EOL> retry_policy = self . query . retry_policy <EOL> if not retry_policy : <EOL> retry_policy = self . session . cluster . default_retry_policy <EOL> if isinstance ( response , ReadTimeoutErrorMessage ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_read_timeout ( ) <EOL> retry = retry_policy . on_read_timeout ( <EOL> self . query , retry_num = self . _query_retries , ** response . info ) <EOL> elif isinstance ( response , WriteTimeoutErrorMessage ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_write_timeout ( ) <EOL> retry = retry_policy . on_write_timeout ( <EOL> self . query , retry_num = self . _query_retries , ** response . info ) <EOL> elif isinstance ( response , UnavailableErrorMessage ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_unavailable ( ) <EOL> retry = retry_policy . on_unavailable ( <EOL> self . query , retry_num = self . _query_retries , ** response . info ) <EOL> elif isinstance ( response , OverloadedErrorMessage ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_other_error ( ) <EOL> log . warning ( "<STR_LIT>" <EOL> "<STR_LIT:host>" , self . _current_host ) <EOL> self . _retry ( reuse_connection = False , consistency_level = None ) <EOL> return <EOL> elif isinstance ( response , IsBootstrappingErrorMessage ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_other_error ( ) <EOL> self . _retry ( reuse_connection = False , consistency_level = None ) <EOL> return <EOL> elif isinstance ( response , PreparedQueryNotFound ) : <EOL> if self . prepared_statement : <EOL> query_id = self . prepared_statement . query_id <EOL> assert query_id == response . info , "<STR_LIT>" "<STR_LIT>" % ( response . info , query_id ) <EOL> else : <EOL> query_id = response . info <EOL> try : <EOL> prepared_statement = self . session . cluster . _prepared_statements [ query_id ] <EOL> except KeyError : <EOL> if not self . prepared_statement : <EOL> log . error ( "<STR_LIT>" , <EOL> query_id . encode ( '<STR_LIT>' ) ) <EOL> self . _set_final_exception ( response ) <EOL> return <EOL> else : <EOL> prepared_statement = self . prepared_statement <EOL> self . session . cluster . _prepared_statements [ query_id ] = prepared_statement <EOL> current_keyspace = self . _connection . keyspace <EOL> prepared_keyspace = prepared_statement . keyspace <EOL> if prepared_keyspace and current_keyspace != prepared_keyspace : <EOL> self . _set_final_exception ( <EOL> ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( current_keyspace , prepared_keyspace ) ) ) <EOL> return <EOL> log . debug ( "<STR_LIT>" , <EOL> self . _current_host , prepared_statement . query_string ) <EOL> prepare_message = PrepareMessage ( query = prepared_statement . query_string ) <EOL> self . session . submit ( self . _reprepare , prepare_message ) <EOL> return <EOL> else : <EOL> if hasattr ( response , '<STR_LIT>' ) : <EOL> self . _set_final_exception ( response . to_exception ( ) ) <EOL> else : <EOL> self . _set_final_exception ( response ) <EOL> return <EOL> retry_type , consistency = retry <EOL> if retry_type in ( RetryPolicy . RETRY , RetryPolicy . RETRY_NEXT_HOST ) : <EOL> self . _query_retries += <NUM_LIT:1> <EOL> reuse = retry_type == RetryPolicy . RETRY <EOL> self . _retry ( reuse_connection = reuse , consistency_level = consistency ) <EOL> elif retry_type is RetryPolicy . RETHROW : <EOL> self . _set_final_exception ( response . to_exception ( ) ) <EOL> else : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_ignore ( ) <EOL> self . _set_final_result ( None ) <EOL> self . _errors [ self . _current_host ] = response . to_exception ( ) <EOL> elif isinstance ( response , ConnectionException ) : <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_connection_error ( ) <EOL> if not isinstance ( response , ConnectionShutdown ) : <EOL> self . _connection . defunct ( response ) <EOL> self . _retry ( reuse_connection = False , consistency_level = None ) <EOL> elif isinstance ( response , Exception ) : <EOL> if hasattr ( response , '<STR_LIT>' ) : <EOL> self . _set_final_exception ( response . to_exception ( ) ) <EOL> else : <EOL> self . _set_final_exception ( response ) <EOL> else : <EOL> msg = "<STR_LIT>" % ( response , ) <EOL> exc = ConnectionException ( msg , self . _current_host ) <EOL> self . _connection . defunct ( exc ) <EOL> self . _set_final_exception ( exc ) <EOL> except Exception as exc : <EOL> log . exception ( "<STR_LIT>" ) <EOL> self . _set_final_exception ( exc ) <EOL> def _set_keyspace_completed ( self , errors ) : <EOL> if not errors : <EOL> self . _set_final_result ( None ) <EOL> else : <EOL> self . _set_final_exception ( ConnectionException ( <EOL> "<STR_LIT>" % ( errors , ) ) ) <EOL> def _execute_after_prepare ( self , response ) : <EOL> """<STR_LIT>""" <EOL> if self . _current_pool and self . _connection : <EOL> self . _current_pool . return_connection ( self . _connection ) <EOL> if self . _final_exception : <EOL> return <EOL> if isinstance ( response , ResultMessage ) : <EOL> if response . kind == RESULT_KIND_PREPARED : <EOL> request_id = self . _query ( self . _current_host ) <EOL> if request_id is None : <EOL> self . send_request ( ) <EOL> else : <EOL> self . _set_final_exception ( ConnectionException ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . _current_host , response ) ) ) <EOL> elif isinstance ( response , ErrorMessage ) : <EOL> if hasattr ( response , '<STR_LIT>' ) : <EOL> self . _set_final_exception ( response . to_exception ( ) ) <EOL> else : <EOL> self . _set_final_exception ( response ) <EOL> elif isinstance ( response , ConnectionException ) : <EOL> log . debug ( "<STR_LIT>" , <EOL> self . _current_host , response ) <EOL> self . _errors [ self . _current_host ] = response <EOL> self . send_request ( ) <EOL> else : <EOL> self . _set_final_exception ( ConnectionException ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . _current_host , response ) ) ) <EOL> def _set_final_result ( self , response ) : <EOL> self . _cancel_timer ( ) <EOL> if self . _metrics is not None : <EOL> self . _metrics . request_timer . addValue ( time . time ( ) - self . _start_time ) <EOL> with self . _callback_lock : <EOL> self . _final_result = response <EOL> self . _event . set ( ) <EOL> for callback in self . _callbacks : <EOL> fn , args , kwargs = callback <EOL> fn ( response , * args , ** kwargs ) <EOL> def _set_final_exception ( self , response ) : <EOL> self . _cancel_timer ( ) <EOL> if self . _metrics is not None : <EOL> self . _metrics . request_timer . addValue ( time . time ( ) - self . _start_time ) <EOL> with self . _callback_lock : <EOL> self . _final_exception = response <EOL> self . _event . set ( ) <EOL> for errback in self . _errbacks : <EOL> fn , args , kwargs = errback <EOL> fn ( response , * args , ** kwargs ) <EOL> def _retry ( self , reuse_connection , consistency_level ) : <EOL> if self . _final_exception : <EOL> return <EOL> if self . _metrics is not None : <EOL> self . _metrics . on_retry ( ) <EOL> if consistency_level is not None : <EOL> self . message . consistency_level = consistency_level <EOL> self . session . submit ( self . _retry_task , reuse_connection ) <EOL> def _retry_task ( self , reuse_connection ) : <EOL> if self . _final_exception : <EOL> return <EOL> if reuse_connection and self . _query ( self . _current_host ) is not None : <EOL> return <EOL> self . send_request ( ) <EOL> def result ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _event . wait ( ) <EOL> if self . _final_result is not _NOT_SET : <EOL> return ResultSet ( self , self . _final_result ) <EOL> else : <EOL> raise self . _final_exception <EOL> def get_query_trace_ids ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ trace . trace_id for trace in self . _query_traces ] <EOL> def get_query_trace ( self , max_wait = None , query_cl = ConsistencyLevel . LOCAL_ONE ) : <EOL> """<STR_LIT>""" <EOL> if self . _query_traces : <EOL> return self . _get_query_trace ( len ( self . _query_traces ) - <NUM_LIT:1> , max_wait , query_cl ) <EOL> def get_all_query_traces ( self , max_wait_per = None , query_cl = ConsistencyLevel . LOCAL_ONE ) : <EOL> """<STR_LIT>""" <EOL> if self . _query_traces : <EOL> return [ self . _get_query_trace ( i , max_wait_per , query_cl ) for i in range ( len ( self . _query_traces ) ) ] <EOL> return [ ] <EOL> def _get_query_trace ( self , i , max_wait , query_cl ) : <EOL> trace = self . _query_traces [ i ] <EOL> if not trace . events : <EOL> trace . populate ( max_wait = max_wait , query_cl = query_cl ) <EOL> return trace <EOL> def add_callback ( self , fn , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> run_now = False <EOL> with self . _callback_lock : <EOL> if self . _final_result is not _NOT_SET : <EOL> run_now = True <EOL> else : <EOL> self . _callbacks . append ( ( fn , args , kwargs ) ) <EOL> if run_now : <EOL> fn ( self . _final_result , * args , ** kwargs ) <EOL> return self <EOL> def add_errback ( self , fn , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> run_now = False <EOL> with self . _callback_lock : <EOL> if self . _final_exception : <EOL> run_now = True <EOL> else : <EOL> self . _errbacks . append ( ( fn , args , kwargs ) ) <EOL> if run_now : <EOL> fn ( self . _final_exception , * args , ** kwargs ) <EOL> return self <EOL> def add_callbacks ( self , callback , errback , <EOL> callback_args = ( ) , callback_kwargs = None , <EOL> errback_args = ( ) , errback_kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> self . add_callback ( callback , * callback_args , ** ( callback_kwargs or { } ) ) <EOL> self . add_errback ( errback , * errback_args , ** ( errback_kwargs or { } ) ) <EOL> def clear_callbacks ( self ) : <EOL> with self . _callback_lock : <EOL> self . _callback = [ ] <EOL> self . _errback = [ ] <EOL> def __str__ ( self ) : <EOL> result = "<STR_LIT>" if self . _final_result is _NOT_SET else self . _final_result <EOL> return "<STR_LIT>" % ( self . query , self . _req_id , result , self . _final_exception , self . _current_host ) <EOL> __repr__ = __str__ <EOL> class QueryExhausted ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ResultSet ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , response_future , initial_response ) : <EOL> self . response_future = response_future <EOL> self . column_names = response_future . _col_names <EOL> self . _set_current_rows ( initial_response ) <EOL> self . _page_iter = None <EOL> self . _list_mode = False <EOL> @ property <EOL> def has_more_pages ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . response_future . has_more_pages <EOL> @ property <EOL> def current_rows ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _current_rows or [ ] <EOL> def __iter__ ( self ) : <EOL> if self . _list_mode : <EOL> return iter ( self . _current_rows ) <EOL> self . _page_iter = iter ( self . _current_rows ) <EOL> return self <EOL> def next ( self ) : <EOL> try : <EOL> return next ( self . _page_iter ) <EOL> except StopIteration : <EOL> if not self . response_future . has_more_pages : <EOL> if not self . _list_mode : <EOL> self . _current_rows = [ ] <EOL> raise <EOL> self . fetch_next_page ( ) <EOL> self . _page_iter = iter ( self . _current_rows ) <EOL> return next ( self . _page_iter ) <EOL> __next__ = next <EOL> def fetch_next_page ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . response_future . has_more_pages : <EOL> self . response_future . start_fetching_next_page ( ) <EOL> result = self . response_future . result ( ) <EOL> self . _current_rows = result . _current_rows <EOL> else : <EOL> self . _current_rows = [ ] <EOL> def _set_current_rows ( self , result ) : <EOL> if isinstance ( result , Mapping ) : <EOL> self . _current_rows = [ result ] if result else [ ] <EOL> return <EOL> try : <EOL> iter ( result ) <EOL> self . _current_rows = result <EOL> except TypeError : <EOL> self . _current_rows = [ result ] if result else [ ] <EOL> def _fetch_all ( self ) : <EOL> self . _current_rows = list ( self ) <EOL> self . _page_iter = None <EOL> def _enter_list_mode ( self , operator ) : <EOL> if self . _list_mode : <EOL> return <EOL> if self . _page_iter : <EOL> raise RuntimeError ( "<STR_LIT>" % operator ) <EOL> if self . response_future . has_more_pages : <EOL> log . warning ( "<STR_LIT>" , operator ) <EOL> self . _fetch_all ( ) <EOL> self . _list_mode = True <EOL> def __eq__ ( self , other ) : <EOL> self . _enter_list_mode ( "<STR_LIT>" ) <EOL> return self . _current_rows == other <EOL> def __getitem__ ( self , i ) : <EOL> self . _enter_list_mode ( "<STR_LIT>" ) <EOL> return self . _current_rows [ i ] <EOL> def __nonzero__ ( self ) : <EOL> return bool ( self . _current_rows ) <EOL> __bool__ = __nonzero__ <EOL> def get_query_trace ( self , max_wait_sec = None ) : <EOL> """<STR_LIT>""" <EOL> return self . response_future . get_query_trace ( max_wait_sec ) <EOL> def get_all_query_traces ( self , max_wait_sec_per = None ) : <EOL> """<STR_LIT>""" <EOL> return self . response_future . get_all_query_traces ( max_wait_sec_per ) <EOL> @ property <EOL> def was_applied ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . response_future . row_factory not in ( named_tuple_factory , dict_factory , tuple_factory ) : <EOL> raise RuntimeError ( "<STR_LIT>" % ( self . response_future . row_factsory , ) ) <EOL> if len ( self . current_rows ) != <NUM_LIT:1> : <EOL> raise RuntimeError ( "<STR_LIT>" % ( len ( self . current_rows ) ) ) <EOL> row = self . current_rows [ <NUM_LIT:0> ] <EOL> if isinstance ( row , tuple ) : <EOL> return row [ <NUM_LIT:0> ] <EOL> else : <EOL> return row [ '<STR_LIT>' ] </s>
