| 61 | |
| 62 | |
| 63 | class TopologyDescription: |
| 64 | def __init__( |
| 65 | self, |
| 66 | topology_type: int, |
| 67 | server_descriptions: dict[_Address, ServerDescription], |
| 68 | replica_set_name: Optional[str], |
| 69 | max_set_version: Optional[int], |
| 70 | max_election_id: Optional[ObjectId], |
| 71 | topology_settings: Any, |
| 72 | ) -> None: |
| 73 | """Representation of a deployment of MongoDB servers. |
| 74 | |
| 75 | :param topology_type: initial type |
| 76 | :param server_descriptions: dict of (address, ServerDescription) for |
| 77 | all seeds |
| 78 | :param replica_set_name: replica set name or None |
| 79 | :param max_set_version: greatest setVersion seen from a primary, or None |
| 80 | :param max_election_id: greatest electionId seen from a primary, or None |
| 81 | :param topology_settings: a TopologySettings |
| 82 | """ |
| 83 | self._topology_type = topology_type |
| 84 | self._replica_set_name = replica_set_name |
| 85 | self._server_descriptions = server_descriptions |
| 86 | self._max_set_version = max_set_version |
| 87 | self._max_election_id = max_election_id |
| 88 | self._candidate_servers = list(self._server_descriptions.values()) |
| 89 | |
| 90 | # The heartbeat_frequency is used in staleness estimates. |
| 91 | self._topology_settings = topology_settings |
| 92 | |
| 93 | # Is PyMongo compatible with all servers' wire protocols? |
| 94 | self._incompatible_err = None |
| 95 | if self._topology_type != TOPOLOGY_TYPE.LoadBalanced: |
| 96 | self._init_incompatible_err() |
| 97 | |
| 98 | # Server Discovery And Monitoring Spec: Whenever a client updates the |
| 99 | # TopologyDescription from an hello response, it MUST set |
| 100 | # TopologyDescription.logicalSessionTimeoutMinutes to the smallest |
| 101 | # logicalSessionTimeoutMinutes value among ServerDescriptions of all |
| 102 | # data-bearing server types. If any have a null |
| 103 | # logicalSessionTimeoutMinutes, then |
| 104 | # TopologyDescription.logicalSessionTimeoutMinutes MUST be set to null. |
| 105 | readable_servers = self.readable_servers |
| 106 | if not readable_servers: |
| 107 | self._ls_timeout_minutes = None |
| 108 | elif any(s.logical_session_timeout_minutes is None for s in readable_servers): |
| 109 | self._ls_timeout_minutes = None |
| 110 | else: |
| 111 | self._ls_timeout_minutes = min( # type: ignore[type-var] |
| 112 | s.logical_session_timeout_minutes for s in readable_servers |
| 113 | ) |
| 114 | |
| 115 | def _init_incompatible_err(self) -> None: |
| 116 | """Internal compatibility check for non-load balanced topologies.""" |
| 117 | for s in self._server_descriptions.values(): |
| 118 | if not s.is_server_type_known: |
| 119 | continue |
| 120 |
no outgoing calls
no test coverage detected