74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
"""Full JSON Schema validation for workflow input/output values.
|
||
|
||
The engine used to check only ``required`` keys; everything else (wrong type,
|
||
unknown enum value, nested ``properties``/``items`` violations) slipped through
|
||
until downstream nodes exploded with confusing errors. This module runs the
|
||
graph's ``inputSchema`` / ``outputSchema`` through ``Draft202012Validator`` and
|
||
returns structured, frontend-friendly errors (path + message) instead of
|
||
leaking Python tracebacks.
|
||
|
||
Kept as pure functions so both the node executors (``WorkflowError``) and the
|
||
HTTP router (400 responses) can share the exact same rules.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from jsonschema import Draft202012Validator
|
||
|
||
# Cap the reported errors so a wildly invalid payload cannot produce a
|
||
# multi-megabyte error body.
|
||
_MAX_ERRORS = 20
|
||
|
||
|
||
def schema_issues(value: Any, schema: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
"""Validate ``value`` against ``schema``.
|
||
|
||
Returns ``[]`` when the schema is empty/absent (nothing declared = nothing
|
||
enforced). Each issue is ``{"path": [...], "message": str}`` with ``path``
|
||
pointing at the offending field, e.g. ``["options", 0, "name"]``.
|
||
"""
|
||
if not isinstance(schema, dict) or not schema:
|
||
return []
|
||
try:
|
||
validator = Draft202012Validator(schema)
|
||
# Construction alone does not check the schema — malformed keywords
|
||
# (e.g. a non-string ``pattern``) only explode during validation.
|
||
errors = sorted(validator.iter_errors(value), key=lambda err: list(err.absolute_path))
|
||
except Exception: # noqa: BLE001 - a malformed user schema must not 500
|
||
return [{"path": [], "message": "schema 定义本身无效，无法用于校验"}]
|
||
out: list[dict[str, Any]] = []
|
||
seen: set[tuple[Any, ...]] = set()
|
||
for err in errors:
|
||
path = [str(p) if not isinstance(p, int) else p for p in err.absolute_path]
|
||
key = tuple(path)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({"path": path, "message": _humanize(err)})
|
||
if len(out) >= _MAX_ERRORS:
|
||
out.append({"path": [], "message": f"……其余问题已省略（共 {len(errors)} 处）"})
|
||
break
|
||
return out
|
||
|
||
|
||
def _humanize(err: Any) -> str:
|
||
"""Map the common jsonschema phrasings to short Chinese messages."""
|
||
message = str(err.message)
|
||
validator = getattr(err, "validator", "")
|
||
path = ".".join(str(p) for p in err.absolute_path) or "输入"
|
||
if validator == "required":
|
||
missing = message.split("'")[1] if "'" in message else "?"
|
||
return f"{path} 缺少必填字段：{missing}"
|
||
if validator == "enum":
|
||
return f"{path} 的值不在允许范围内：{message}"
|
||
if validator == "type":
|
||
return f"{path} 类型不正确：{message}"
|
||
if validator == "additionalProperties":
|
||
return f"{path} 含有未声明的额外字段：{message}"
|
||
return f"{path} 不符合 schema：{message}"
|
||
|
||
|
||
__all__ = ["schema_issues"]
|