kirche-einfach/backend/app/flows/input_schema.py

152 lines
5.4 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:
# Whitespace-only string values are the flow generator's sentinel
# for "required, filled at runtime" (see make_flow_fixtures.py) —
# not a real default for the web form.
if isinstance(value, str) and not value.strip():
return prop, ui
# 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,
}