LangGraph

LangGraph nodes for Hotdata — wire SQL execution and database management into your graphs as first-class nodes.

Install

pip install hotdata-langgraph

Authentication

Set HOTDATA_API_KEY in your environment. Optionally set HOTDATA_WORKSPACE to pin a specific workspace.

Quickstart

from hotdata_runtime import from_env
from hotdata_langgraph import make_execute_sql_node
from langgraph.graph import StateGraph
from typing import TypedDict, Any

client = from_env()

class State(TypedDict):
    sql: str
    result: Any

builder = StateGraph(State)
builder.add_node("query", make_execute_sql_node(client=client))
builder.set_entry_point("query")
builder.set_finish_point("query")

graph = builder.compile()
state = graph.invoke({"sql": "SELECT COUNT(*) AS total FROM orders"})
print(state["result"].to_records())

Nodes

SQL execution

from hotdata_langgraph import make_execute_sql_node, execute_sql_node

# Factory — binds client and key names, returns a callable node
node = make_execute_sql_node(
    client=client,
    sql_key="sql",        # state key to read SQL from (default: "sql")
    output_key="result",  # state key to write QueryResult to (default: "result")
    database=None,        # pin to a specific database id (optional)
)

# Or call the node directly if you prefer
output = execute_sql_node(
    {"sql": "SELECT 1 AS ok"},
    client=client,
)
result = output["result"]
rows = result.to_records()

Process results

from hotdata_langgraph import result_rows_for_llm

# Trim a QueryResult to a list of dicts for passing to an LLM
rows = result_rows_for_llm(result, max_rows=20)

Managed database nodes

Wire database creation and table loading as graph nodes:

from hotdata_langgraph import (
    make_create_managed_database_node,
    make_load_managed_table_node,
)
from langgraph.graph import StateGraph
from typing import TypedDict, Any

class State(TypedDict):
    database_name: str
    tables: list[str]
    database: Any
    table: str
    file: str
    load_result: Any

builder = StateGraph(State)
builder.add_node(
    "create_db",
    make_create_managed_database_node(
        client=client,
        name_key="database_name",   # state key for the database name
        tables_key="tables",        # state key for the table list
        output_key="database",      # state key to write ManagedDatabase to
    ),
)
builder.add_node(
    "load_table",
    make_load_managed_table_node(
        client=client,
        database_key="database",    # state key holding the ManagedDatabase
        table_key="table",
        file_key="file",            # or upload_id_key="upload_id"
        output_key="load_result",
    ),
)
builder.set_entry_point("create_db")
builder.add_edge("create_db", "load_table")
builder.set_finish_point("load_table")

graph = builder.compile()
state = graph.invoke({
    "database_name": "sales",
    "tables": ["orders"],
    "table": "orders",
    "file": "orders.parquet",
})
print(state["load_result"].full_name)

Reference

See also

• hotdata-langgraph on GitHub
• LangGraph
• Python SDK — core HotdataClient reference