def register_sqlalchemy()

in databao/duckdb/utils.py [0:0]

• 20 lines of code
• 4 McCabe index (conditional complexity)

def register_sqlalchemy(con: DuckDBPyConnection, sqlalchemy_engine: Engine, name: str) -> None:
    """Attach an external DB to DuckDB using an existing SQLAlchemy engine.

    Supports PostgreSQL and MySQL/MariaDB (via DuckDB extensions). The external
    database becomes available under the given `name` within the DuckDB connection.
    """
    sa_url = sqlalchemy_engine.url.render_as_string(hide_password=False)
    dialect = getattr(getattr(sqlalchemy_engine, "dialect", None), "name", "")
    if dialect.startswith("postgres"):
        con.execute("INSTALL postgres;")
        con.execute("LOAD postgres;")
        pg_url = sqlalchemy_to_postgres_url(sqlalchemy_engine.url)
        con.execute(f"ATTACH '{pg_url}' AS {name} (TYPE POSTGRES);")
    elif dialect.startswith(("mysql", "mariadb")):
        con.execute("INSTALL mysql;")
        con.execute("LOAD mysql;")
        mysql_url = sqlalchemy_to_duckdb_mysql(sa_url)
        con.execute(f"ATTACH '{mysql_url}' AS {name} (TYPE MYSQL);")
    elif dialect.startswith("sqlite"):
        con.execute("INSTALL sqlite;")
        con.execute("LOAD sqlite;")
        sqlite_path = re.sub("^sqlite:///", "", sa_url)
        con.execute(f"ATTACH '{sqlite_path}' AS {name} (TYPE SQLITE);")
    else:
        raise ValueError(f"Database engine '{dialect}' is not supported yet")