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 co
(self, peers_query_type, connection=None)
| 4123 | return dict((version, list(nodes)) for version, nodes in versions.items()) |
| 4124 | |
| 4125 | def _get_peers_query(self, peers_query_type, connection=None): |
| 4126 | """ |
| 4127 | Determine the peers query to use. |
| 4128 | |
| 4129 | :param peers_query_type: Should be one of PeersQueryType enum. |
| 4130 | |
| 4131 | If _uses_peers_v2 is True, return the proper peers_v2 query (no templating). |
| 4132 | Else, apply the logic below to choose the peers v1 address column name: |
| 4133 | |
| 4134 | Given a connection: |
| 4135 | |
| 4136 | - find the server product version running on the connection's host, |
| 4137 | - use that to choose the column name for the transport address (see APOLLO-1130), and |
| 4138 | - use that column name in the provided peers query template. |
| 4139 | """ |
| 4140 | if peers_query_type not in (self.PeersQueryType.PEERS, self.PeersQueryType.PEERS_SCHEMA): |
| 4141 | raise ValueError("Invalid peers query type: %s" % peers_query_type) |
| 4142 | |
| 4143 | if self._uses_peers_v2: |
| 4144 | if peers_query_type == self.PeersQueryType.PEERS: |
| 4145 | query = self._SELECT_PEERS_V2 if self._token_meta_enabled else self._SELECT_PEERS_NO_TOKENS_V2 |
| 4146 | else: |
| 4147 | query = self._SELECT_SCHEMA_PEERS_V2 |
| 4148 | else: |
| 4149 | if peers_query_type == self.PeersQueryType.PEERS and self._token_meta_enabled: |
| 4150 | query = self._SELECT_PEERS |
| 4151 | else: |
| 4152 | query_template = (self._SELECT_SCHEMA_PEERS_TEMPLATE |
| 4153 | if peers_query_type == self.PeersQueryType.PEERS_SCHEMA |
| 4154 | else self._SELECT_PEERS_NO_TOKENS_TEMPLATE) |
| 4155 | |
| 4156 | host_release_version = self._cluster.metadata.get_host(connection.endpoint).release_version |
| 4157 | host_dse_version = self._cluster.metadata.get_host(connection.endpoint).dse_version |
| 4158 | uses_native_address_query = ( |
| 4159 | host_dse_version and Version(host_dse_version) >= self._MINIMUM_NATIVE_ADDRESS_DSE_VERSION) |
| 4160 | |
| 4161 | if uses_native_address_query: |
| 4162 | query = query_template.format(nt_col_name="native_transport_address") |
| 4163 | elif host_release_version: |
| 4164 | query = query_template.format(nt_col_name="rpc_address") |
| 4165 | else: |
| 4166 | query = self._SELECT_PEERS |
| 4167 | |
| 4168 | return query |
| 4169 | |
| 4170 | def _signal_error(self): |
| 4171 | with self._lock: |
no test coverage detected