hotdata-langgraph

LangGraph nodes for Hotdata — run SQL and manage databases as first-class graph nodes that read from and write to your state dict.

Install

pip install hotdata-langgraph

Authentication

Set HOTDATA_API_KEY in your environment. Optionally set HOTDATA_WORKSPACE to pin a specific workspace (the first available workspace is used if unset).

Quickstart

from langgraph.graph import StateGraph
import hotdata_langgraph as hlg

client = hlg.from_env()

builder = StateGraph(dict)
builder.add_node("run_sql", hlg.make_execute_sql_node(client=client))
builder.set_entry_point("run_sql")
builder.set_finish_point("run_sql")

graph = builder.compile()
result = graph.invoke({"sql": "SELECT * FROM orders LIMIT 10"})
print(result["result"])  # list of row dicts

SQL node

The SQL node reads a query from state, executes it on Hotdata, and writes the rows back to state.

# Factory: create a reusable node function
execute_sql = hlg.make_execute_sql_node(client=client)
builder.add_node("run_sql", execute_sql)

# State in:  {"sql": "SELECT ..."}
# State out: {"result": [{"col": val, ...}, ...]}

Managed database nodes

# Create a managed database
create_db = hlg.make_create_managed_database_node(client=client)
# State in:  {"database_name": "sales", "schema": "public", "tables": "orders,customers"}
# State out: {"database": {"id": "...", "description": "sales"}}

# Load a parquet file into a table
load_table = hlg.make_load_managed_table_node(client=client)
# State in:  {"database": "sales", "table": "orders", "file": "/path/to/orders.parquet"}
# State out: {"load_result": {"full_name": "sales.public.orders", "row_count": 1500, ...}}

Scoping queries to a managed database

Pass database= so all SQL the node runs resolves against a specific managed database:

execute_sql = hlg.make_execute_sql_node(client=client, database="sales")

Custom state keys

Override the default key names to fit your graph's state schema:

execute_sql = hlg.make_execute_sql_node(
    client=client,
    sql_key="query",    # reads from state["query"] instead of state["sql"]
    output_key="rows",  # writes to state["rows"] instead of state["result"]
)

The same *_key pattern applies to all nodes.

Run the example

uv run python examples/langgraph_basic.py

Development

uv sync --locked
uv run pytest