(
uri: str,
default_port: Optional[int] = DEFAULT_PORT,
validate: bool = True,
warn: bool = False,
normalize: bool = True,
connect_timeout: Optional[float] = None,
srv_service_name: Optional[str] = None,
srv_max_hosts: Optional[int] = None,
)
| 122 | |
| 123 | |
| 124 | async def _parse_srv( |
| 125 | uri: str, |
| 126 | default_port: Optional[int] = DEFAULT_PORT, |
| 127 | validate: bool = True, |
| 128 | warn: bool = False, |
| 129 | normalize: bool = True, |
| 130 | connect_timeout: Optional[float] = None, |
| 131 | srv_service_name: Optional[str] = None, |
| 132 | srv_max_hosts: Optional[int] = None, |
| 133 | ) -> dict[str, Any]: |
| 134 | if uri.startswith(SCHEME): |
| 135 | is_srv = False |
| 136 | scheme_free = uri[SCHEME_LEN:] |
| 137 | else: |
| 138 | is_srv = True |
| 139 | scheme_free = uri[SRV_SCHEME_LEN:] |
| 140 | |
| 141 | options = _CaseInsensitiveDictionary() |
| 142 | |
| 143 | host_plus_db_part, _, opts = scheme_free.partition("?") |
| 144 | if "/" in host_plus_db_part: |
| 145 | host_part, _, _ = host_plus_db_part.partition("/") |
| 146 | else: |
| 147 | host_part = host_plus_db_part |
| 148 | |
| 149 | if opts: |
| 150 | options.update(split_options(opts, validate, warn, normalize)) |
| 151 | if srv_service_name is None: |
| 152 | srv_service_name = options.get("srvServiceName", SRV_SERVICE_NAME) |
| 153 | if "@" in host_part: |
| 154 | _, _, hosts = host_part.rpartition("@") |
| 155 | else: |
| 156 | hosts = host_part |
| 157 | |
| 158 | hosts = unquote_plus(hosts) |
| 159 | srv_max_hosts = srv_max_hosts or options.get("srvMaxHosts") |
| 160 | if is_srv: |
| 161 | nodes = split_hosts(hosts, default_port=None) |
| 162 | fqdn, port = nodes[0] |
| 163 | |
| 164 | # Use the connection timeout. connectTimeoutMS passed as a keyword |
| 165 | # argument overrides the same option passed in the connection string. |
| 166 | connect_timeout = connect_timeout or options.get("connectTimeoutMS") |
| 167 | dns_resolver = _SrvResolver(fqdn, connect_timeout, srv_service_name, srv_max_hosts) |
| 168 | nodes = await dns_resolver.get_hosts() |
| 169 | dns_options = await dns_resolver.get_options() |
| 170 | if dns_options: |
| 171 | parsed_dns_options = split_options(dns_options, validate, warn, normalize) |
| 172 | if set(parsed_dns_options) - _ALLOWED_TXT_OPTS: |
| 173 | raise ConfigurationError( |
| 174 | "Only authSource, replicaSet, and loadBalanced are supported from DNS" |
| 175 | ) |
| 176 | for opt, val in parsed_dns_options.items(): |
| 177 | if opt not in options: |
| 178 | options[opt] = val |
| 179 | if options.get("loadBalanced") and srv_max_hosts: |
| 180 | raise InvalidURI("You cannot specify loadBalanced with srvMaxHosts") |
| 181 | if options.get("replicaSet") and srv_max_hosts: |
no test coverage detected