(self, cluster, hosts, keyspace=None)
| 2537 | _graph_paging_available = False |
| 2538 | |
| 2539 | def __init__(self, cluster, hosts, keyspace=None): |
| 2540 | self.cluster = cluster |
| 2541 | self.hosts = hosts |
| 2542 | self.keyspace = keyspace |
| 2543 | |
| 2544 | self._lock = RLock() |
| 2545 | self._pools = {} |
| 2546 | self._profile_manager = cluster.profile_manager |
| 2547 | self._metrics = cluster.metrics |
| 2548 | self._request_init_callbacks = [] |
| 2549 | self._protocol_version = self.cluster.protocol_version |
| 2550 | |
| 2551 | self.encoder = Encoder() |
| 2552 | |
| 2553 | # create connection pools in parallel |
| 2554 | self._initial_connect_futures = set() |
| 2555 | for host in hosts: |
| 2556 | future = self.add_or_renew_pool(host, is_host_addition=False) |
| 2557 | if future: |
| 2558 | self._initial_connect_futures.add(future) |
| 2559 | |
| 2560 | futures = wait_futures(self._initial_connect_futures, return_when=FIRST_COMPLETED) |
| 2561 | while futures.not_done and not any(f.result() for f in futures.done): |
| 2562 | futures = wait_futures(futures.not_done, return_when=FIRST_COMPLETED) |
| 2563 | |
| 2564 | if not any(f.result() for f in self._initial_connect_futures): |
| 2565 | msg = "Unable to connect to any servers" |
| 2566 | if self.keyspace: |
| 2567 | msg += " using keyspace '%s'" % self.keyspace |
| 2568 | raise NoHostAvailable(msg, [h.address for h in hosts]) |
| 2569 | |
| 2570 | self.session_id = uuid.uuid4() |
| 2571 | self._graph_paging_available = self._check_graph_paging_available() |
| 2572 | |
| 2573 | if self.cluster.column_encryption_policy is not None: |
| 2574 | try: |
| 2575 | self.client_protocol_handler = type( |
| 2576 | str(self.session_id) + "-ProtocolHandler", |
| 2577 | (ProtocolHandler,), |
| 2578 | {"column_encryption_policy": self.cluster.column_encryption_policy}) |
| 2579 | except AttributeError: |
| 2580 | log.info("Unable to set column encryption policy for session") |
| 2581 | |
| 2582 | if self.cluster.monitor_reporting_enabled: |
| 2583 | cc_host = self.cluster.get_control_connection_host() |
| 2584 | valid_insights_version = (cc_host and version_supports_insights(cc_host.dse_version)) |
| 2585 | if valid_insights_version: |
| 2586 | self._monitor_reporter = MonitorReporter( |
| 2587 | interval_sec=self.cluster.monitor_reporting_interval, |
| 2588 | session=self, |
| 2589 | ) |
| 2590 | else: |
| 2591 | if cc_host: |
| 2592 | log.debug('Not starting MonitorReporter thread for Insights; ' |
| 2593 | 'not supported by server version {v} on ' |
| 2594 | 'ControlConnection host {c}'.format(v=cc_host.release_version, c=cc_host)) |
| 2595 | |
| 2596 | log.debug('Started Session with client_id {} and session_id {}'.format(self.cluster.client_id, |
nothing calls this directly
no test coverage detected