def get_db_connection_string()

in images/airflow/2.9.2/python/mwaa/config/database.py [0:0]

• 29 lines of code
• 3 McCabe index (conditional complexity)

def get_db_connection_string() -> str:
    """
    Retrieve the connection string for communicating with metadata database.

    :returns The connection string.

    :raises RuntimeError if the required environment variables are not set.
    """
    env_vars_names = [
        "MWAA__DB__POSTGRES_HOST",
        "MWAA__DB__POSTGRES_PORT",
        "MWAA__DB__POSTGRES_DB",
        "MWAA__DB__POSTGRES_SSLMODE",
    ]
    try:
        (
            postgres_host,
            postgres_port,
            postgres_db,
            postgres_sslmode,
        ) = itemgetter(*env_vars_names)(os.environ)
        (postgres_user, postgres_password) = get_db_credentials()
    except Exception as e:
        raise RuntimeError(
            "One or more of the required environment variables for "
            "configuring Postgres are not set. Please ensure you set all "
            "all the following environment variables: "
            f'{", ".join(env_vars_names)}. This was the result of the '
            f"following exception: {e}"
        )

    if not postgres_sslmode:
        postgres_sslmode = "require"

    protocol = "postgresql+psycopg2"
    creds = f"{postgres_user}:{postgres_password}"
    addr = f"{postgres_host}:{postgres_port}"
    return f"{protocol}://{creds}@{addr}/{postgres_db}?sslmode={postgres_sslmode}"