(
uri: str,
default_port: Optional[int] = DEFAULT_PORT,
validate: bool = True,
warn: bool = False,
normalize: bool = True,
srv_max_hosts: Optional[int] = None,
)
| 508 | |
| 509 | |
| 510 | def _validate_uri( |
| 511 | uri: str, |
| 512 | default_port: Optional[int] = DEFAULT_PORT, |
| 513 | validate: bool = True, |
| 514 | warn: bool = False, |
| 515 | normalize: bool = True, |
| 516 | srv_max_hosts: Optional[int] = None, |
| 517 | ) -> dict[str, Any]: |
| 518 | if uri.startswith(SCHEME): |
| 519 | is_srv = False |
| 520 | scheme_free = uri[SCHEME_LEN:] |
| 521 | elif uri.startswith(SRV_SCHEME): |
| 522 | if not _have_dnspython(): |
| 523 | python_path = sys.executable or "python" |
| 524 | raise ConfigurationError( |
| 525 | 'The "dnspython" module must be ' |
| 526 | "installed to use mongodb+srv:// URIs. " |
| 527 | "To fix this error install pymongo again:\n " |
| 528 | "%s -m pip install pymongo>=4.3" % (python_path) |
| 529 | ) |
| 530 | is_srv = True |
| 531 | scheme_free = uri[SRV_SCHEME_LEN:] |
| 532 | else: |
| 533 | raise InvalidURI(f"Invalid URI scheme: URI must begin with '{SCHEME}' or '{SRV_SCHEME}'") |
| 534 | |
| 535 | if not scheme_free: |
| 536 | raise InvalidURI("Must provide at least one hostname or IP") |
| 537 | |
| 538 | user = None |
| 539 | passwd = None |
| 540 | dbase = None |
| 541 | collection = None |
| 542 | options = _CaseInsensitiveDictionary() |
| 543 | |
| 544 | host_plus_db_part, _, opts = scheme_free.partition("?") |
| 545 | if "/" in host_plus_db_part: |
| 546 | host_part, _, dbase = host_plus_db_part.partition("/") |
| 547 | else: |
| 548 | host_part = host_plus_db_part |
| 549 | |
| 550 | if dbase: |
| 551 | dbase = unquote_plus(dbase) |
| 552 | if "." in dbase: |
| 553 | dbase, collection = dbase.split(".", 1) |
| 554 | if _BAD_DB_CHARS.search(dbase): |
| 555 | raise InvalidURI('Bad database name "%s"' % dbase) |
| 556 | else: |
| 557 | dbase = None |
| 558 | |
| 559 | if opts: |
| 560 | options.update(split_options(opts, validate, warn, normalize)) |
| 561 | if "@" in host_part: |
| 562 | userinfo, _, hosts = host_part.rpartition("@") |
| 563 | user, passwd = parse_userinfo(userinfo) |
| 564 | else: |
| 565 | hosts = host_part |
| 566 | |
| 567 | if "/" in hosts: |
no test coverage detected