stack/notebooks/sql_generator.py

import marimo

__generated_with = "0.19.8"
app = marimo.App(width="full")


@app.cell(hide_code=True)
def _():
    import marimo as mo

    return (mo,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    # SQL Transpilation Generator

    Generates **Databricks-dialect SQL** from every pipeline step by running
    narwhals express functions against DuckDB relations and extracting the SQL
    via `.sql_query()`, then transpiling with sqlglot.

    Uses schema-only DuckDB (no data file needed). Each cell below transpiles
    one pipeline module and displays the generated `CREATE OR REPLACE TABLE`
    statements.
    """)
    return


@app.cell(hide_code=True)
def _():
    from aco.lake.transpile import transpile
    from conf import cfg

    DIALECT = "databricks"
    CATALOG = cfg.platform.repo.split("/")[0]  # "homelab"

    def run_transpile(pipeline, label):
        """Transpile a pipeline and return formatted SQL."""
        result = transpile(
            pipeline,
            target_dialect=DIALECT,
            catalog=CATALOG,
            output_mode="ctas",
        )
        parts = []
        errors = 0
        for name, sql in result.items():
            if sql.startswith("-- ERROR"):
                errors += 1
            parts.append(f"-- {name}\n{sql}")
        header = f"/* {label}: {len(result)} steps, {errors} errors */\n"
        return header + "\n\n".join(parts), len(result), errors

    return (run_transpile,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## core
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import core as _core

    _sql, _n, _e = run_transpile(_core.pipeline, "core")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## input_layer
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import input_layer as _input_layer

    _sql, _n, _e = run_transpile(_input_layer.pipeline, "input_layer")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## data_quality
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import data_quality as _data_quality

    _sql, _n, _e = run_transpile(_data_quality.pipeline, "data_quality")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## claims_preprocessing
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import claims_preprocessing as _claims_preprocessing

    _sql, _n, _e = run_transpile(_claims_preprocessing.pipeline, "claims_preprocessing")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## readmissions
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import readmissions as _readmissions

    _sql, _n, _e = run_transpile(_readmissions.pipeline, "readmissions")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## pharmacy
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import pharmacy as _pharmacy

    _sql, _n, _e = run_transpile(_pharmacy.pipeline, "pharmacy")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## ahrq_measures
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import ahrq_measures as _ahrq_measures

    _sql, _n, _e = run_transpile(_ahrq_measures.pipeline, "ahrq_measures")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## quality_measures
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import quality_measures as _quality_measures

    _sql, _n, _e = run_transpile(_quality_measures.pipeline, "quality_measures")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## hcc_suspecting
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import hcc_suspecting as _hcc_suspecting

    _sql, _n, _e = run_transpile(_hcc_suspecting.pipeline, "hcc_suspecting")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## provider_attribution
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import provider_attribution as _provider_attribution

    _sql, _n, _e = run_transpile(_provider_attribution.pipeline, "provider_attribution")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## cclf
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import cclf as _cclf

    _sql, _n, _e = run_transpile(_cclf.pipeline, "cclf")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## main
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import main as _main

    _sql, _n, _e = run_transpile(_main.pipeline, "main")
    mo.md(f"**{_n} steps, {_e} errors**\n\n```sql\n{_sql}\n```")
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ---
    ## Summary
    """)
    return


@app.cell
def _(mo, run_transpile):
    from aco.pipe import (
        ahrq_measures,
        cclf,
        claims_preprocessing,
        core,
        data_quality,
        hcc_suspecting,
        input_layer,
        main,
        pharmacy,
        provider_attribution,
        quality_measures,
        readmissions,
    )

    _modules = [
        ("core", core),
        ("input_layer", input_layer),
        ("data_quality", data_quality),
        ("claims_preprocessing", claims_preprocessing),
        ("readmissions", readmissions),
        ("pharmacy", pharmacy),
        ("ahrq_measures", ahrq_measures),
        ("quality_measures", quality_measures),
        ("hcc_suspecting", hcc_suspecting),
        ("provider_attribution", provider_attribution),
        ("cclf", cclf),
        ("main", main),
    ]

    _rows = []
    _total_steps = 0
    _total_errors = 0
    for _name, _mod in _modules:
        _, _n, _e = run_transpile(_mod.pipeline, _name)
        _total_steps += _n
        _total_errors += _e
        _rows.append(f"| {_name} | {_n} | {_e} |")

    _table = "| Pipeline | Steps | Errors |\n|----------|-------|--------|\n"
    _table += "\n".join(_rows)
    _table += f"\n| **Total** | **{_total_steps}** | **{_total_errors}** |"

    mo.md(_table)
    return


if __name__ == "__main__":
    app.run()