147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
"""Derive a web-form schema from a Langflow export.
|
|
|
|
Looks for ``CustomInput*`` nodes (``data.type`` or the inner
|
|
``display_name`` starting with ``CustomInput`` — see
|
|
``backend/components/kirche/form_inputs.py``) and maps their ``template``
|
|
(minus ``code``) to ``{json_schema, ui_schema}`` for RJSF:
|
|
|
|
* ``name`` → property key, ``display_name`` → ``title``,
|
|
``info`` → ``description``, ``value`` → ``default``,
|
|
``options`` → ``enum``, ``required`` → required array
|
|
* ``show: false`` fields are plumbing (e.g. the run envelope input), not
|
|
form fields, and are skipped
|
|
* ``advanced`` fields render last via ``ui:order``
|
|
* ``_input_type`` → JSON type + widget (unknown types fall back to string
|
|
so the parser never crashes on new fields)
|
|
|
|
A flow without such a node gets ``{fallback: "textarea"}``: the frontend
|
|
renders one free-text box instead of a generated form.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
def _form_templates(langflow_json: dict) -> list[dict]:
|
|
"""Inner node dicts (``data.node``) of all CustomInput* nodes."""
|
|
if not isinstance(langflow_json, dict):
|
|
return []
|
|
data = langflow_json.get("data") or {}
|
|
nodes = data.get("nodes") or []
|
|
found = []
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
node_data = node.get("data") or {}
|
|
node_type = str(node_data.get("type") or "")
|
|
inner = node_data.get("node") or {}
|
|
display = str(inner.get("display_name") or "")
|
|
if node_type.startswith("CustomInput") or display.startswith("CustomInput"):
|
|
found.append(inner)
|
|
return found
|
|
|
|
|
|
def _field_schema(name: str, field: dict) -> tuple[dict, dict]:
|
|
"""(JSON-Schema property, uiSchema entry) for one template field."""
|
|
input_type = str(field.get("_input_type") or "")
|
|
prop: dict = {}
|
|
ui: dict = {}
|
|
title = field.get("display_name") or name
|
|
info = field.get("info") or ""
|
|
if title:
|
|
prop["title"] = title
|
|
if info:
|
|
prop["description"] = info
|
|
|
|
options = field.get("options") or []
|
|
if input_type in ("StrInput", "MessageTextInput"):
|
|
prop["type"] = "string"
|
|
elif input_type == "MultilineInput":
|
|
prop["type"] = "string"
|
|
ui["ui:widget"] = "textarea"
|
|
elif input_type == "SecretStrInput":
|
|
prop["type"] = "string"
|
|
ui["ui:widget"] = "password"
|
|
elif input_type == "IntInput":
|
|
prop["type"] = "integer"
|
|
elif input_type == "FloatInput":
|
|
prop["type"] = "number"
|
|
elif input_type == "BoolInput":
|
|
prop["type"] = "boolean"
|
|
elif input_type == "DropdownInput":
|
|
prop["type"] = "string"
|
|
if options:
|
|
prop["enum"] = list(options)
|
|
elif input_type == "MultiselectInput":
|
|
prop["type"] = "array"
|
|
items: dict = {"type": "string"}
|
|
if options:
|
|
items["enum"] = list(options)
|
|
prop["items"] = items
|
|
prop["uniqueItems"] = True
|
|
elif input_type == "SliderInput":
|
|
prop["type"] = "number"
|
|
spec = field.get("range_spec") or {}
|
|
if spec.get("min") is not None:
|
|
prop["minimum"] = spec["min"]
|
|
if spec.get("max") is not None:
|
|
prop["maximum"] = spec["max"]
|
|
if spec.get("step"):
|
|
prop["multipleOf"] = spec["step"]
|
|
ui["ui:widget"] = "range"
|
|
elif input_type == "FileInput":
|
|
# Stub for now: string reference only, real upload comes later.
|
|
prop["type"] = "string"
|
|
else:
|
|
prop["type"] = "string"
|
|
|
|
value = field.get("value")
|
|
if value is not None:
|
|
# A default outside its enum would fail validation on touch.
|
|
enum = prop.get("enum")
|
|
if not (enum is not None and value not in enum):
|
|
# bool is an int subclass; keep integer defaults clean.
|
|
if prop.get("type") == "integer" and isinstance(value, bool):
|
|
pass
|
|
else:
|
|
prop["default"] = value
|
|
return prop, ui
|
|
|
|
|
|
def extract_form_schema(langflow_json: dict) -> dict:
|
|
"""``{json_schema, ui_schema}``, or with ``fallback: "textarea"``."""
|
|
templates = _form_templates(langflow_json)
|
|
if not templates:
|
|
return {
|
|
"json_schema": {"type": "object", "properties": {}},
|
|
"ui_schema": {},
|
|
"fallback": "textarea",
|
|
}
|
|
template = templates[0].get("template") or {}
|
|
properties: dict = {}
|
|
required: list[str] = []
|
|
ui_schema: dict = {}
|
|
normal: list[str] = []
|
|
advanced: list[str] = []
|
|
for key, field in template.items():
|
|
if key == "code" or not isinstance(field, dict):
|
|
continue
|
|
if field.get("show") is False:
|
|
continue
|
|
name = field.get("name") or key
|
|
prop, ui = _field_schema(name, field)
|
|
properties[name] = prop
|
|
if field.get("required"):
|
|
required.append(name)
|
|
if ui:
|
|
ui_schema[name] = ui
|
|
(advanced if field.get("advanced") else normal).append(name)
|
|
if advanced:
|
|
ui_schema["ui:order"] = normal + advanced + ["*"]
|
|
return {
|
|
"json_schema": {
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required,
|
|
},
|
|
"ui_schema": ui_schema,
|
|
}
|