Format an error message if server selection fails. Hold the lock when calling this.
(self, selector: Callable[[Selection], Selection])
| 1014 | ) |
| 1015 | |
| 1016 | def _error_message(self, selector: Callable[[Selection], Selection]) -> str: |
| 1017 | """Format an error message if server selection fails. |
| 1018 | |
| 1019 | Hold the lock when calling this. |
| 1020 | """ |
| 1021 | is_replica_set = self._description.topology_type in ( |
| 1022 | TOPOLOGY_TYPE.ReplicaSetWithPrimary, |
| 1023 | TOPOLOGY_TYPE.ReplicaSetNoPrimary, |
| 1024 | ) |
| 1025 | |
| 1026 | if is_replica_set: |
| 1027 | server_plural = "replica set members" |
| 1028 | elif self._description.topology_type == TOPOLOGY_TYPE.Sharded: |
| 1029 | server_plural = "mongoses" |
| 1030 | else: |
| 1031 | server_plural = "servers" |
| 1032 | |
| 1033 | if self._description.known_servers: |
| 1034 | # We've connected, but no servers match the selector. |
| 1035 | if selector is writable_server_selector: |
| 1036 | if is_replica_set: |
| 1037 | return "No primary available for writes" |
| 1038 | else: |
| 1039 | return "No %s available for writes" % server_plural |
| 1040 | else: |
| 1041 | return f'No {server_plural} match selector "{selector}"' |
| 1042 | else: |
| 1043 | addresses = list(self._description.server_descriptions()) |
| 1044 | servers = list(self._description.server_descriptions().values()) |
| 1045 | if not servers: |
| 1046 | if is_replica_set: |
| 1047 | # We removed all servers because of the wrong setName? |
| 1048 | return 'No {} available for replica set name "{}"'.format( |
| 1049 | server_plural, |
| 1050 | self._settings.replica_set_name, |
| 1051 | ) |
| 1052 | else: |
| 1053 | return "No %s available" % server_plural |
| 1054 | |
| 1055 | # 1 or more servers, all Unknown. Are they unknown for one reason? |
| 1056 | error = servers[0].error |
| 1057 | same = all(server.error == error for server in servers[1:]) |
| 1058 | if same: |
| 1059 | if error is None: |
| 1060 | # We're still discovering. |
| 1061 | return "No %s found yet" % server_plural |
| 1062 | |
| 1063 | if is_replica_set and not set(addresses).intersection(self._seed_addresses): |
| 1064 | # We replaced our seeds with new hosts but can't reach any. |
| 1065 | return ( |
| 1066 | "Could not reach any servers in %s. Replica set is" |
| 1067 | " configured with internal hostnames or IPs?" % addresses |
| 1068 | ) |
| 1069 | |
| 1070 | return str(error) |
| 1071 | else: |
| 1072 | return ",".join(str(server.error) for server in servers if server.error) |
| 1073 |
no test coverage detected