def get_available_engine_specs()

in superset/db_engine_specs/__init__.py [0:0]

• 54 lines of code
• 22 McCabe index (conditional complexity)

def get_available_engine_specs() -> dict[type[BaseEngineSpec], set[str]]:  # noqa: C901
    """
    Return available engine specs and installed drivers for them.
    """
    drivers: dict[str, set[str]] = defaultdict(set)

    # native SQLAlchemy dialects
    for attr in sqlalchemy.dialects.__all__:
        try:
            dialect = sqlalchemy.dialects.registry.load(attr)
            if (
                issubclass(dialect, DefaultDialect)
                and hasattr(dialect, "driver")
                # adodbapi dialect is removed in SQLA 1.4 and doesn't implement the
                # `dbapi` method, hence needs to be ignored to avoid logging a warning
                and dialect.driver != "adodbapi"
            ):
                try:
                    dialect.dbapi()
                except ModuleNotFoundError:
                    continue
                except Exception as ex:  # pylint: disable=broad-except
                    logger.warning("Unable to load dialect %s: %s", dialect, ex)
                    continue
                drivers[attr].add(dialect.driver)
        except NoSuchModuleError:
            continue

    # installed 3rd-party dialects
    for ep in entry_points(group="sqlalchemy.dialects"):
        try:
            dialect = ep.load()
        except Exception as ex:  # pylint: disable=broad-except
            logger.warning("Unable to load SQLAlchemy dialect %s: %s", ep.name, ex)
        else:
            backend = dialect.name
            if isinstance(backend, bytes):
                backend = backend.decode()
            backend = backend_replacements.get(backend, backend)

            driver = getattr(dialect, "driver", dialect.name)
            if isinstance(driver, bytes):
                driver = driver.decode()
            drivers[backend].add(driver)

    dbs_denylist = app.config["DBS_AVAILABLE_DENYLIST"]
    if not feature_flag_manager.is_feature_enabled("ENABLE_SUPERSET_META_DB"):
        dbs_denylist["superset"] = {""}
    dbs_denylist_engines = dbs_denylist.keys()
    available_engines = {}

    for engine_spec in load_engine_specs():
        driver = drivers[engine_spec.engine]
        if (
            engine_spec.engine in dbs_denylist_engines
            and hasattr(engine_spec, "default_driver")
            and engine_spec.default_driver in dbs_denylist[engine_spec.engine]
        ):
            # do not add denied db engine specs to available list
            continue

        # lookup driver by engine aliases.
        if not driver and engine_spec.engine_aliases:
            for alias in engine_spec.engine_aliases:
                driver = drivers[alias]
                if driver:
                    break

        available_engines[engine_spec] = driver

    return available_engines