Can't Connect PostgreSQL Database When Connection URIs Contains Special Characters

[LeeWantfly]

Hello, I had met a problem, it's failed when i was connecting my PostgreSQL database, i had read the pg documents, https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING, Because my passwords contains special characters '@', i had to convert my passwords with urllib.parse.quote_plus() function, but it was still failed. By contrast, it was success when i connecting another PostgreSQL database without special charaters passwords。 Here is the code:

from peewee_aio import Manager
manager = Manager('postgresql://user:pass%40word123@myhost:5432/mydatabase')
.....

Traceback (most recent call last): File "D:\Miniconda3\envs\env_myproj\lib\asyncio\windows_events.py", line 457, in finish_recv return ov.getresult() OSError: [WinError 64] The network name specified is no longer available

[klen]

@LeeWantfly try to use password without encoding like 'postgresql://user:pass@word123@myhost:5432/mydatabase'

[Insighttful]

Just roll your own and make sure you use the unquote_password param set to True when connecting.

# thirdparty
from playhouse.db_url import connect

db_params = {...}
db_url = construct_db_url(db_params, quote_pw: True)
connect(db_url, unquote_password=True)

Where construct_db_url is defined as:

# stdlib
from typing import Dict, Optional
from urllib.parse import urlparse, quote_plus

def deconstruct_db_url(
    url: str, quote_pw: bool = True, w_db: bool = True
) -> Dict[str, Optional[str]]:
    """Converts a database URL to a dictionary of parameters and quoted password if required.

    Args:
        url (str): The database URL to convert.
        quote_pw (bool): Whether to URL-encode the password (default: True).
        w_db (bool): Whether to include the database name in the dictionary (default: True).

    Returns:
        dict: A dictionary containing the parsed database parameters.
    """
    parsed = urlparse(url)
    params = {
        "dialect": (scheme := parsed.scheme).split("+")[0].lower(),
        "driver": scheme.split("+")[1].lower() if "+" in scheme else None,
        "username": parsed.username,
        "password": quote_plus(parsed.password) if quote_pw else parsed.password,
        "host": parsed.hostname,
        "port": parsed.port,
        "database": (path := parsed.path[1:]) if parsed.path and w_db else None,
    }
    return {k: None if (v := params.get(k)) == "" else v for k in params}

def construct_db_url(
    params: Dict[str, Optional[str]],
    quote_pw: bool = False,
    w_db: bool = True,
    w_driver: bool = False,
) -> str:
    """Converts a dictionary of parameters to a database URL and quotes password if required.

    Args:
        params (dict): The dictionary of database parameters.
        quote_pw (bool): Whether to URL-encode the password (default: False).
        w_db (bool): Whether to include the database name in the URL (default: True).
        w_driver (bool): Whether to include the driver in the URL (default: False).

    Returns:
        str: The database URL.
    """

    required_keys = ["dialect", "username", "password", "host", "port"]
    if w_db:
        required_keys.append("database")
    if w_driver:
        required_keys.append("driver")

    for key in required_keys:
        if key not in params:
            raise KeyError(f"Missing required key: {key}")

    params["password"] = (
        quote_plus(params["password"]) if quote_pw else params["password"]
    )
    params["database"] = params["database"] if w_db else None
    scheme = (dialect := params["dialect"]) + (
        f"+{params['driver']}" if w_driver else ""
    )
    db_url = f"{scheme}://{params['username']}:{params['password']}@{params['host']}:{params['port']}"
    return f"{db_url}/{params['database']}" if w_db else db_url