| 865 | |
| 866 | |
| 867 | def _parse_url(name: str) -> URL: |
| 868 | pattern = re.compile( |
| 869 | r""" |
| 870 | (?P<name>[\w\+]+):// |
| 871 | (?: |
| 872 | (?P<username>[^:/]*) |
| 873 | (?::(?P<password>[^@]*))? |
| 874 | @)? |
| 875 | (?: |
| 876 | (?: |
| 877 | \[(?P<ipv6host>[^/\?]+)\] | |
| 878 | (?P<ipv4host>[^/:\?]+) |
| 879 | )? |
| 880 | (?::(?P<port>[^/\?]*))? |
| 881 | )? |
| 882 | (?:/(?P<database>[^\?]*))? |
| 883 | (?:\?(?P<query>.*))? |
| 884 | """, |
| 885 | re.X, |
| 886 | ) |
| 887 | |
| 888 | m = pattern.match(name) |
| 889 | if m is not None: |
| 890 | components = m.groupdict() |
| 891 | query: Optional[Dict[str, Union[str, List[str]]]] |
| 892 | if components["query"] is not None: |
| 893 | query = {} |
| 894 | |
| 895 | for key, value in parse_qsl(components["query"]): |
| 896 | if key in query: |
| 897 | query[key] = util.to_list(query[key]) |
| 898 | cast("List[str]", query[key]).append(value) |
| 899 | else: |
| 900 | query[key] = value |
| 901 | else: |
| 902 | query = None |
| 903 | components["query"] = query |
| 904 | |
| 905 | if components["username"] is not None: |
| 906 | components["username"] = unquote(components["username"]) |
| 907 | |
| 908 | if components["password"] is not None: |
| 909 | components["password"] = unquote(components["password"]) |
| 910 | |
| 911 | ipv4host = components.pop("ipv4host") |
| 912 | ipv6host = components.pop("ipv6host") |
| 913 | components["host"] = ipv4host or ipv6host |
| 914 | name = components.pop("name") |
| 915 | |
| 916 | if components["port"]: |
| 917 | components["port"] = int(components["port"]) |
| 918 | |
| 919 | return URL.create(name, **components) # type: ignore |
| 920 | |
| 921 | else: |
| 922 | raise exc.ArgumentError( |
| 923 | "Could not parse SQLAlchemy URL from given URL string" |
| 924 | ) |