Function generate_db_urls

lib/sqlalchemy/testing/provision.py:118–188

Generate a set of URLs to test given configured URLs plus additional driver names. Given: .. sourcecode:: text --dburi postgresql://db1 \ --dburi postgresql://db2 \ --dburi postgresql://db2 \ --dbdriver=psycopg2 --dbdriver=asyncpg Noting that th

(db_urls, extra_drivers)

def generate_db_urls(db_urls, extra_drivers):
    class="st">"""Generate a set of URLs to test given configured URLs plus additional
    driver names.

    Given:

    .. sourcecode:: text

        --dburi postgresql://db1  \
        --dburi postgresql://db2  \
        --dburi postgresql://db2  \
        --dbdriver=psycopg2 --dbdriver=asyncpg

    Noting that the default postgresql driver is psycopg2,  the output
    would be:

    .. sourcecode:: text

        postgresql+psycopg2://db1
        postgresql+asyncpg://db1
        postgresql+psycopg2://db2
        postgresql+psycopg2://db3

    That is, for the driver in a --dburi, we want to keep that and use that
    driver for each URL it&class="cm">#x27;s part of .   For a driver that is only
    in --dbdrivers, we want to use it just once for one of the URLs.
    for a driver that is both coming from --dburi as well as --dbdrivers,
    we want to keep it in that dburi.

    Driver specific query options can be specified by added them to the
    driver name. For example, to a sample option the asyncpg:

    .. sourcecode:: text

        --dburi postgresql://db1  \
        --dbdriver=asyncpg?some_option=a_value

    class="st">"""
    urls = set()

    backend_to_driver_we_already_have = collections.defaultdict(set)

    urls_plus_dialects = [
        (url_obj, url_obj.get_dialect())
        for url_obj in [sa_url.make_url(db_url) for db_url in db_urls]
    ]

    for url_obj, dialect in urls_plus_dialects:
        class="cm"># use get_driver_name instead of dialect.driver to account for
        class="cm"># class="st">"_async" virtual drivers like oracledb and psycopg
        driver_name = url_obj.get_driver_name()
        backend_to_driver_we_already_have[dialect.name].add(driver_name)

    backend_to_driver_we_need = {}

    for url_obj, dialect in urls_plus_dialects:
        backend = dialect.name
        dialect.load_provisioning()