Catalog schema

Your state machine needs real data. A list of doctors by specialty, a list of appointments by patient. Storing this in SQLite keeps the workshop runnable on a laptop while teaching the right pattern: a thin data layer with focused helpers, called from a graph node that contains zero SQL.

db_utils.py
python
def init_db() -> None:
    """Create tables and seed doctors on first run."""
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS doctors (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                specialty TEXT NOT NULL,
                years_experience INTEGER NOT NULL
            )
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS appointments (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                patient_id TEXT NOT NULL,
                doctor_id INTEGER NOT NULL,
                doctor_name TEXT NOT NULL,
                specialty TEXT NOT NULL,
                slot TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'confirmed',
                created_at TEXT NOT NULL,
                FOREIGN KEY (doctor_id) REFERENCES doctors(id)
            )
            """
        )

Two tables: a catalog (doctors) and a transactional log (appointments). Specialty is a text column that matches the closed vocabulary used by the classifier node. That alignment is the whole reason the classifier works.

db_utils.py
python
def list_doctors_by_specialty(specialty: str) -> List[Dict[str, Any]]:
    """Return doctors that match a specialty label, e.g. 'cardiology'."""
    init_db()
    conn = _connect()
    try:
        rows = conn.execute(
            "SELECT id, name, specialty, years_experience "
            "FROM doctors WHERE specialty = ? "
            "ORDER BY years_experience DESC",
            (specialty,),
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()

One function, one query, parameterized for safety. The graph node calls this helper and receives plain dicts. No SQL in the node. No ORM leakage. Swap SQLite for Postgres later by rewriting this file, and the graph does not notice.

booking_graph.py
python
from db_utils import list_doctors_by_specialty

async def find_doctor(state: BookingState) -> BookingState:
    specialty = state.get("specialty") or "general_medicine"
    doctors = list_doctors_by_specialty(specialty)
    if not doctors:
        doctors = list_doctors_by_specialty("general_medicine")
    state["doctor"] = doctors[0] if doctors else None
    _record_event(
        state,
        "find_doctor",
        {"doctor": state["doctor"]["name"] if state.get("doctor") else None},
    )
    return state

The graph node is four lines of business logic. Fetch by specialty, degrade to general medicine if empty, pick the most experienced, record the event. All the SQL noise is behind list_doctors_by_specialty.

Quiz: Quiz

Loading practice…