(
dsn_uri: str, alias_name: str
)
| 68 | |
| 69 | |
| 70 | def expand_dsn_alias_env_vars( |
| 71 | dsn_uri: str, alias_name: str |
| 72 | ) -> tuple[str | None, str | None, str | None, int | None, str, dict[str, list[str]]]: |
| 73 | uri = urlparse(dsn_uri) |
| 74 | username, password, host, port = split_dsn_netloc(uri.netloc) |
| 75 | |
| 76 | expanded_port = expand_dsn_alias_env_var(port, alias_name) |
| 77 | try: |
| 78 | port_number = int(expanded_port) if expanded_port else None |
| 79 | except ValueError as exc: |
| 80 | raise DsnAliasEnvVarError(f'Port in DSN alias {alias_name} must be an integer.') from exc |
| 81 | |
| 82 | params = {key: [expand_dsn_alias_env_var(value, alias_name) or '' for value in values] for key, values in parse_qs(uri.query).items()} |
| 83 | |
| 84 | return ( |
| 85 | expand_dsn_alias_env_var(unquote(username) if username is not None else None, alias_name), |
| 86 | expand_dsn_alias_env_var(unquote(password) if password is not None else None, alias_name), |
| 87 | expand_dsn_alias_env_var(host, alias_name), |
| 88 | port_number, |
| 89 | expand_dsn_alias_env_var(uri.path[1:], alias_name) or '', |
| 90 | params, |
| 91 | ) |
| 92 | |
| 93 | |
| 94 | def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> None: |
no test coverage detected