stack/notebooks/api_explorer.py

import marimo

__generated_with = "0.20.4"
app = marimo.App(width="medium")


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

    mo.md("""
    # API Explorer

    Interactive test harness for the Stack REST API. Exercises every endpoint
    group: **health**, **pipelines**, **lineage**, **bib**, **schema**, and **auth**.

    Defaults to the compose-internal `api` service; set `STACK_API_URL` to
    point elsewhere (e.g. `http://localhost:8080` with a local
    `uv run stack api serve --port 8080`).
    """)
    return (mo,)


@app.cell
def _():
    import os

    import httpx

    BASE = os.environ.get("STACK_API_URL", "http://api:8000")
    client = httpx.Client(base_url=BASE, timeout=30)
    return (client,)


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


@app.cell
def _(client, mo):
    _r = client.get("/health")
    health = _r.json()

    _rows = []
    for _svc in health["services"]:
        _icon = "\u2705" if _svc["status"] == "ok" else "\u26a0\ufe0f"
        _rows.append(
            f"| {_icon} | {_svc['name']} | {_svc['status']} | {_svc.get('detail', '')} |"
        )

    mo.md(f"""
    **Status:** {health["status"]} | **Version:** {health["version"]}

    | | Service | Status | Detail |
    |---|---------|--------|--------|
    {chr(10).join(_rows)}
    """)
    return (health,)


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


@app.cell
def _(client, mo):
    _r = client.get("/pipelines")
    pipelines = _r.json()

    _rows = []
    for _p in pipelines:
        _rows.append(f"| {_p['name']} | {_p['steps']} |")

    mo.md(f"""
    **{len(pipelines)} pipelines registered**

    | Pipeline | Steps |
    |----------|-------|
    {chr(10).join(_rows)}
    """)
    return (pipelines,)


@app.cell
def _(client, mo, pipelines):
    _name = pipelines[0]["name"] if pipelines else "readmissions"
    _r = client.get(f"/pipelines/{_name}")
    detail = _r.json()

    _inputs_list = "\n".join(f"- `{i}`" for i in detail.get("inputs", []))
    _outputs_list = "\n".join(f"- `{o}`" for o in detail.get("outputs", []))

    mo.md(f"""
    ### Pipeline Detail: `{detail["name"]}`

    **Steps:** {detail["steps"]} | **Inputs:** {len(detail.get("inputs", []))} | **Outputs:** {len(detail.get("outputs", []))}

    **Inputs:**
    {_inputs_list}

    **Outputs:**
    {_outputs_list}
    """)
    return


@app.cell
def _(client, mo):
    _r = client.get("/pipelines/nonexistent_xyz")
    pipeline_404 = _r.status_code == 404
    mo.md(
        f"**GET /pipelines/nonexistent_xyz** \u2192 `{_r.status_code}` {'PASS' if pipeline_404 else 'FAIL'}"
    )
    return (pipeline_404,)


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


@app.cell
def _(client, mo):
    _r = client.get("/lineage")
    lineage = _r.json()

    table_count = len(lineage["tables"])
    edge_count = sum(len(v) for v in lineage["edges"].values())
    col_sources = sum(len(v) for v in lineage["column_sources"].values())

    mo.md(f"""
    **Full lineage graph:** {table_count} tables, {edge_count} edges, {col_sources} column-level sources
    """)
    return (lineage,)


@app.cell
def _(client, lineage, mo):
    _test_table = None
    for _t, _deps in lineage["edges"].items():
        if _deps and "." in _t:
            _test_table = _t
            break

    _content = ""
    if _test_table:
        _r = client.get(f"/lineage/{_test_table}")
        _tl = _r.json()
        _content = f"""
    ### Table Lineage: `{_tl["table"]}`

    **Inputs:** {", ".join(f"`{i}`" for i in _tl["inputs"])}
    """
    else:
        _content = "No table with dependencies found"

    _ = mo.md(_content)
    return


@app.cell
def _(client, mo):
    _r = client.get("/lineage/export/mermaid")
    mermaid_data = _r.json()
    _lines = mermaid_data["content"].split("\n")

    mo.md(f"""
    ### Mermaid Export

    **Format:** {mermaid_data["format"]} | **Lines:** {len(_lines)}

    ```mermaid
    {chr(10).join(_lines[:15])}
    ...
    ```
    """)
    return


@app.cell
def _(client, mo):
    _r = client.get("/lineage/export/dot")
    dot_data = _r.json()
    _lines = dot_data["content"].split("\n")

    mo.md(f"""
    ### DOT Export

    **Format:** {dot_data["format"]} | **Lines:** {len(_lines)}

    ```dot
    {chr(10).join(_lines[:10])}
    ...
    ```
    """)
    return


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


@app.cell
def _(client, mo):
    _r1 = client.get("/bib/items?limit=10")
    bib_items = _r1.json()

    _r2 = client.get("/bib/tags")
    bib_tags = _r2.json()

    _item_rows = []
    for _item in bib_items[:10]:
        _item_rows.append(
            f"| {_item['key']} | {_item['title'][:50]} | {_item['item_type']} |"
        )

    _tag_rows = []
    for _tag in bib_tags[:10]:
        _tag_rows.append(f"| {_tag['namespace']} | {_tag['count']} |")

    mo.md(f"""
    **{len(bib_items)} items returned** | **{len(bib_tags)} tag namespaces**

    ### Items (first 10)

    | Key | Title | Type |
    |-----|-------|------|
    {chr(10).join(_item_rows)}

    ### Tag Namespaces

    | Namespace | Count |
    |-----------|-------|
    {chr(10).join(_tag_rows)}
    """)
    return


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


@app.cell
def _(client, mo):
    _r = client.get("/schema/nonexistent_table")
    missing_ok = _r.status_code == 404
    mo.md(
        f"**GET /schema/nonexistent_table** \u2192 `{_r.status_code}` {'PASS' if missing_ok else 'FAIL'}"
    )
    return (missing_ok,)


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


@app.cell
def _(client, mo):
    _r1 = client.post("/pipelines/run/readmissions")
    auth_required = _r1.status_code in (401, 403)

    _r2 = client.post("/auth/token", json={"secret": "wrong"})
    wrong_secret = _r2.status_code == 401

    mo.md(f"""
    | Test | Status | Result |
    |------|--------|--------|
    | Run requires auth | `{_r1.status_code}` | {"PASS" if auth_required else "FAIL"} |
    | Wrong secret rejected | `{_r2.status_code}` | {"PASS" if wrong_secret else "FAIL"} |
    """)
    return auth_required, wrong_secret


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


@app.cell
def _(
    auth_required,
    client,
    health,
    lineage,
    missing_ok,
    mo,
    pipeline_404,
    pipelines,
    wrong_secret,
):
    tests = {
        "GET /health": health["status"] in ("ok", "degraded"),
        "GET /pipelines": len(pipelines) > 0,
        "GET /pipelines/{name}": client.get(
            f"/pipelines/{pipelines[0]['name']}"
        ).status_code
        == 200,
        "GET /pipelines/404": pipeline_404,
        "GET /lineage": len(lineage["tables"]) > 0,
        "GET /lineage/export/mermaid": client.get("/lineage/export/mermaid").status_code
        == 200,
        "GET /lineage/export/dot": client.get("/lineage/export/dot").status_code == 200,
        "GET /bib/items": client.get("/bib/items").status_code == 200,
        "GET /bib/tags": client.get("/bib/tags").status_code == 200,
        "GET /schema/404": missing_ok,
        "POST /pipelines/run (no auth)": auth_required,
        "POST /auth/token (wrong)": wrong_secret,
    }

    passed = sum(1 for v in tests.values() if v)
    total = len(tests)

    _rows = []
    for _name, _ok in tests.items():
        _icon = "\u2705" if _ok else "\u274c"
        _rows.append(f"| {_icon} | {_name} | {'PASS' if _ok else 'FAIL'} |")

    mo.md(f"""
    ### Results: {passed}/{total} passed

    | | Endpoint | Result |
    |---|----------|--------|
    {chr(10).join(_rows)}
    """)
    return


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