o NDi~ @sdZddlmZddlZddlmZddlmZddlm Z ddl m Z m Z m ZddlmZdd lmZmZmZdd lmZmZmZddlZddlZdd lmZdd lmZddlZddlZddlZddl Z ddl!Z!dd l"m#Z#m$Z$m%Z%m&Z&ddl'Z'ddl(Z(ddl(m)Z)ddl*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3ddl4m5Z5m6Z6ddl7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@mAZAddlBmCZCddlDmEZEddlFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNmOZOmPZPmQZQmRZRmSZSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[ddl\m]Z]m^Z^m_Z_m`Z`ddlambZbmcZcmdZdmeZemfZfmgZgmhZhmiZimjZjmkZkmlZlddlmmnZnmoZompZpmqZqmrZrmsZsddltmuZumvZvmwZwmxZxmyZymzZzm{Z{m|Z|m}Z}m~Z~mZmZddlmZddlmZddlmZmZddlmZddlmZddlmZmZmZmZmZmZmZmZmZmZddlmZmZdd lmZzdd!lmZWn eydZYnwzdd"lmZWneefydZYnwzdd#l(mZWneydd#lmZYnwd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1ZeeZeeeefZeeedgf\ZZes e3d2eeZd3d4d5Zd6Zd7Zd8Zd9Zd7Zed:ZeZGd;d<dZd?d@ZeZdAdBZdCdDZdEdFZeeġdGdHZGdIdJdJeZdKdLZdMdNZGdOdPdPeZGdQdRdReʃZGdSdTdTe˃ZGdUdVdVeZeZ eZ eZ eZ GdWdXdXeZGdYdZdZeZGd[d\d\eZGd]d^d^eZGd_d`d`eoZdadbZdcddZGdedfdfeZdgdhZGdidjdje%ZdkdlZGdmdndneZGdodpdpeZGdqdrdreZdS)szf This module houses the main classes you will interact with, :class:`.Cluster` and :class:`.Session`. )absolute_importN)hexlify) defaultdict)Mapping)ThreadPoolExecutorFIRST_COMPLETEDwait)copy)partialreducewraps)groupbycountchain)warn)random)LockRLockThreadEvent)WeakValueDictionary) ConsistencyLevelAuthenticationFailedOperationTimedOutUnsupportedOperationSchemaTargetTypeDriverExceptionProtocolVersionUnresolvableContactPointsDependencyException)_proxy_execute_keyPlainTextAuthProvider) ConnectionExceptionConnectionShutdownConnectionHeartbeatProtocolVersionUnsupportedEndPointDefaultEndPointDefaultEndPointFactoryContinuousPagingStateSniEndPointFactoryConnectionBusy)UserType)Encoder) QueryMessage ResultMessage ErrorMessageReadTimeoutErrorMessageWriteTimeoutErrorMessageUnavailableErrorMessageOverloadedErrorMessagePrepareMessageExecuteMessagePreparedQueryNotFoundIsBootstrappingErrorMessage TruncateError ServerError BatchMessageRESULT_KIND_PREPAREDRESULT_KIND_SET_KEYSPACERESULT_KIND_ROWSRESULT_KIND_SCHEMA_CHANGEProtocolHandlerRESULT_KIND_VOIDProtocolException)Metadata protect_namemurmur3 _NodeInfo) TokenAwarePolicyDCAwareRoundRobinPolicySimpleConvictionPolicyExponentialReconnectionPolicy HostDistance RetryPolicyIdentityTranslatorNoSpeculativeExecutionPlanNoSpeculativeExecutionPolicyDefaultLoadBalancingPolicyNeverRetryPolicy)Host_ReconnectionHandler_HostReconnectionHandlerHostConnectionPoolHostConnectionNoConnectionsAvailable) SimpleStatementPreparedStatementBoundStatementBatchStatement bind_params QueryTraceTraceUnavailablenamed_tuple_factory dict_factory tuple_factoryFETCH_SIZE_UNSETHostTargetingStatement) int64_pack)MonotonicTimestampGenerator)%_resolve_contact_points_to_string_mapVersion)MonitorReporter)version_supports_insights) graph_object_row_factory GraphOptionsGraphSON1Serializer GraphProtocolGraphSON2SerializerGraphStatementSimpleGraphStatementgraph_graphson2_row_factorygraph_graphson3_row_factoryGraphSON3Serializer)_request_timeout_key_GraphSONContextRowFactory)cloud)TwistedConnectionEventletConnection)WeakSetcCs$dtjvrdSddl}tj|jjuS)Nz gevent.monkeyFr)sysmodules gevent.socketsocket)geventrJ/var/www/Datamplify/venv/lib/python3.10/site-packages/cassandra/cluster.py_is_gevent_monkey_patchedrs rcCtr ddlm}|dfSdS)NrGeventConnectionNN)rcassandra.io.geventreactorrrrrr_try_gevent_importx rcCs:dtjvrdSz ddl}|jdWStyYdSw)Nzeventlet.patcherFrr~)r{r|eventlet.patcherpatcheris_monkey_patchedAttributeError)eventletrrr_is_eventlet_monkey_patcheds  rcCr)Nrrxr)rcassandra.io.eventletreactorryrxrrr_try_eventlet_importrrc CDz ddlm}|dfWSty!}z d|fWYd}~Sd}~ww)Nr)LibevConnection)cassandra.io.libevreactorrr)rerrr_try_libev_import  rc Cr)Nr)AsyncoreConnection)cassandra.io.asyncorereactorrr)rrrrr_try_asyncore_importrrcCs4|\}}|r|S|\}}|r|||p||fSN)append)val import_fnrvexcs import_resultexcrrr_connection_reduce_fns   rz)Unable to load a default connection classutf8dz6.8.0c@seZdZdZdZ ddZdS)NoHostAvailablez Raised when an operation is attempted but all connections are busy, defunct, closed, or resulted in errors when used. NcCst|||||_dSr) Exception__init__errors)selfmessagerrrrrs zNoHostAvailable.__init__)__name__ __module__ __qualname____doc__rrrrrrrs  rcCs"|}|rtjd|ddSdS)z Helper for run_in_executor() zFailed to run task on executorexc_infoN) exceptionlogdebugfuturerrrr_future_completedsrcstfdd}|S)zH A decorator to run the given method in the ThreadPoolExecutor. csV|jrdSz|jj|g|Ri|}|tWdSty*tdYdSw)Nz!Failed to submit task to executor) is_shutdownexecutorsubmitadd_done_callbackrrrr)rargskwargsrfrrnew_fs zrun_in_executor..new_f)r )rrrrrrun_in_executors rcCt|dSr)_clusters_for_shutdownaddclusterrrr_register_cluster_shutdownrcCrr)rdiscardrrrr_discard_cluster_shutdownrrcCst}|D]}|qdSr)rr shutdown)clustersrrrr_shutdown_clusterss rcCstdur ttStSr)rErGrHrrrrdefault_lbp_factorys rc@sPeZdZGdddeZdZ dZ dZ dZ ej dddfddZ dd Z dS) ContinuousPagingOptionsc@seZdZdZdZdS)z"ContinuousPagingOptions.PagingUnitrrN)rrrBYTESROWSrrrr PagingUnit srNrcCs,||_||_||_|dkrtd||_dS)Nrz;ContinuousPagingOptions.max_queue_size must be 2 or greater) page_unit max_pagesmax_pages_per_second ValueErrormax_queue_size)rrrrrrrrr(s  z ContinuousPagingOptions.__init__cCs|jtjjkSr)rrrrrrrrpage_unit_bytes0rz'ContinuousPagingOptions.page_unit_bytes) rrrobjectrrrrrrrrrrrrr s rc Cs@z t||tjtjWStjytd||YdSw)z A helper function that wraps socket.getaddrinfo and returns None when it fails to, e.g. resolve one of the hostnames. Used to address PYTHON-895. z,Could not resolve hostname "{}" with port {}N)r~ getaddrinfo AF_UNSPEC SOCK_STREAMgaierrorrrformat) contact_pointportrrr_addrinfo_or_none4srcCs.tdtdtdtdi}||vr||Sd|fS)NEXEC_PROFILE_DEFAULTEXEC_PROFILE_GRAPH_DEFAULT!EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT$EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULTz"%s")rrrr)namedefault_profilesrrr_execution_profile_to_stringCs rc@sdeZdZdZ dZ ejZ dZ dZ e e Z dZ dZ dZdZededde ddfddZdS)ExecutionProfileN$@Fc Cs|tur d|_t|_nd|_||_|turd|_tj|_nd|_||_|p(t|_ |dur7t |s7t d||_ ||_ ||_|pDt|_||_dS)NFTzaserial_consistency_level must be either ConsistencyLevel.SERIAL or ConsistencyLevel.LOCAL_SERIAL.)_NOT_SET_load_balancing_policy_explicitrload_balancing_policy_consistency_level_explicitr LOCAL_ONEconsistency_levelrL retry_policy is_serialrserial_consistency_levelrequest_timeout row_factoryrOspeculative_execution_policycontinuous_paging_options) rrrrrrrrrrrrrs(     zExecutionProfile.__init__)rrrrrrrrrr staticmethodr_rrrrrrrrrrrrQs0 rcs4eZdZdZ ededdddeffdd ZZS)GraphExecutionProfileNg>@c s>|pt}tt|j|||||||d|ptddd|_dS)ay Default execution profile for graph execution. See :class:`.ExecutionProfile` for base attributes. Note that if not explicitly set, the row_factory and graph_options.graph_protocol are resolved during the query execution. These options will resolve to graph_graphson3_row_factory and GraphProtocol.GRAPHSON_3_0 for the core graph engine (DSE 6.8+), otherwise graph_object_row_factory and GraphProtocol.GRAPHSON_1_0 In addition to default parameters shown in the signature, this profile also defaults ``retry_policy`` to :class:`cassandra.policies.NeverRetryPolicy`. )rggremlin-groovy) graph_sourcegraph_languageN)rQsuperrrrk graph_options) rrrrrrrrr __class__rrrs  zGraphExecutionProfile.__init__)rrrrrr __classcell__rrrrrs rcs,eZdZddeddddffdd ZZS)GraphAnalyticsExecutionProfileNgu"Ac sF|ptt}|p tdd}tt|||||||||jdS)aR Execution profile with timeout and load balancing appropriate for graph analytics queries. See also :class:`~.GraphExecutionPolicy`. In addition to default parameters shown in the signature, this profile also defaults ``retry_policy`` to :class:`cassandra.policies.NeverRetryPolicy`, and ``load_balancing_policy`` to one that targets the current Spark master. Note: The graph_options.graph_source is set automatically to b'a' (analytics) when using GraphAnalyticsExecutionProfile. This is mandatory to target analytics nodes. r)rN)rPrrkrr rrset_source_analytics)rrrrrrrrrrrrsz'GraphAnalyticsExecutionProfile.__init__)rrrrrr rrrrr s r c@s`eZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ e ddZ dS)ProfileManagercCs t|_dSr)dictprofilesrrrrr zProfileManager.__init__cCs&dd|jD}tdd|DS)Ncss|] \}}|js|VqdSr)r).0 profile_nameprofilerrr s zAProfileManager._profiles_without_explicit_lbps..css |] }|tur dn|VqdS)rN)r)rnrrrrs  )ritemstuple)rnamesrrr_profiles_without_explicit_lbpss z.ProfileManager._profiles_without_explicit_lbpscsBtfdd|jD}tj|vrtjStj|vrtjStjS)Nc3s|] }|jVqdSr)rdistance)rphostrrrsz*ProfileManager.distance..)setrvaluesrKLOCALREMOTEIGNORED)rr distancesrrrrs zProfileManager.distancecCs"|jD] }|j||qdSr)rrrpopulate)rrhostsrrrrr# szProfileManager.populatecCs|jD]}|jqdSr)rrrcheck_supported)rrrrrr%s zProfileManager.check_supportedcC |jD]}|j|qdSr)rrron_uprrrrrrr'zProfileManager.on_upcCr&r)rrron_downr(rrrr*r)zProfileManager.on_downcCr&r)rrron_addr(rrrr+r)zProfileManager.on_addcCr&r)rrr on_remover(rrrr, r)zProfileManager.on_removecCs |jtS)zc internal-only; no checks are done because this entry is populated on cluster init )rrrrrrdefault$s zProfileManager.defaultN)rrrrrrr#r%r'r*r+r,propertyr-rrrrr s r c@eZdZdZdZdZdS) _ConfigModerrrN)rrr UNCOMMITTEDLEGACYPROFILESrrrrr0Nr0c)@seZdZdZdgZ dZdZ dZ ej Z dZ dZ dZ dZdZeddZejd dZdZed d Zejd d Zed dZeddZ eZeddZejddZeZ eZ dZ dZ dZ! dZ" dZ# dZ$ dZ% dZ& e'Z( dZ) dZ* dZ+ dZ, dZ- dZ. dZ/ dZ0 dZ1 dZ2 dZ3 dZ4 dZ5 dZ6 dZ7 dZ8 dZ9 eddZ:e:jddZ:eddZ;e;jddZ;dZ< dZ=e>j?Z@dZAdZBdZCdZDdZEdZFdZGdZHdZIdZJdZKdZL dZMdZNeOddddddddddddeOdddddddddddddddddddddddddddf)d d!ZPd"d#ZQd$d%ZRdd&d'ZSd(d)ZTd*d+ZUd,d-ZVd.d/ZWd0d1ZXd2d3ZYd4d5ZZd6d7Z[d8d9Z\d:d;Z]dd?Z_dd@dAZ`dBdCZadDdEZbdFdGZcdHdIZddJdKZedLdMZfdNdOZgdPdQZhdRdSZidTdUZjdVdWZkelddXdYZmddZd[Zndd\d]Zod^d_Zpdd`daZqddbdcZrdddeZsdfdgZtdhdiZuedjdkZvdldmZwexdndoZyexdpdqZzdrdsZ{ddtduZ|ddvdwZ}ddxdyZ~ddzd{Zdd|d}Zdd~dZdddZdddZddZedddZddZddZdS)Clusteraf 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. 127.0.0.1NiR#FTcC|jS)a 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`. )_auth_providerrrrr auth_provider zCluster.auth_providercCs`|s||_dSz|j|_Wnty*|jdkrtdt|s%td||_Ynw||_dS)Nrzaauth_provider must implement the cassandra.auth.AuthProvider interface when protocol_version >= 2z9auth_provider must be callable when protocol_version == 1)r8new_authenticator_auth_provider_callablerprotocol_version TypeErrorcallable)rvaluerrrr9s     cCr7)a/ 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.** )_load_balancing_policyrrrrrszCluster.load_balancing_policycC&|jtjkr td||_tj|_dS)NzkCannot set Cluster.load_balancing_policy while using Configuration Profiles. Set this in a profile instead.) _config_moder0r3rrAr2)rlbprrrr  cCs |jjjSr)profile_managerr-rrrrr_default_load_balancing_policys z&Cluster._default_load_balancing_policyg?g@cCr7)z A default :class:`.policies.RetryPolicy` instance to use for all :class:`.Statement` objects which do not have a :attr:`~.Statement.retry_policy` explicitly set. )_default_retry_policyrrrrdefault_retry_policyszCluster.default_retry_policycCrB)NzjCannot set Cluster.default_retry_policy while using Configuration Profiles. Set this in a profile instead.)rCr0r3rrHr2)rpolicyrrrrIrE @rrrcC|jjS)a 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. )control_connection_schema_meta_enabledrrrrschema_metadata_enableds zCluster.schema_metadata_enabledcCt||j_dSr)boolrOrPrenabledrrrrQ'cCrN)a 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. )rO_token_meta_enabledrrrrtoken_metadata_enabled+s zCluster.token_metadata_enabledcCrRr)rSrOrWrTrrrrX7rVc*4s| dur| |_|(dur^|(|_|tuss|!s| rtdto#t|jt}*to+t|jt}+tj|(|*p2|+d},|,j }!ddi} |durM|,j rM|,j rMt |,j |,j }t |,j|,jfdd|,jD}|dur|turmd|_d g}nd|_t|trytd d|vrtd ||_||_|)dur|)|_pt|jd |_|j|g}-d d|jDD]}.|-t|.tr|.n|.|fqdd|jD|_dd|jD|_t|-}/|j t!t"dd|/#D|j$dd|/%D|r|jst&|j||_'|tur||_(d|_)||_*| |_+||_,|dur t|t-rtd||_.nt/|_0|dur6t|t-r3td||_1|durHt|t-rEtd||_2|durYt3|sVtd||_4|durkt|t-rhtd||_5|dur}t3|sytd||_6nt7|_6t8|_9t:|j.|j2t;jt?<|s|r|rtdt@jA|_BtCdtDn3|j9j>}0|r|0$|t@jE|_BtF|j9jGj.}1|0HtItJ|1d|0HtKtJ|1dd|0HtLtM|1d|jr|js|jBt@jEur|j9N}2|2rtOPdjQ||2d n|durtOPd!jQ||d"| |_R| r|!stCd#tD| |_S|!|_ | |_T| |_U||_V||_W||_X||_Y||_Z||_[||_\||_]||_^||__|%|_`|&|_atb|_ctd|_etf|_gth|_id|_jtk|_ltd|_mtnto|_ptqjrtstqjttsi|_utqjrtvtqjttvi|_wtqjrtxtqjttyi|_ztqjrt{tqjtt|i|_}|j~|d$|_t|j|_t|_|jRrd%d&lm}3|3t||_t||jW|jZ|j[|j\|||_j|'durt|_|#dur|#|_|$dur|$|_dSdS)'a ``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. Nzmcontact_points, endpoint_factory, ssl_context, and ssl_options cannot be specified with a cloud configuration)create_pyopenssl_contextcheck_hostnameTcsg|]}|qSr)create_from_sni)rhost_idendpoint_factoryrr sz$Cluster.__init__..Fr6zUcontact_points should not be a string, it should be a sequence (e.g. list) of stringszDcontact_points should not contain None (it can resolve to localhost))rcSsg|] }t|ts|qSr isinstancer&rcprrrr_cSg|] }t|tr|qSrr`rbrrrr_rdcSs$i|]}t|dj|j|jdqSz {ip}:{port})ipr)reprraddressrreprrr sz$Cluster.__init__..cSs"g|] }|durdd|DqS)NcSs"g|] \}}|durt||qSr)r'rrgrrrrr_s"z/Cluster.__init__...r)rxsrrrr_s  cSs(i|]\}}|dur|dd|DqS)NcSsg|] \}}dj||dqSrf)rrmrrrr_sz/Cluster.__init__...r)rkeyr@rrrrlszSload_balancing_policy should not be a class, it should be an instance of that classzQreconnection_policy should not be a class, it should be an instance of that classzRdefault_retry_policy should not be a class, it should be an instance of that classz*conviction_policy_factory must be callablezPaddress_translator should not be a class, it should be an instance of that classz$timestamp_generator must be callable)rrzClusters constructed with execution_profiles should not specify legacy parameters load_balancing_policy or default_retry_policy. Configure this in a profile instead.zVLegacy execution parameters will be removed in 4.0. Consider using execution profiles.)rgf@)rraCluster.__init__ called with contact_points specified, but load-balancing policies are not specified in some ExecutionProfiles. In the next major version, this will raise an error; please specify a load-balancing policy. (contact_points = {cp}, EPs without explicit LBPs = {eps}))rcepszCluster.__init__ called with contact_points specified, but no load_balancing_policy. In the next major version, this will raise an error; please specify a load-balancing policy. (contact_points = {cp}, lbp = {lbp}))rcrDzUsing ssl_options without ssl_context is deprecated and will result in an error in the next major release. Please use ssl_context to prepare for that release.) max_workersr)Metrics)connection_classrvrrrw issubclassrydscloudget_cloud_config ssl_contextusernamepasswordr!r*sni_hostsni_porthost_ids_contact_points_explicitrastrr>contact_pointsrcolumn_encryption_policyr(r^ configurerrendpoints_resolved_endpoint_map_for_insightsrfextendlistrrupdaterr compressionr=_protocol_version_explicitallow_beta_protocol_version no_compactr9typerrrAreconnection_policyrIr?conviction_policy_factoryaddress_translatortimestamp_generatorrer rFrSession_default_timeout _row_factoryrrr0r2rCrDeprecationWarningr3rPr- setdefaultrrrrr rrwarningrmetrics_enabled ssl_optionssockopts cql_versionmax_schema_agreement_waitcontrol_connection_timeoutidle_heartbeat_intervalidle_heartbeat_timeoutschema_event_refresh_windowtopology_event_refresh_windowstatus_event_refresh_windowconnect_timeoutprepare_on_all_hostsreprepare_on_upmonitor_reporting_enabledmonitor_reporting_intervalr _listenersr_listener_lockrzsessionsrCmetadatarOr_prepared_statements_prepared_statement_lockrr  _user_typesrKrDEFAULT_MIN_REQUESTSr _min_requests_per_connectionDEFAULT_MAX_REQUESTS_max_requests_per_connection&DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST'DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST_core_connections_per_host&DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST'DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST_max_connections_per_host_create_thread_pool_executorr _Scheduler schedulerr_lockcassandra.metricsrrweakrefproxymetricsControlConnectionuuiduuid4 client_idapplication_nameapplication_version)4rrrrr9rrrIrrrsrrrr=executor_threadsrrrrrrrQrXrrrrexecution_profilesrrrrrwr^rrrrrrvr uses_twisted uses_eventlet cloud_configraw_contact_pointsrcstrs_resolved_maprrDdefault_lbp_profilesrrrr]rrYsz0                                 zCluster.__init__cKst}tjddkrCtjddkrCzddlm}t|j|}Wn |d i|YS|rCz ddlm}|}Wn t yBt dw|d i|S) a Create a ThreadPoolExecutor for the cluster. In most cases, the built-in `concurrent.futures.ThreadPoolExecutor` is used. Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor` to hang indefinitely. In that case, the user needs to have the `futurist` package so we can use the `futurist.GreenThreadPoolExecutor` class instead. :param kwargs: All keyword args are passed to the ThreadPoolExecutor constructor. :return: A ThreadPoolExecutor instance. rrrx)GreenThreadPoolExecutora2Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor` to hang indefinitely. If you want to use the Eventlet reactor, you need to install the `futurist` package to allow the driver to use the GreenThreadPoolExecutor. See https://github.com/eventlet/eventlet/issues/508 for more details.Nr) rr{ version_inforryrtrsfuturistr ImportError)rr tpe_classry is_eventletrrrrrs"     z$Cluster._create_thread_pool_executorcCsX|jdkrtd|j||||j||<t|jD] }||||qt||dS)a 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) rzUser 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.N) r=rrrrruser_type_registeredr,evict_udt_class)rkeyspace user_typeklasssessionrrrregister_user_types 0zCluster.register_user_typec Cst|ts td|jtjkrtd||jjvr td ||j o&|j }|r6t dj t|||d||jj|<|j||jtdd|jD]}|j|qPt}t|jD]}||||qat||\}} | r}tdd S) a} 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 `_ for timeout semantics). z/profile must be an instance of ExecutionProfilezHCannot add execution profiles when legacy parameters are set explicitly.zProfile {} already existsaTried to add an ExecutionProfile with name {name}. {self} was explicitly configured with contact_points, but {ep} was not explicitly configured with a load_balancing_policy. In the next major version, trying to add an ExecutionProfile without an explicitly configured LBP to a cluster with explicitly configured contact_points will raise an exception; please specify a load-balancing policy in the ExecutionProfile.)rrrkcSr7r)is_up)hrrr z/Cluster.add_execution_profile..z=Failed to create all new connection pools in the %ss timeout.N)rarr>rCr0r2rrFrrr}rrrrrr#r all_hostsfilterr'rrr_set_default_dbaas_consistencyrupdate_created_pools wait_futuresr) rrrpool_wait_timeoutcontact_points_but_no_lbprfuturesr_not_donerrradd_execution_profiles8      zCluster.add_execution_profilecC |j|Sr)rr host_distancerrrget_min_requests_per_connection z'Cluster.get_min_requests_per_connectioncCsR|jdkr td|dks|dks||j|kr"td|j|f||j|<dS)a. 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}. rz`Cluster.set_min_requests_per_connection() only has an effect when using protocol_version 1 or 2.r~zUmin_requests must be 0-126 and less than the max_requests for this host_distance (%d)N)r=rrrr)rr min_requestsrrrset_min_requests_per_connection  z'Cluster.set_min_requests_per_connectioncCrr)rrrrrget_max_requests_per_connection(rz'Cluster.get_max_requests_per_connectioncCsR|jdkr td|dks|dks||j|kr"td|j|f||j|<dS)a( Sets a threshold for concurrent requests per connection, above which new connections will be created to a host (up to max connections; see :meth:`~Cluster.set_max_connections_per_host`). Pertains to connection pool management in protocol versions {1,2}. rz`Cluster.set_max_requests_per_connection() only has an effect when using protocol_version 1 or 2.rzXmax_requests must be 1-127 and greater than the min_requests for this host_distance (%d)N)r=rrrr)rr max_requestsrrrset_max_requests_per_connection+rz'Cluster.set_max_requests_per_connectioncCr)an Gets the minimum number of connections per Session that will be opened for each host with :class:`~.HostDistance` equal to `host_distance`. The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for :attr:`~HostDistance.REMOTE`. This property is ignored if :attr:`~.Cluster.protocol_version` is 3 or higher. )rrrrrget_core_connections_per_host= z%Cluster.get_core_connections_per_hostcCs>|jdkr td|j|}||j|<||kr|dSdS)a Sets the minimum number of connections per Session that will be opened for each host with :class:`~.HostDistance` equal to `host_distance`. The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for :attr:`~HostDistance.REMOTE`. Protocol version 1 and 2 are limited in the number of concurrent requests they can send per connection. The driver implements connection pooling to support higher levels of concurrency. If :attr:`~.Cluster.protocol_version` is set to 3 or higher, this is not supported (there is always one connection per host, unless the host is remote and :attr:`connect_to_remote_hosts` is :const:`False`) and using this will result in an :exc:`~.UnsupportedOperation`. rz^Cluster.set_core_connections_per_host() only has an effect when using protocol_version 1 or 2.N)r=rr_ensure_core_connections)rrcore_connectionsoldrrrset_core_connections_per_hostIs    z%Cluster.set_core_connections_per_hostcCr)an Gets the maximum number of connections per Session that will be opened for each host with :class:`~.HostDistance` equal to `host_distance`. The default is 8 for :attr:`~HostDistance.LOCAL` and 2 for :attr:`~HostDistance.REMOTE`. This property is ignored if :attr:`~.Cluster.protocol_version` is 3 or higher. )rrrrrget_max_connections_per_hostbrz$Cluster.get_max_connections_per_hostcCs |jdkr td||j|<dS)a= Sets the maximum number of connections per Session that will be opened for each host with :class:`~.HostDistance` equal to `host_distance`. The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for :attr:`~HostDistance.REMOTE`. If :attr:`~.Cluster.protocol_version` is set to 3 or higher, this is not supported (there is always one connection per host, unless the host is remote and :attr:`connect_to_remote_hosts` is :const:`False`) and using this will result in an :exc:`~.UnsupportedOperation`. rz]Cluster.set_max_connections_per_host() only has an effect when using protocol_version 1 or 2.N)r=rr)rrmax_connectionsrrrset_max_connections_per_hostns z$Cluster.set_max_connections_per_hostcOs*|||}|jj||jg|Ri|S)zv Called to create a new connection with proper configuration. Intended for internal use only. )_make_connection_kwargsrsfactoryr)rendpointrrrrrconnection_factorys zCluster.connection_factorycOs0||j|}t|jj|j|jg|Ri|Sr)rrr rsrr)rrrrrrr_make_connection_factorys"z Cluster._make_connection_factorycCs|jr |d||j|d|j|d|j|d|j|d|j|d|j|d|j|d|j |d |j |d |j |d |j |S) N authenticatorrrrrrwrr= user_type_maprr) r<rrirrrrrwrr=rrr)rr kwargs_dictrrrrszCluster._make_connection_kwargscCsT|jr td|ft|}|tjkrtdtjftd|j||||_dS)NzXProtocolError returned from server while using explicitly set client protocol_version %dzECannot downgrade protocol version below minimum supported version: %daDowngrading core protocol version from %d to %d for %s. To avoid this, it is best practice to explicitly set Cluster(protocol_version) to the version supported by your cluster. http://datastax.github.io/python-driver/api/cassandra/cluster.html#cassandra.cluster.Cluster.protocol_version)rrrget_lower_supported MIN_SUPPORTEDrrr=)r host_endpointprevious_version new_versionrrrprotocol_downgrades    zCluster.protocol_downgradec Cs|j|jr td|jstd|j|j|j t ||j D]}|j |dd\}}|r?| |jD]}||q7q#|jt||j|jt||jz&|j|j D]}|j|}|ry|j|tjkryd|_qctdWntytd| w|j!|j"rt#|j"|j$|j%d|_&d |_Wdn1swY|'|}|rt(|j)|*||S) a Creates and returns a new :class:`~.Session` object. If `keyspace` is specified, that keyspace will be the default keyspace for operations on the ``Session``. `wait_for_all_pools` specifies whether this call should wait for all connection pools to be established or attempted. Default is `False`, which means it will return when the first successful connection is established. Remaining pools are added asynchronously. zCluster is already shut downz?Connecting to cluster, contact points: %s; protocol version: %sF)signalNzControl connection createdzwYzdd|DD])}tjd||d||W|j d|_WddS1snwYdSt |st d|||W|j d|_WddS1swYdSt d|| |j D]}||qW|j d|_Wdn1swYn|j d|_Wdw1swYwt|jD]}|qdS)NcSrerrarrrrrrr_>rdz3Cluster._on_up_future_completed..z,Unexpected failure while marking node %s up:rFwYtdddDD] }tjd|ddStsitddS dS)Nz,All futures have completed for added host %scSrerr7r8rrrr_rdz.future_completed..z:Unexpected failure while adding node %s, will not mark up:rr9) rrr:rrrr;r>r _finalize_addrrrJrKrrrrfuture_completeds*   z(Cluster.on_add..future_completedTr2)rrrrFrrKr!rFr+rOr_rrrrrGrr)rr refresh_nodesrrarHrrrr`rr+s<         zCluster.on_addcCs>|r||jD]}||q t|jD]}|qdSr)rrr+rrr)rrrrrrrrr_s   zCluster._finalize_addcCs|jrdStd|||j|t|jD]}||q|jD]}||q%|j || d}|r@| dSdS)NzRemoving host %s) rrrrXrFr,rrrrOrDrE)rrrrreconnection_handlerrrrr,"s        zCluster.on_removecCs ||}|r|||||Sr)signal_connection_failurer*)rrconnection_excr3rZis_downrrrrd3s z!Cluster.signal_connection_failurecCsD|jt||j||\}}|r|rtd||||||fS)a Called when adding initial contact points and when the control connection subsequently discovers a new node. Returns a Host instance, and a flag indicating whether it was new in the metadata. Intended for internal use only. z New Cassandra host %r discovered)radd_or_return_hostrRrrr?r+)rr datacenterrackrrbrrrrrr9s   zCluster.add_hostcCs2|r|j|rtd|||dSdSdS)z Called when the control connection observes that a node has left the ring. Intended for internal use only. zCassandra host %s removedN)r remove_hostrr?r,rrrrrrjHs zCluster.remove_hostcC8|j|j|WddS1swYdS)z Adds a :class:`cassandra.policies.HostStateListener` subclass instance to the list of listeners to be notified when a host is added, removed, marked up, or marked down. N)rrrrrrrrregister_listenerQs"zCluster.register_listenercCrl)z Removes a registered listener. N)rrremovermrrrunregister_listenerZs"zCluster.unregister_listenercCs4|j |jWdS1swYdSr)rrr rrrrr_s$zCluster.listenerscCs0t|jD]}t|jD]}|qqdS)z If any host has fewer than the configured number of core connections open, attempt to open connections until that number is met. N)rr_poolsrensure_core_connections)rrpoolrrrrds  z Cluster._ensure_core_connectionscCsHt||||fr |stdtdd|||fDdkr"tddSdS)NzZkeyspace is required to refresh specific sub-entity {table, usertype, function, aggregate}css|]}|rdVqdS)rNr)rrrrrrrz3Cluster._validate_refresh_schema..rz={table, usertype, function, aggregate} are mutually exclusive)anyrsumrtableusertypefunction aggregaterrr_validate_refresh_schemamsz Cluster._validate_refresh_schemacCs6|rtjS|r tjS|rtjS|rtjS|rtjSdSr)r AGGREGATEFUNCTIONTYPETABLEKEYSPACErwrrr_target_type_from_refresh_argsusz&Cluster._target_type_from_refresh_argscCs*|jj}|r |jnd}|r|j|SdS)z? Returns the control connection host metadata. N)rO _connectionrrr)r connectionrrrrget_control_connection_hostsz#Cluster.get_control_connection_hostcCs|jj|dds tddS)a Synchronously refresh all schema metadata. By default, the timeout for this operation is governed by :attr:`~.Cluster.max_schema_agreement_wait` and :attr:`~.Cluster.control_connection_timeout`. Passing max_schema_agreement_wait here overrides :attr:`~.Cluster.max_schema_agreement_wait`. Setting max_schema_agreement_wait <= 0 will bypass schema agreement and refresh schema immediately. An Exception is raised if schema refresh fails for any reason. T)schema_agreement_waitforcez7Schema metadata was not refreshed. See log for details.N)rOrefresh_schemar)rrrrrrefresh_schema_metadatas zCluster.refresh_schema_metadatacCs"|jjtj||ddstddS)aN Synchronously refresh keyspace metadata. This applies to keyspace-level information such as replication and durability settings. It does not refresh tables, types, etc. contained in the keyspace. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior T) target_typerrrz9Keyspace metadata was not refreshed. See log for details.N)rOrrrr)rrrrrrrefresh_keyspace_metadatas z!Cluster.refresh_keyspace_metadatacC$|jjtj|||ddstddS)z Synchronously refresh table metadata. This applies to a table, and any triggers or indexes attached to the table. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior Trrrxrrz6Table metadata was not refreshed. See log for details.NrOrrrr)rrrxrrrrrefresh_table_metadatas zCluster.refresh_table_metadatacCr)z Synchronously refresh materialized view metadata. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior Trz5View metadata was not refreshed. See log for details.Nr)rrviewrrrr"refresh_materialized_view_metadata z*Cluster.refresh_materialized_view_metadatacCr)z Synchronously refresh user defined type metadata. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior T)rrrrrz:User Type metadata was not refreshed. See log for details.N)rOrrrr)rrrrrrrrefresh_user_type_metadatarz"Cluster.refresh_user_type_metadatacCr)z Synchronously refresh user defined function metadata. ``function`` is a :class:`cassandra.UserFunctionDescriptor`. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior T)rrrzrrz>User Function metadata was not refreshed. See log for details.N)rOrrr~r)rrrzrrrrrefresh_user_function_metadata z&Cluster.refresh_user_function_metadatacCr)a Synchronously refresh user defined aggregate metadata. ``aggregate`` is a :class:`cassandra.UserAggregateDescriptor`. See :meth:`~.Cluster.refresh_schema_metadata` for description of ``max_schema_agreement_wait`` behavior T)rrr{rrz?User Aggregate metadata was not refreshed. See log for details.N)rOrrr}r)rrr{rrrrrefresh_user_aggregate_metadatarz'Cluster.refresh_user_aggregate_metadatacCs|j|s tddS)a Synchronously refresh the node list and token metadata `force_token_rebuild` can be used to rebuild the token map metadata, even if no new nodes are discovered. An Exception is raised if node refresh fails for any reason. z1Node list was not refreshed. See log for details.N)rOrefresh_node_list_and_token_maprrforce_token_rebuildrrrrbs zCluster.refresh_nodescCstdt||_||_dS)a *Deprecated:* set :attr:`~.Cluster.schema_metadata_enabled` :attr:`~.Cluster.token_metadata_enabled` instead Sets a flag to enable (True) or disable (False) all metadata refresh queries. This applies to both schema and node topology. Disabling this is useful to minimize refreshes during multiple changes. Meta refresh must be enabled for the driver to become aware of any cluster topology changes or schema updates. zCluster.set_meta_refresh_enabled is deprecated and will be removed in 4.0. Set Cluster.schema_metadata_enabled and Cluster.token_metadata_enabled instead.N)rrrQrXrTrrrset_meta_refresh_enableds   z Cluster.set_meta_refresh_enabledc sR|D]$}fdd|D}|j|ddd}|D] \}} |s%td|| qqdS)Ncs$g|]}t|jr |jnddqS)Nqueryr)r5 query_stringr)rr% set_keyspacerrr_s   z(Cluster._send_chunks..g@Fr fail_on_errorz?Got unexpected response when preparing statement on host %s: %r)wait_for_responsesrr) clsrrchunksrks_chunkmessages responsessuccessresponserrr _send_chunkss  zCluster._send_chunksc Cs|jr|jsdStd|d}zzt||j}t|j}t |j rFg}t dt |dD]}| |||d||||dq/n8t|ddD]0\}}|durZ||t|}g}t dt |dD] }| |||dqh||||qMtd|WnBty}z td||WYd}~n.d}~wttjfy} z td || WYd} ~ nd} ~ wtytd |YnwW|r|dSdS|r|ww) Nz7Preparing all known prepared statements against host %srrKTcSr7r)r)r%rrrr rz.Cluster._prepare_all_queries..zd?Z=d@dAZ>dBdCZ?dDdEZ@dFdGZAdHdIZBdJdKZCdLdMZDdNdOZEdPdQZFdRdSZGdTdUZHdVdWZIdXdYZJdZd[ZKdS)^ra A collection of connection pools for each host in the cluster. Instances of this class should not be created directly, only using :meth:`.Cluster.connect()`. Queries and statements can be executed through ``Session`` instances using the :meth:`~.Session.execute()` and :meth:`~.Session.execute_async()` methods. Example usage:: >>> session = cluster.connect() >>> session.set_keyspace("mykeyspace") >>> session.execute("SELECT * FROM mycf") NFcCr7)a  The format to return row results in. By default, each returned row will be a named tuple. You can alternatively use any of the following: - :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 )rrrrrrK r:zSession.row_factorycC|d|dS)Nr_validate_set_legacy_config)rrfrrrrZ rVrcCr7)a A default timeout, measured in seconds, for queries executed through :meth:`.execute()` or :meth:`.execute_async()`. This default may be overridden with the `timeout` parameter for either of those methods. Setting this to :const:`None` will cause no timeouts to be set by default. Please see :meth:`.ResponseFuture.result` for details on the scope and effect of this timeout. .. versionadded:: 2.0.0 )rrrrrdefault_timeout` zSession.default_timeoutcCr)Nrr)rrrrrrp rVcCr7)a *Deprecated:* use execution profiles instead The default :class:`~ConsistencyLevel` for operations executed through this session. This default may be overridden by setting the :attr:`~.Statement.consistency_level` on individual statements. .. versionadded:: 1.2.0 .. versionchanged:: 3.0.0 default changed from ONE to LOCAL_ONE )r"rrrrdefault_consistency_levelv rz!Session.default_consistency_levelcCstdt|d|dS)z> *Deprecated:* use execution profiles instead zSetting the consistency level at the session level will be removed in 4.0. Consider using execution profiles and setting the desired consistency level to the EXEC_PROFILE_DEFAULT profile.rN)rrrrclrrrr scCr7)a9 The default :class:`~ConsistencyLevel` for serial phase of conditional updates executed through this session. This default may be overridden by setting the :attr:`~.Statement.serial_consistency_level` on individual statements. Only valid for ``protocol_version >= 2``. )!_default_serial_consistency_levelrrrr default_serial_consistency_level s z(Session.default_serial_consistency_levelcCs*|dur t|s td|d|dS)Nzidefault_serial_consistency_level must be either ConsistencyLevel.SERIAL or ConsistencyLevel.LOCAL_SERIAL.r)rrrrrrrrr s rLiTc Cs||_||_||_t|_i|_|j|_|j|_ g|_ |jj |_ t |_t|_|D]}|j|dd}|r;|j|q*t|jtd}|jrdtdd|jDsdt|jtd}|jrdtdd|jDrPtdd|jDsd}|jrz|d|j7}t|d d |Dt|_||_|jjdurztt |jd t!fd |jji|_"Wnt#yt$%d Ynw|jj&r|j'}|ot(|j)} | rt*|jj+|d|_,n |rt$-dj.|j/|dt$-d.|jj0|jdS)NFr2) return_whencs|]}|VqdSrr:r8rrrr$ rtz#Session.__init__..csrrrr8rrrr' rt Unable to connect to any serversz using keyspace '%s'cSg|]}|jqSr)ri)rrrrrr_+ z$Session.__init__..z-ProtocolHandlerrz2Unable to set column encryption policy for session) interval_secrzsNot starting MonitorReporter thread for Insights; not supported by server version {v} on ControlConnection host {c})vcz3Started Session with client_id {} and session_id {})1rr$rrrrqrF_profile_managerr_metrics_request_init_callbacksr=_protocol_versionr-encoderrrrGrrrrrudonerrr session_id_check_graph_paging_available_graph_paging_availablerrr~r@client_protocol_handlerrrr?rrri dse_versionrhr_monitor_reporterrrrelease_versionr) rrr$rrrrmsgcc_hostvalid_insights_versionrrrr sh           zSession.__init__c Cs|||||||||| S)a Execute the given query and synchronously wait for the response. If an error is encountered while executing the query, an Exception will be raised. `query` may be a query string or an instance of :class:`cassandra.query.Statement`. `parameters` may be a sequence or dict of parameters to bind. If a sequence is used, ``%s`` should be used the placeholder for each argument. If a dict is used, ``%(name)s`` style placeholders must be used. `timeout` should specify a floating-point timeout (in seconds) after which an :exc:`.OperationTimedOut` exception will be raised if the query has not completed. If not set, the timeout defaults to the request_timeout of the selected ``execution_profile``. If set to :const:`None`, there is no timeout. Please see :meth:`.ResponseFuture.result` for details on the scope and effect of this timeout. If `trace` is set to :const:`True`, the query will be sent with tracing enabled. The trace details can be obtained using the returned :class:`.ResultSet` object. `custom_payload` is a :ref:`custom_payload` dict to be passed to the server. If `query` is a Statement with its own custom_payload. The message payload will be a union of the two, with the values specified here taking precedence. `execution_profile` is the execution profile to use for this request. It can be a key to a profile configured via :meth:`Cluster.add_execution_profile` or an instance (from :meth:`Session.execution_profile_clone_update`, for example `paging_state` is an optional paging state, reused from a previous :class:`ResultSet`. `host` is the :class:`cassandra.pool.Host` that should handle the query. If the host specified is down or not yet connected, the query will fail with :class:`NoHostAvailable`. Using this is discouraged except in a few cases, e.g., querying node-local tables and applying schema changes. `execute_as` the user that will be used on the server to execute the request. This is only available on a DSE cluster. ) execute_asyncr:) rr parametersrtracecustom_payloadexecution_profile paging_stater execute_asrrrexecuteJ s+zSession.executec CsR|r|ni}| r| |t<|||||||||} |j| _|| | | S)av Execute the given query and return a :class:`~.ResponseFuture` object which callbacks may be attached to for asynchronous response delivery. You may also call :meth:`~.ResponseFuture.result()` on the :class:`.ResponseFuture` to synchronously block for results at any time. See :meth:`Session.execute` for parameter definitions. Example usage:: >>> session = cluster.connect() >>> future = session.execute_async("SELECT * FROM mycf") >>> def log_results(results): ... for row in results: ... log.info("Results: %s", row) >>> def log_error(exc): >>> log.error("Operation failed: %s", exc) >>> future.add_callbacks(log_results, log_error) Async execution with blocking wait for results:: >>> future = session.execute_async("SELECT * FROM mycf") >>> # do other stuff... >>> try: ... results = future.result() ... except Exception: ... log.exception("Operation failed:") )encoder _create_response_futurer_protocol_handler _on_request send_request) rrrrrrrrrrrrrrrw s %   zSession.execute_asynccCs||||||S)a Executes a Gremlin query string or GraphStatement synchronously, and returns a ResultSet from this execution. `parameters` is dict of named parameters to bind. The values must be JSON-serializable. `execution_profile`: Selects an execution profile for the request. `execute_as` the user that will be used on the server to execute the request. )execute_graph_asyncr:)rrrrrrrrr execute_graph s zSession.execute_graphc Cs&|jjtjur tdt|tst|}||}t |ds"td| |zt |j t r5| |j|_ Wn ty?Ynw||d}|rQ|j||jd}|j}|r^||t<tt|jd|t<|j|d||t|d}||j_|j|_|jjrt|j t!r|"||S|#|S)a  Execute the graph query and return a :class:`ResponseFuture` object which callbacks may be attached to for asynchronous response delivery. You may also call ``ResponseFuture.result()`` to synchronously block for results at any time. zCannot execute graph queries using Cluster legacy parameters. Consider using Execution profiles: https://docs.datastax.com/en/developer/python-driver/latest/execution_profiles/#execution-profilesrzeExecution profile for graph queries must derive from GraphExecutionProfile, and provide graph_optionsN)ri)rrrrr)$rrCr0r2rrarorpexecution_profile_clone_updatehasattr"_resolve_execution_profile_optionsrtrrur>_maybe_set_graph_paging_transform_paramsrget_options_maprr rdintrrtrrr query_paramsrris_analytics_sourcerrP_target_analytics_masterr) rrrrrrgraph_parametersrrrrrr sJ            zSession.execute_graph_asynccCs*|j}|jtur|jrtnd}||_dSr)rrrr)rr graph_pagingrrrr s  zSession._maybe_set_graph_pagingcCsNt|jsdS|jjD]}|jdurdSt|j}|tkr$dSqdS)zZVerify if we can enable graph paging. This executed only once when the session is created.FNT) r has_continuous_paging_next_pagesrrrrrrg_GRAPH_PAGING_MIN_DSE_VERSION)rrversionrrrr s   z%Session._check_graph_paging_availablecCs|jjdur |jdur dS|j}d}|jr0|jd}||jjjvr0|jjj|}|jdkr0d}|r8t j }t }n|j t jkrDt j}t}nt j}t}|jdurQ||_|jdur[||_dSdS)a Determine the GraphSON protocol and row factory for a graph query. This is useful to configure automatically the execution profile when executing a query on a core graph. If `graph_protocol` is not explicitly specified, the following rules apply: - Default to GraphProtocol.GRAPHSON_1_0, or GRAPHSON_2_0 if the `graph_language` is not gremlin-groovy. - If `graph_options.graph_name` is specified and is a Core graph, set GraphSON_3_0. If `row_factory` is not explicitly specified, the following rules apply: - Default to graph_object_row_factory. - If `graph_options.graph_name` is specified and is a Core graph, set graph_graphson3_row_factory. NFutf-8CoreT)rgraph_protocolr graph_namedecoderr keyspaces graph_enginerm GRAPHSON_3_0rrrrkDEFAULT_GRAPH_LANGUAGEDEFAULT_GRAPH_PROTOCOLrj GRAPHSON_2_0rq)rrr is_core_graphr ks_metadatarrrrrr s0       z*Session._resolve_execution_profile_optionscCsxt|ts tdt}|jtjkrt}n|jtjkr.|j |j r&|j dndd}t |}| |}t|dgS)NzHThe parameters must be a dictionary. Unnamed parameters are not allowed.r)rr)rar rrlrrmr rnrrrrrs serializejsondumpsr)rrr serializercontextserialized_parametersrrrr3 s    zSession._transform_paramscCsP||jdddd|jd}t|_||j}||f}|j||||ddS)Nz,CALL DseClientTool.getAnalyticsGraphServer()F)rrrr)rL callback_argserrback errback_args) _start_timerrrrarr_on_analytics_master_result add_callbacks)rrmaster_query_futurecbrrrrrF sz Session._target_analytics_mastercCsz,|d}|dd}|d}|dkr|d|}t|j|}|j|j||_Wnty<t j dddYnw| |j dS)Nrlocation:zFailed querying analytics master (request might not be routed optimally). Make sure the session is connecting to a graph analytics datacenter.Tr) r:rfindrcr_load_balancermake_query_planr query_planrrrrr)rr master_future query_futurerowaddrdelimiter_indextargeted_queryrrrrR s       z#Session._on_analytics_master_resultc Csd} t|tr t|}n t|tr||}|jjtjkrU|t ur%t d|t ur,|j }|j dur4|j n|j} |jdur?|jn|j} |jpH|jj} |j} |jj}d}d}n4||}|t ura|j}|j duri|j n|j } |jdurt|jn|j} |j}|jp|j} |j} |j}|j}|j}|tur|jdkr|j}n|jdkrd}t}|jdkr|jr|j}nd}t !|j}|r|rt"|j#}nd}t|tr|j$}t %|jr|j&nd}|rt'|||j(}t)|| | |||||}nWt|t*r |j+} t,| j-|j.| | |||t/| j0|| j1d }n9t|t2r4|jdkrt3dt %|jr&|j&nd}t4|j5|j6| | ||}nt|t7rEt)|j8| | ||||}||_9|:|j;|:||jj<|_<|j=rj|rj|>|j&pg|j&|nd}t?|||||j@| | | |||||d S) z@ Returns the ResponseFuture before calling send_request() on it Nz?Cannot specify execution_profile while using legacy parameters.rrr) skip_metarresult_metadata_idzBatchStatement execution is only supported with protocol version 2 or higher (supported in Cassandra 2.0 and higher). Consider setting Cluster.protocol_version to 2 to support this operation.) rrrr load_balancer start_timespeculative_execution_plancontinuous_paging_stater)Arar~rXrYbindrrCr0r2rrrrrrrrrrIrr_maybe_get_execution_profilerrr fetch_sizerbrdefault_fetch_sizetimeuse_client_timestamprrrr)rrrrr\rr.rZrr6rrrSresult_metadatar(r[rr; batch_type_statements_and_parametersrortracingupdate_custom_payloadrr is_idempotentnew_planResponseFuturer)rrrrrrrrrrr serial_clrrrspec_exec_policyrr/r* timestamp supports_continuous_paging_stater,rstatement_keyspacerspec_exec_planrrrra s                     ( zSession._create_response_futurec CsP|jjj}z||WSty'dd|D}tdt|d|fw)z Returns the execution profile associated with the provided ``name``. :param name: The name (or key) of the execution profile. cSsg|]}t|qSr)rrjrrrr_ sz1Session.get_execution_profile..z6Invalid execution_profile: %s; valid profiles are: %s.z, )rrFrKeyErrorkeysrrjoin)rrrrprrrget_execution_profile s   zSession.get_execution_profilecCst|tr|S||Sr)rarrD)rrkrrrr. sz$Session._maybe_get_execution_profilecKs0t||}|D] \}}t|||q |S)a Returns a clone of the ``ep`` profile. ``kwargs`` can be specified to update attributes of the returned profile. This is a shallow clone, so any objects referenced by the profile are shared. This means Load Balancing Policy is maintained by inclusion in the active profiles. It also means updating any other rich objects will be seen by the active profile. In cases where this is not desirable, be sure to replace the instance instead of manipulating the shared object. )r r.rsetattr)rrkrcloneattrr@rrrr s z&Session.execution_profile_clone_updatecO|j|||fdS)a Adds a callback with arguments to be called when any request is created. It will be invoked as `fn(response_future, *args, **kwargs)` after each client request is created, and before the request is sent. This can be used to create extensions by adding result callbacks to the response future. `response_future` is the :class:`.ResponseFuture` for the request. Note that the init callback is done on the client thread creating the request, so you may need to consider synchronization if you have multiple threads. Any callbacks added to the response future will be executed on the event loop thread, so the normal advice about minimizing cycles and avoiding blocking apply (see Note in :meth:`.ResponseFuture.add_callbacks`. See `this example `_ in the source tree for an example. N)rrrfnrrrrradd_request_init_listener sz!Session.add_request_init_listenercOrH)zz Removes a callback and arguments from the list. See :meth:`.Session.add_request_init_listener`. N)rrorIrrrremove_request_init_listener sz$Session.remove_request_init_listenercCs,|jD]\}}}||g|Ri|qdSr)r)rresponse_futurerJrrrrrr szSession._on_requestc Cst||d}t||d|jd}z ||}Wn ty(tdw|r-|nd}t |j |j |j |jj|||j|j|j|jj }|j|_|j|j ||jjru|j} z ||j| |W|StyttdY|Sw|S)aE Prepares a query string, returning a :class:`~cassandra.query.PreparedStatement` instance which can be used as follows:: >>> session = cluster.connect("mykeyspace") >>> query = "INSERT INTO users (id, name, age) VALUES (?, ?, ?)" >>> prepared = session.prepare(query) >>> session.execute(prepared, (user.id, user.name, user.age)) Or you may bind values to the prepared statement ahead of time:: >>> prepared = session.prepare(query) >>> bound_stmt = prepared.bind((user.id, user.name, user.age)) >>> session.execute(bound_stmt) Of course, prepared statements may (and should) be reused:: >>> prepared = session.prepare(query) >>> for user in users: ... bound = prepared.bind((user.id, user.name, user.age)) ... session.execute(bound) Alternatively, if :attr:`~.Cluster.protocol_version` is 5 or higher (requires Cassandra 4.0+), the keyspace can be specified as a parameter. This will allow you to avoid specifying the keyspace in the query without specifying a keyspace in :meth:`~.Cluster.connect`. It even will let you prepare and use statements against a keyspace other than the one originally specified on connection: >>> analyticskeyspace_prepared = session.prepare( ... "INSERT INTO user_activity id, last_activity VALUES (?, ?)", ... keyspace="analyticskeyspace") # note the different keyspace **Important**: PreparedStatements should be prepared only once. Preparing the same query more than once will likely affect performance. `custom_payload` is a key value map to be passed along with the prepare message. See :ref:`custom_payload`. rN)rrzError preparing query:z#Error preparing query on all hosts:)r5r:rrr:onerrrrY from_messager bind_metadata pk_indexesrrrcolumn_metadatar(rrrr _current_hostr) rrrrrrrprepared_keyspacerrrrrprepare s4 (    zSession.preparec Csg}t|jD]C}||krL|jrLt|t||dd|j}z||}Wnty4t d|Yq w|durEt d||j |q |||fq |D]\}}z|WqOtyht d|YqOwdS)z| Prepare the given query on all hosts, excluding ``excluded_host``. Intended for internal use only. rNz"Error preparing query for host %s:z'Failed to prepare query for host %s: %r)rrqrBrr:r5r_queryrrrr_errorsrWrr:)rr excluded_hostrrrr request_idrrrrN s4      zSession.prepare_on_all_hostscCs|j|jr WddSd|_Wdn1swY|jD]}|q%t|j|jr9|jt|j D]}| q@dS)z Close all connections. ``Session`` instances should not be used for any purpose after being shutdown. NT) rrrrErrr&rrqrr)rrrsrrrrp s    zSession.shutdowncCr'rrrrrrr( r)zSession.__enter__cGr*rr+r,rrrr- rzSession.__exit__cCsz|WdSYdSrr+rrrr__del__ szSession.__del__cs6jtjkr dSfdd}|S)z( For internal use only. Nc szjdkr t}nt}WnGty7}ztt|d}j|WYd}~dSd}~wty[}zt j d|djj|ddWYd}~dSd}~wwj }j b|jjkrj tgfdd }|j|jjrrt d j|j  WddSj |jjksl|j <Wdn1swYt d |r|dS) Nr)rFz1Failed to create connection pool for new host %s:rT)rZcs|dSr)rr)rsrerrors_returnedset_keyspace_eventrrrL s  zJSession.add_or_renew_pool..run_add_or_renew_pool..callbackzJFailed setting keyspace for pool after keyspace changed during connect: %sz!Added pool for host %s to session)rrVrUrr"r~rrdrrrrqrWr _keyspacerreleaser_set_keyspace_for_all_connsrris_setr*racquirer)new_poolauth_excconn_excpreviousrLrrr3rr[rrun_add_or_renew_pool sX           z8Session.add_or_renew_pool..run_add_or_renew_pool)rrrKr!r)rrr3rhrrgrrG s   0zSession.add_or_renew_poolcCs.|j|d}|rtd|||jSdS)NzRemoved connection pool for %r)rqpoprrrr)rrrsrrrr4 s   zSession.remove_poolcCst}|jjD]@}|j|}|j|}d}|r|jr/|t j kr.|j dvr.| |d}n||j krB|t j kr?||}n||_ |rI||q |S)a When the set of live nodes change, the loadbalancer will change its mind on host distances. It might change it on the node that came/left but also on other nodes (for instance, if a node dies, another previously ignored node may be now considered). This method ensures that all hosts for which a pool should exist have one, and hosts that shouldn't don't. For internal use only. N)TNF)rrrrrrrqrWrrKr!rrGrr4r)rrrrrsrrrrr s"         zSession.update_created_poolscs(|}|r|fdddSdS)z{ Called by the parent Cluster instance when a node is marked down. Only intended for internal use. csSr)rrrrrr sz!Session.on_down..N)r4r)rrrrrrr* s zSession.on_downcCs||dS Internal N)r*rkrrrr, szSession.on_removecCs|dt|fdS)z Set the default keyspace for all queries made through this Session. This operation blocks until complete. zUSE %sN)rrD)rrrrrr szSession.set_keyspacecs|j||_t|jWdn1swYis'dSfdd}t|jD]}|||q6dS)a Asynchronously sets the keyspace on all pools. When all pools have set all of their connections, `callback` will be called with a dictionary of all errors that occurred, keyed by the `Host` that they occurred against. Ncs,||r ||j<s|dSdSr)ror)rs host_errorsrLrremaining_callbacksrrpool_finished_setting_keyspace s   zKSession._set_keyspace_for_all_pools..pool_finished_setting_keyspace)rrrrqrrr`)rrrLrorsrrmr_set_keyspace_for_all_pools sz#Session._set_keyspace_for_all_poolscsz jjj|}Wntytd|fwz|j|}Wnty.td||fw|jfdd}|jj|<dS)z Called by the parent Cluster instance when the user registers a new mapping from a user-defined type to a class. Intended for internal use only. zCKeyspace %s does not exist or has not been discovered by the driverz*User type %s does not exist in keyspace %scsddfddDS)Nz{ %s }z , c 3s,|]}d|jt|dfVqdS)z%s : %sN)rcql_encode_all_typesgetattr)r field_name)rrrrr> sz?Session.user_type_registered..encode..)rCr field_namesrrtrr= s z,Session.user_type_registered..encodeN) rrrrAUserTypeDoesNotExist user_typesrvrmapping)rrrrks_meta type_metarrrurr) s"   zSession.user_type_registeredcOs&|js|jjj|g|Ri|SdSrj)rrrrrIrrrrE szSession.submitcCstddt|jDS)Ncss |] \}}||fVqdSr) get_state)rrrsrrrrK z)Session.get_pool_state..)r rrqrrrrrrVJ szSession.get_pool_statecC |jSr)rqrrrrrr#M rzSession.get_poolscCs:|jjtjkrtd|ft|d||tj|j_dS)NzXCannot set Session.%s while using Configuration Profiles. Set this in a profile instead.r)rrCr0r3rrEr2)r attr_namer@rrrrP sz#Session._validate_set_legacy_configrr)Lrrrrrr$rrrrrr_rr.rrrrrrr"rrrmax_trace_waitr0r2rrr@rrrqrrrrrrrrrrrrrrrrrrrrDr.rrKrLrrUrrr(r-rZrGr4rr*r,rrprrrVr#rrrrrr1 s               < - 18.  k  B" :!  rc@eZdZdZdS)rwzj An attempt was made to use a user-defined type that does not exist. .. versionadded:: 2.1.0 NrrrrrrrrrwW srwc@s0eZdZdZddZddZddZdd Zd S) _ControlReconnectionHandler Internal cOs(tj|g|Ri|t||_dSr)rSrrrrO)rrOrrrrrre sz$_ControlReconnectionHandler.__init__cCr~r)rO_reconnect_internalrrrr try_reconnecti rz)_ControlReconnectionHandler.try_reconnectcCs|j|dSr)rO_set_new_connectionrrrrron_reconnectionl sz+_ControlReconnectionHandler.on_reconnectioncCst|trdStd|dS)NFz0Error trying to reconnect control connection: %rT)rarrr)rr next_delayrrr on_exceptiono s  z(_ControlReconnectionHandler.on_exceptionN)rrrrrrrrrrrrr` s  rcOs*|}|dur dSt|||i|dS)zZ A callback handler for the ControlConnection that tolerates weak references. Nrr) obj_weakref method_namerrobjrrr_watch_callbackx srcCs$z|WdStyYdSw)z Called when the ControlConnection object is about to be finalized. This clears watchers on the underlying Connection object. N)control_conn_disposedReferenceError)connexpiring_weakrefrrr_clear_watcher s  rc@seZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z ed ZGd ddeZdZdZdZdZdZdZdZdZdZeZ  dPddZddZddZddZ ddZ!ddZ"ddZ#d d!Z$d"d#Z%d$d%Z&dQd&d'Z'dRd(d)Z(dQd*d+Z)  dSd,d-Z*e+d.d/Z,d0d1Z-d2d3Z.d4d5Z/d6d7Z0d8d9Z1d:d;Z2dTdd?Z4dUd@dAZ5dBdCZ6dDdEZ7dFdGZ8dVdHdIZ9dJdKZ:dLdMZ;dNdOZtt|o|do|do|dod|vp|dS)Nr\rrir)rSrFrrW)r#rrrr?sz ControlConnection._is_valid_peercCsD|j|kr |j|kr dS|jj|||||jj|dS)NFT)rhrirrFr*set_location_infor')rrrhrirrrrEs  z'ControlConnection._update_location_infocCsR|j|d}|j}||kr|d}||}n t|}||}||j|<|S)Nr{Gz?)rrW_timer1r)r event_type delay_window next_timenow this_timedelayrrr_delay_for_event_typeQs    z'ControlConnection._delay_for_event_typecCs|r|js |dSdS)z Used to mitigate refreshes for nodes that are already known. Some versions of the server send superfluous NEW_NODE messages in addition to UP events. N)rrrkrrr_refresh_nodes_if_not_up_s  z*ControlConnection._refresh_nodes_if_not_upcCs|d}|d\}}|jj||}|dks|dkr4|jdkr2|d|j}|jj||j|dSdS|dkrE|jjd|jj|dSdS)N change_typeriNEW_NODE MOVED_NODErtopology_change REMOVED_NODE) rrrrrrschedule_uniquerrjreventrr$rrrrrrrgs  z)ControlConnection._handle_topology_changecCs|d}|d\}}|jj||}|dkr9|d|j}|dur,|jj||jdS|jj||jj|dS|dkrK|durM|jj |dddSdSdS)NrriUP status_changeDOWNFr2) rrrrrrrrr'r*rrrrrrs z'ControlConnection._handle_status_changecCs:|jdkrdS|d|j}|jjj||jfi|dS)Nr schema_change)rrrrrr)rrrrrrrs z'ControlConnection._handle_schema_changec Cs |dur|n|jj}|dkrdS|j|jr WddS|s%|j}|rItd|d}|d}||||j}|durI WddStd|j }d} t j } d}| |jj|} | |krt| | d} t|j| d} zt|j|| }|j| | |d\}}Wn7ty}ztd||j |} WYd}~qbd}~wty|jrtd YWddSw||||j}|dur WddStd |j d |j |} | |ksftd |j| Wdd S1swYdS)NrTzM[control connection] Attempting to use preloaded results for schema agreementrz1[control connection] Waiting for schema agreementrrzU[control connection] Timed out waiting for response during schema agreement check: %szC[control connection] Aborting wait for schema match due to shutdownz5[control connection] Schemas mismatched, trying againg?z.Node %s is reporting a schema disagreement: %sF)rrrrrrr_get_schema_mismatchesrrr1rrrrrr._SELECT_SCHEMA_LOCALminrrrr#sleepr)rrrr total_timeoutrrschema_mismatchesrPelapsedrselect_peers_queryrrrrrrrst       ( . $z+ControlConnection.wait_for_schema_agreementc Cst|j|j}tt}|jr&t|j|jd}|dr&||d||D]&}|d}|s2q(|jj |}|jj |} | rN| j durN|||q(t |dkr\tddStdd|DS)Nrschema_versionFrz"[control connection] Schemas matchcss |] \}}|t|fVqdSrr)rrnodesrrrrr}z;ControlConnection._get_schema_mismatches..)r`rrrrrWrrr^rrrrrrrr r) rrr local_addressversionsrr# schema_verrrrrrr s&    z(ControlConnection._get_schema_mismatchescCs||jj|jjfvrtd||jr)||jjkr$|jr|jn|j}|S|j}|S||jjkr7|jr7|j }|S||jjkr@|j n|j }|j j |jj}|j j |jj}|o]t||jk}|rh|jdd}|S|rr|jdd}|S|j }|S)a. Determine the peers query to use. :param peers_query_type: Should be one of PeersQueryType enum. If _uses_peers_v2 is True, return the proper peers_v2 query (no templating). Else, apply the logic below to choose the peers v1 address column name: Given a connection: - find the server product version running on the connection's host, - use that to choose the column name for the transport address (see APOLLO-1130), and - use that column name in the provided peers query template. zInvalid peers query type: %snative_transport_address) nt_col_name rpc_address)rrrrrrW_SELECT_PEERS_V2_SELECT_PEERS_NO_TOKENS_V2_SELECT_SCHEMA_PEERS_V2 _SELECT_PEERS_SELECT_SCHEMA_PEERS_TEMPLATE _SELECT_PEERS_NO_TOKENS_TEMPLATErrrrrrrg#_MINIMUM_NATIVE_ADDRESS_DSE_VERSIONr)rpeers_query_typerrquery_templatehost_release_versionhost_dse_versionuses_native_address_queryrrrrs6     z"ControlConnection._get_peers_querycCs|j:|jr WddS|jr6|jjr6|jj|jj}|r6|jj||jj dd WddSWdn1s@wY| dSr1) rrr is_defunctrrrrrd last_errorrrkrrrrs  zControlConnection._signal_errorcCsdSrrrkrrrr'%r)zControlConnection.on_upcCsD|j}|r|j|jkr|jdur td||dSdSdSdS)Nz[[control connection] Control connection host (%s) is considered down, starting reconnection)rrrrrr)rrrrrrr*(s  zControlConnection.on_downcCs|r |jdddSdS)NTr)r)rrrbrrrr+2szControlConnection.on_addcCs>|j}|r|j|jkrtd||dS|jdddS)NzP[control connection] Control connection host (%s) is being removed. ReconnectingTr)rrrrrr)rrrrrrr,6s   zControlConnection.on_removecCst|dd}|r |gSgS)Nrr)rrrrrget_connections?s z!ControlConnection.get_connectionscCs*||jur|js |jr|dSdSdSr)rr) is_closedrrrrrreturn_connectionCs z#ControlConnection.return_connection)TTr)NNFr)NNNrr)=rrrrr r"rrrr!r rrrrgr#rrrrrrrrrPrWrr1rrrrrrrrrrrrrrrrrrrrrrrrr rrr'r*r+r,r+r-rrrrr sz   D         @ -  rcCs2z |js|Wn tyYnw|dSr)rrrrC)rthreadrrr_stop_schedulerHs  r/c@sTeZdZdZdZdZdZddZddZddZ d d Z d d Z d dZ ddZ dS)rNFcCs@t|_t|_t|_||_tj |ddd|_ | dS)NzTask Scheduler)rT) queue PriorityQueue_queuer_scheduled_tasksr_count _executorrrdaemonrP)rrrrrrYs  z_Scheduler.__init__cCs@ztdWn tyYnwd|_|jd|dS)NzShutting down Cluster SchedulerT)rrN)rrrrr2 put_nowaitrCrrrrrcs   z_Scheduler.shutdowncOs||||t|fdSr) _insert_taskrr)rrrJrrrrrrQmsz_Scheduler.schedulecOs<||t|f}||jvr|||dStd|dS)Nz7Ignoring schedule_unique for already-scheduled task: %r)rrr3r8rr)rrrJrrtaskrrrrps z_Scheduler.schedule_uniquecCsJ|jst|}|j||j|t|j|fdSt d|dS)Nz*Ignoring scheduled task after shutdown: %r) rr1r3rr2r7nextr4rr)rrr9run_atrrrr8ws   z_Scheduler._insert_taskcCs |jrdSzO |jjddd\}}}|jr |rtdWdS|tkrI|j||\}}}t|}|j j |g|Ri|}| |j n |j |||fnqWn tjy_Ynwtdq)NT)blockrz6Not executing scheduled task due to Scheduler shutdowng?)rr2rWrrr1r3rr r5rr_log_if_failedr7r0Emptyr)rr;rr9rJrrrrrrruns2     z_Scheduler.runcCs"|}|rtjd|ddSdS)NzBAn internally scheduled tasked failed with an unhandled exception:r)rrr)rrrrrrr=s z_Scheduler._log_if_failed)rrrr2r3r5rrrrQrr8r?r=rrrrrRs   rc Kszd.d/Z?d0d1Z@d2d3ZAd4d5ZBd6d7ZCdeDjEfd8d9ZFdeDjEfd:d;ZGdd?ZId@dAZJ  dLdCdDZKdEdFZLdGdHZMeMZNdS)Mr:a An asynchronous response delivery mechanism that is returned from calls to :meth:`.Session.execute_async()`. There are two ways for results to be delivered: - Synchronously, by calling :meth:`.result()` - Asynchronously, by attaching callback and errback functions via :meth:`.add_callback()`, :meth:`.add_errback()`, and :meth:`.add_callbacks()`. NTrFcCs||_|p|j|_| p|jj|_||_||_||_||_||_ ||_ t |_ | p+t |_| |_| p4|j|_|t|_i|_g|_g|_g|_|| |_dSr)rrrrGrrrr _retry_policyrrr_callback_lockr1 _start_time_host_spec_execution_plan_make_query_planr_eventrW _callbacks _errbacksattempted_hostsr_continuous_paging_state)rrrrrrrrrr)r*r+r,rrrrrs*   zResponseFuture.__init__cCs"|jdurdS|j|jtSr)rrFr1rrrr_time_remainings zResponseFuture._time_remainingcCs||jdur:|j|j}|dkr'|jdus|j|kr'|jjj||j |_dS|jdur<|jjj|j|j |_dSdSdS)Nr) _timerrHnext_executionrSrOrrrs create_timer_on_speculative_execute _on_timeout)r spec_delayrrrrs  zResponseFuture._start_timercCs|jr |jdSdSr)rPrErrrr _cancel_timer szResponseFuture._cancel_timercCst|jdur|dkr|jjjdt|j|dd|_dS|jdurz |jj |j Wnt yDd}|di}| t ||jYdSw|jj|j}|r|js|jj|jj|j t|jj|jjkrmd|j_Wdn1swwY|j|jdd |j}|s|jr|jrt|jjnd }|di}n|jjjj}|rt|jnd }|d i}| t ||jdS) a< Called when the request associated with this ResponseFuture times out. This function may reschedule itself. The ``_attempts`` parameter tracks the number of times this has happened. This parameter should only be set in those cases, where ``_on_timeout`` reschedules itself. Nrrr) _attemptszConnection defunct by heartbeatzt|jj|j|||}|j|||d}|dur|dSdS)N)r)r rr_execute_after_preparerVr)rprepare_messagerrrsrrYrrr _reprepares  zResponseFuture._repreparec CsFz||_|r ||t|dd}|r$|jsg|_|jt||jt|dd|_t|dd|_t |t r|j t krRt|dd}|rO| |j|jWdSWdS|j tkrnd|_|jjt|jjj||fi|jWdS|j tkr|j|_|j|_|j|_t|jddr|||WdS| |!|j|j"WdS|j t#kr| dWdS| |WdSt |t$r|j%}t |t&r|j'dur|j'(|j(|j)fd|j*i|j+}n-t |t,r|j'dur|j'-|j-|j)fd|j*i|j+}nt |t.r|j'dur |j'/|j/|j)fd|j*i|j+}nt |t0t1t2t3frIt45d||j6|j'dur6|j'7t|jd d} |j8|j)| ||j*d }nt |t9r|j:rg|j:j;} | |j+ksfJd |j+| fn|j+} z |jjj<| } Wn*t=y|j:st4>d | ?d |@|YWdS|j:} | |jjj<| <Ynw|jAjB} | jB} tCD|jjjEs| r| | kr|@tFd| | fWdSt4Gd|| jHtCD|jjjEr| jBnd} tI| jH| d}|j|jJ||||WdStK|dr|@|LWdS|@|WdS|M|||WdSt |tNrK|j'dur"|j'Ot |tPs.|jAQ|t|jd d} |j%j8|j)| ||j*d }|M|||WdSt |tRritK|dra|@|LWdS|@|WdSd|f}tN||}|S|jAQ||@|WdStRy}zt4Td|@|WYd}~dSd}~ww)Ntrace_idrprrFr retry_numzHost %s error: %s.r)r;r|zFGot different query ID in server response (%s) than we had before (%s)z2Tried to execute unknown prepared statement: id=%shexzdThe Session's current keyspace (%s) does not match the keyspace the statement was prepared with (%s)z@Re-preparing unrecognized prepared statement against host %s: %sr to_exceptionzGot unexpected message: %rz=Unexpected exception while handling result in ResponseFuture:)Ucoordinator_hostr-rr _query_tracesrr]rrorrrar/kindr=rp new_keyspace_set_keyspace_completedr?r@rrCrrOschema_change_eventr>rrlr _col_names column_types _col_typesr(_handle_continuous_paging_first_responserArrrAr0rDr1ron_read_timeoutr_query_retriesr?r2on_write_timeoutr3on_unavailabler4r8r9r:rrsummaryon_other_erroron_request_errorr7rrrrAr;rr\rrrrr=rrrr5rzrr~_handle_retry_decisionr"rjr#defunctrrVr)rrrrsrr{rrretryrrrcurrent_keyspacerTryrrrrrres:                                        zResponseFuture._set_resultcCs<||j|jj|j|j|_|j|||j dSr) new_continuous_paging_session stream_idrrhrrN_continuous_paging_session on_messagerAr@)rrrrrrrs z7ResponseFuture._handle_continuous_paging_first_responsecCs*|s |ddS|td|fdS)Nz'Failed to set keyspace on all hosts: %s)rAr\r")rrrrrrs  z&ResponseFuture._set_keyspace_completedcCs>|r|||jr dSt|tra|jtkrT|jrC|jj|jkr3|t dj t |jjt |jd|j |j_ |j}|durC||j_||}|durR|dSdS|td||fdSt|tr{t|drt||dS||dSt|trtd||||j|<|dS|td||fdS)z Handle the response to our attempt to prepare a statement. If it succeeded, run the original query again against the same host. NzID mismatch while trying to reprepare (expected {expected}, got {got}). This prepared statement won't work anymore. This usually happens when you run a 'USE...' query after the statement was prepared.)expectedgotz?Got unexpected response when preparing statement on host %s: %sr~z8Connection error when preparing statement on host %s: %szDGot unexpected response type when preparing statement on host %s: %s)r-rvrar/rr<rrr\rrrrRr3r(rVrr"r0rr~rrrW)rrrrsrnew_metadata_idrYrrrrxsT            z%ResponseFuture._execute_after_preparec||jdur|jjt|j|j|_tfdd|j D}Wdn1s2wY|j |D]}|q>dS)Nc3.|]\}}}t|g|Ri|VqdSrr rrJrrrrrr  z3ResponseFuture._set_final_result..) rVr request_timeraddValuer1rFrErurrKrJrrrto_callcallback_partialrrrrAs    z ResponseFuture._set_final_resultcr)Nc3rrrrrrrrrz6ResponseFuture._set_final_exception..) rVrrrr1rFrErvrrLrJrrrrrr\s    z#ResponseFuture._set_final_exceptioncCsdd}|\}}|tjtjfvr$|jd7_|tjk}||||n|tjur1|||n|jdur;|j| d|||j |<dS)NcSst|dr |S|S)Nr~)rr~rrrrexception_from_responses zFResponseFuture._handle_retry_decision..exception_from_responser) rLRETRYRETRY_NEXT_HOSTr_retryRETHROWr\r on_ignorerArW)rretry_decisionrrr retry_type consistencyreuserrrrs     z%ResponseFuture._handle_retry_decisioncCsD|jrdS|jdur|j|dur||j_|j|j||dSr)rvron_retryrrrr _retry_task)rreuse_connectionrrrrrr%s  zResponseFuture._retrycCs,|jrdS|r||durdS|dSr)rvrVr)rrrrrrr3s  zResponseFuture._retry_taskcCs&|j|jturt||jS|j)a Return the final result or raise an Exception if errors were encountered. If the final result or error has not been set yet, this method will block until it is set, or the timeout set for the request expires. Timeout is specified in the Session request execution functions. If the timeout is exceeded, an :exc:`cassandra.OperationTimedOut` will be raised. This is a client-side timeout. For more information about server-side coordinator timeouts, see :class:`.policies.RetryPolicy`. Example usage:: >>> future = session.execute_async("SELECT * FROM mycf") >>> # do other stuff... >>> try: ... rows = future.result() ... for row in rows: ... ... # process results ... except Exception: ... log.exception("Operation failed:") )rJrrur ResultSetrvrrrrr:?s   zResponseFuture.resultcCsdd|jDS)zt Returns the trace session ids for this future, if tracing was enabled (does not fetch trace data). cSrr)r{)rrrrrr_brz6ResponseFuture.get_query_trace_ids..)rrrrrget_query_trace_ids^sz"ResponseFuture.get_query_trace_idscCs>|jtur|jdurtd|jr|t|jd||SdS)aO Fetches and returns the query trace of the last response, or `None` if tracing was not enabled. Note that this may raise an exception if there are problems retrieving the trace details from Cassandra. If the trace is not available after `max_wait`, :exc:`cassandra.query.TraceUnavailable` will be raised. If the ResponseFuture is not done (async execution) and you try to retrieve the trace, :exc:`cassandra.query.TraceUnavailable` will be raised. `query_cl` is the consistency level used to poll the trace tables. NzDTrace information was not available. The ResponseFuture is not done.r)rurrvr^r_get_query_tracer)rmax_waitquery_clrrrget_query_tracedszResponseFuture.get_query_tracecs*jrfddttjDSgS)z Fetches and returns the query traces for all query pages, if tracing was enabled. See note in :meth:`~.get_query_trace` regarding possible exceptions. csg|] }|qSr)r)rr max_wait_perrrrrr_rdz7ResponseFuture.get_all_query_traces..)rrr)rrrrrrget_all_query_tracesys z#ResponseFuture.get_all_query_tracescCs"|j|}|js|j||d|S)N)rr)reventsr#)rrrrrrrrrs zResponseFuture._get_query_tracecOsjd}|j|j|||f|jturd}Wdn1s wY|r3||jg|Ri||S)ay Attaches a callback function to be called when the final results arrive. By default, `fn` will be called with the results as the first and only argument. If `*args` or `**kwargs` are supplied, they will be passed through as additional positional or keyword arguments to `fn`. If an error is hit while executing the operation, a callback attached here will not be called. Use :meth:`.add_errback()` or :meth:`add_callbacks()` if you wish to handle that case. If the final result has already been seen when this method is called, the callback will be called immediately (before this method returns). Note: in the case that the result is not available when the callback is added, the callback is executed by IO event thread. This means that the callback should not block or attempt further synchronous requests, because no further IO will be processed until the callback returns. **Important**: if the callback you attach results in an exception being raised, **the exception will be ignored**, so please ensure your callback handles all error cases that you care about. Usage example:: >>> session = cluster.connect("mykeyspace") >>> def handle_results(rows, start_time, should_log=False): ... if should_log: ... log.info("Total time: %f", time.time() - start_time) ... ... >>> future = session.execute_async("SELECT * FROM users") >>> future.add_callback(handle_results, time.time(), should_log=True) FTN)rErKrrurrrJrrrun_nowrrr add_callbacks% zResponseFuture.add_callbackcOsfd}|j|j|||f|jrd}Wdn1swY|r1||jg|Ri||S)z Like :meth:`.add_callback()`, but handles error cases. An Exception instance will be passed as the first positional argument to `fn`. FTN)rErLrrvrrrr add_errbackszResponseFuture.add_errbackrcCs<|j|g|Ri|p i|j|g|Ri|pidS)a A convenient combination of :meth:`.add_callback()` and :meth:`.add_errback()`. Example usage:: >>> session = cluster.connect() >>> query = "SELECT * FROM mycf" >>> future = session.execute_async(query) >>> def log_results(results, level='debug'): ... for row in results: ... log.log(level, "Result: %s", row) >>> def log_error(exc, query): ... log.error("Query '%s' failed: %s", query, exc) >>> future.add_callbacks( ... callback=log_results, callback_kwargs={'level': 'info'}, ... errback=log_error, errback_args=(query,)) N)rr)rrLrrcallback_kwargsrerrback_kwargsrrrrs zResponseFuture.add_callbackscCs8|jg|_g|_WddS1swYdSr)rErKrLrrrrclear_callbackss"zResponseFuture.clear_callbackscCs.|jturdn|j}d|j|j||j|jfS)Nz(no result yet)zU)rurrr[rvrrr:rrr__str__szResponseFuture.__str__)rrr)rNrN)Orrrrrr@rirrMrrrrrDrr[rrurrrvrrKrLrSrrrFrrlrrrorPr@rrNrH_continuous_paging_optionsrrG_warned_timeoutrLrr.rOrrVrTrSrIrrVrmrprrwrzrerrrxrAr\rrrr:rrrrrrrrrrr__repr__rrrrr:s     7 .   6  1 r:c@r)rsz Raised when :meth:`.ResponseFuture.start_fetching_next_page()` is called and there are no more pages. You can check :attr:`.ResponseFuture.has_more_pages` before calling to avoid this. .. versionadded:: 2.0.0 Nrrrrrrssrsc@seZdZdZddZeddZeddZdd Zd d Z d d Z ddZ e Z ddZ ddZddZddZddZddZddZeZd)dd Zd)d!d"Zd#d$Zed%d&Zed'd(ZdS)*ra An iterator over the rows from a query result. Also supplies basic equality and indexing methods for backward-compatability. These methods materialize the entire result set (loading all pages), and should only be used if the total result size is understood. Warnings are emitted when paged results are materialized in this fashion. You can treat this as a normal iterator over rows:: >>> from cassandra.query import SimpleStatement >>> statement = SimpleStatement("SELECT * FROM users", fetch_size=10) >>> for user_row in session.execute(statement): ... process_user(user_row) Whenever there are no more rows in the current page, the next page will be fetched transparently. However, note that it *is* possible for an :class:`Exception` to be raised while fetching the next page, just like you might see on a normal call to ``session.execute()``. cCs0||_|j|_|j|_||d|_d|_dSr)rMrrrr_set_current_rows _page_iter _list_mode)rrMinitial_responserrrrs   zResultSet.__init__cCrN)zQ True if the last response indicated more pages; False otherwise )rMrmrrrrrmszResultSet.has_more_pagescCs |jpgS)zx The list of current page rows. May be empty if the result was empty, or this is the last page. ) _current_rowsrrrr current_rows#s zResultSet.current_rowscCst|S)z Returns all the remaining rows as a list. This is basically a convenient shortcut to `list(result_set)`. This function is not recommended for queries that return a large number of elements. rrrrrr>+sz ResultSet.allcCsBd}|jrz|jd}W|Stytt|j}Y|Sw|S)a' Return a single row of the results or None if empty. This is basically a shortcut to `result_set.current_rows[0]` and should only be used when you know a query returns a single row. Consider using an iterator if the ResultSet contains more than one row. Nr)rr>r:ra)rr#rrrrN4s  z ResultSet.onecCs |jrt|jSt|j|_|Sr)rrarrrrrr__iter__Ds  zResultSet.__iter__cCsdzt|jWSty|jjs|jsg|_Ynw|jjs-|t |j|_|St|jSr) r:r StopIterationrMrmrrrfetch_next_pagerarrrrr:Js    zResultSet.nextcCs2|jjr|j|j}|j|_dSg|_dS)a Manually, synchronously fetch the next page. Supplied for manually retrieving pages and inspecting :meth:`~.current_page`. It is not necessary to call this when iterating through results; paging happens implicitly in iteration. N)rMrmrwr:rrrrrras     zResultSet.fetch_next_pagecCs^t|tr|r |g|_dSg|_dSz t|||_WdSty.|r(|gng|_YdSwr)rarrrar>rrrrrns    zResultSet._set_current_rowscCst||_d|_dSr)rrrrrrr _fetch_allx  zResultSet._fetch_allcCsB|jrdS|jrtd||jjrtd||d|_dS)Nz.Cannot use %s when results have been iterated.zFUsing %s on paged results causes entire result set to be materialized.T)rr RuntimeErrorrMrmrrr)roperatorrrr_enter_list_mode|s   zResultSet._enter_list_modecCs|d|j|kS)Nzequality operator)rr)rotherrrr__eq__rzResultSet.__eq__cCs&|dkr tdt|d|j|S)NrzfResultSet indexing support will be removed in 4.0. Consider using ResultSet.one() to get a single row.zindex operator)rrrr)rrrrr __getitem__s   zResultSet.__getitem__cCs t|jSr)rSrrrrr __nonzero__rzResultSet.__nonzero__NcC |j|S)z Gets the last query trace from the associated future. See :meth:`.ResponseFuture.get_query_trace` for details. )rMr)r max_wait_secrrrr zResultSet.get_query_tracecCr)z Gets all query traces from the associated future. See :meth:`.ResponseFuture.get_all_query_traces` for details. )rMr)rmax_wait_sec_perrrrrrzResultSet.get_all_query_tracescCs*z |jjWdStytdw)NzkAttempted to cancel paging with no active session. This is only for requests with ContinuousdPagingOptions.)rMrrErrrrrrcancel_continuous_pagings  z"ResultSet.cancel_continuous_pagingcCs|jjtttfvrtd|jjft|jjt}|r)|j r%|j ddkr)td|s;t |j dkr;tdt |j |j d}t|t rI|dS|dS)aq For LWT results, returns whether the transaction was applied. Result is indeterminate if called on a result that was not an LWT request or on a :class:`.query.BatchStatement` containing LWT. In the latter case either all the batch succeeds or fails. Only valid when one of the of the internal row factories is in use. z/Cannot determine LWT result with row factory %srz [applied]z)No LWT were present in the BatchStatementrz4LWT result should have exactly one row. This has %d.) rMrr_r`rarrarr[rrrr)ris_batch_statementr#rrr was_applieds   zResultSet.was_appliedcCrN)a Server paging state of the query. Can be `None` if the query was not paged. The driver treats paging state as opaque, but it may contain primary key data, so applications may want to avoid sending this to untrusted parties. )rMrlrrrrrszResultSet.paging_stater)rrrrrr.rmrr>rNrr:__next__rrrrrrr__bool__rrrrrrrrrrs6         r)r __future__ratexitbinasciir collectionsrcollections.abcrconcurrent.futuresrrrrr functoolsr r r itertoolsr rrrloggingrprrrer0r~r{r1 threadingrrrrrrr cassandrarrrrrrrrrcassandra.authr r!cassandra.connectionr"r#r$r%r&r'r(r)r*r+cassandra.cqltypesr,cassandra.encoderr-cassandra.protocolr.r/r0r1r2r3r4r5r6r7r8r9r:r;r<r=r>r?r@rArBcassandra.metadatarCrDrErFcassandra.policiesrGrHrIrJrKrLrMrNrOrPrQcassandra.poolrRrSrTrUrVrWcassandra.queryrXrYrZr[r\r]r^r_r`rarbrccassandra.marshalrdcassandra.timestampsrecassandra.utilrfrg$cassandra.datastax.insights.reporterrh cassandra.datastax.insights.utilricassandra.datastax.graphrjrkrlrmrnrorprqrrrscassandra.datastax.graph.queryrtrucassandra.datastaxrvrucassandra.io.twistedreactorrwrrryrrzrrrrrrr getLoggerrrconn_fns conn_classrrrrrrrrrrrrrrrrrrrrrregisterrrrrrrr r rrrrr0r5rrwrrrrr/rrCr:rsrrrrrs(        ,0  \ 4 8    0      )i#4j.   ? O M