| """ |
| Concept E — many subjects → many REST endpoints |
| =============================================== |
| |
| A workflow isn't just a canvas — it's an API. Each *disconnected* pipeline |
| (weakly-connected group of nodes ending in a subject) becomes ONE REST endpoint, |
| named after its first subject. Free references become the endpoint's parameters. |
| |
| This graph has two independent pipelines, so it exposes two endpoints: |
| /word_count (text → number) |
| /fahrenheit (number → number) |
| |
| Once launched: |
| curl http://127.0.0.1:7860/gradio_api/call/word_count -s \ |
| -H "Content-Type: application/json" -d '{"data": ["hello there friend"]}' |
| |
| This file also prints its endpoint schema at startup via `describe_workflow_api`. |
| |
| Run it: python concepts/e_multi_endpoint_api.py |
| """ |
|
|
| import json |
| import os |
|
|
| import gradio as gr |
|
|
|
|
| def word_count(text: str) -> int: |
| return len(text.split()) |
|
|
|
|
| def to_fahrenheit(celsius: float) -> float: |
| return round(celsius * 9 / 5 + 32, 1) |
|
|
|
|
| def _ref(nid, label, ptype, default, y): |
| return {"id": nid, "role": "reference", "label": label, "asset_type": ptype, |
| "inputs": [{"id": "in", "label": label, "type": ptype}], |
| "outputs": [{"id": "out", "label": label, "type": ptype}], |
| "data": {"out": default}, "x": 60, "y": y, "width": 200, "height": 90} |
|
|
|
|
| def _op(nid, fn, in_type, out_type, y): |
| return {"id": nid, "role": "operator", "kind": "fn", "fn": fn, "label": fn, |
| "inputs": [{"id": "in_0", "label": "input", "type": in_type, "required": True}], |
| "outputs": [{"id": "out_0", "label": "output", "type": out_type}], |
| "data": {}, "x": 320, "y": y, "width": 200, "height": 90} |
|
|
|
|
| def _sub(nid, label, ptype, y): |
| return {"id": nid, "role": "subject", "label": label, "asset_type": ptype, |
| "inputs": [{"id": "in", "label": label, "type": ptype}], |
| "outputs": [{"id": "out", "label": label, "type": ptype}], |
| "data": {}, "x": 580, "y": y, "width": 200, "height": 120} |
|
|
|
|
| def _edge(eid, s, sp, t, tp, ty): |
| return {"id": eid, "from_node_id": s, "from_port_id": sp, |
| "to_node_id": t, "to_port_id": tp, "type": ty} |
|
|
|
|
| GRAPH = { |
| "schema_version": "2", "name": "Two Endpoints", |
| "references": [_ref("ref_text", "Text", "text", "hello there friend", 60), |
| _ref("ref_c", "Celsius", "number", 20, 260)], |
| "operators": [_op("op_wc", "word_count", "text", "number", 60), |
| _op("op_f", "to_fahrenheit", "number", "number", 260)], |
| "subjects": [_sub("sub_wc", "Word count", "number", 60), |
| _sub("sub_f", "Fahrenheit", "number", 260)], |
| "edges": [_edge("e1", "ref_text", "out", "op_wc", "in_0", "text"), |
| _edge("e2", "op_wc", "out_0", "sub_wc", "in", "number"), |
| _edge("e3", "ref_c", "out", "op_f", "in_0", "number"), |
| _edge("e4", "op_f", "out_0", "sub_f", "in", "number")], |
| } |
|
|
| GRAPH_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "e_multi_endpoint_api.json") |
| with open(GRAPH_PATH, "w", encoding="utf-8") as f: |
| json.dump(GRAPH, f, indent=2) |
|
|
| demo = gr.Workflow(GRAPH_PATH, bind={"word_count": word_count, "to_fahrenheit": to_fahrenheit}) |
|
|
| if __name__ == "__main__": |
| from gradio.workflow_api import WorkflowGraph, describe_workflow_api |
| for ep in describe_workflow_api(WorkflowGraph.from_json(json.dumps(GRAPH))): |
| print(f" {ep['api_name']:14s} params={[p['type'] for p in ep['parameters']]} " |
| f"returns={[r['type'] for r in ep['returns']]}") |
| demo.launch() |
|
|
