| """JSON Schema validator and compiler: schema -> deterministic fill plan. |
| |
| The mode of every leaf is decided purely by its JSON type and keywords (see SPEC.md §3); |
| there is no router and no model choice. The only author-controlled switch is |
| ``x-generate: true``, allowed on ``string`` and on ``array`` of ``string``. |
| |
| Compilation is a pure function of the schema: the same schema always yields the same plan, |
| and a leaf's :attr:`FieldPlan.fingerprint` depends only on that leaf (its key path, its own |
| keywords and annotations), never on its siblings or on property order elsewhere. |
| |
| Paths are tuples of property names, with :data:`ITEM` (``None``) standing for "each item of |
| the enclosing array". :attr:`FieldPlan.pointer` renders them as a JSON-Pointer-like string |
| with ``*`` for :data:`ITEM` (a literal ``*`` key is escaped as ``~2``). |
| |
| Invalid or unsupported schemas raise :class:`SchemaError` carrying every problem found as a |
| list of ``{"loc", "msg"}`` (HTTP 422 in the planned server). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import math |
| import re |
| from collections.abc import Mapping, Sequence |
| from dataclasses import dataclass, field, replace |
| from typing import Any, Literal |
|
|
| from jsonschema import exceptions as js_exceptions |
| from jsonschema import validators as js_validators |
|
|
| Mode = Literal[ |
| "decide.choice", |
| "decide.score", |
| "decide.noul", |
| "extract.span", |
| "extract.number", |
| "extract.date", |
| "generate", |
| ] |
| DECIDE_MODES: frozenset[str] = frozenset({"decide.choice", "decide.score", "decide.noul"}) |
| EXTRACT_MODES: frozenset[str] = frozenset({"extract.span", "extract.number", "extract.date"}) |
|
|
| GENERATE_KEY = "x-generate" |
| MAX_TOKENS_KEY = "x-max-tokens" |
| LEVELS_KEY = "x-levels" |
| EXTENSIONS = frozenset({GENERATE_KEY, MAX_TOKENS_KEY, LEVELS_KEY}) |
| ITEM = None |
|
|
| MIN_OPTIONS = 2 |
| MAX_OPTIONS = 255 |
| MAX_SCORE_LEVELS = 11 |
| |
|
|
| DATE_FORMATS: frozenset[str] = frozenset({"date", "date-time"}) |
| SPAN_FORMATS: frozenset[str] = frozenset({"email", "uri", "uuid"}) |
|
|
| ANNOTATIONS: frozenset[str] = frozenset( |
| { |
| "title", |
| "description", |
| "default", |
| "examples", |
| "$comment", |
| "$schema", |
| "$id", |
| "$anchor", |
| "$defs", |
| "definitions", |
| "readOnly", |
| "writeOnly", |
| "deprecated", |
| } |
| ) |
| STRING_KEYWORDS = frozenset({"minLength", "maxLength", "pattern", "format"}) |
| NUMBER_KEYWORDS = frozenset( |
| {"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf"} |
| ) |
| OBJECT_KEYWORDS = frozenset({"properties", "required", "additionalProperties"}) |
| ARRAY_KEYWORDS = frozenset({"items", "minItems", "maxItems", "uniqueItems"}) |
| TYPE_KEYWORDS: dict[str, frozenset[str]] = { |
| "string": STRING_KEYWORDS, |
| "integer": NUMBER_KEYWORDS, |
| "number": NUMBER_KEYWORDS, |
| "boolean": frozenset(), |
| "object": OBJECT_KEYWORDS, |
| "array": ARRAY_KEYWORDS, |
| } |
| COMMON_KEYWORDS = frozenset({"type", "enum", "$ref", "anyOf", "oneOf"}) | EXTENSIONS |
|
|
| |
| UNSUPPORTED: dict[str, str] = { |
| "allOf": "schema intersection is not supported; merge the subschemas", |
| "not": "negated schemas cannot be filled", |
| "if": "conditional schemas are not supported", |
| "then": "conditional schemas are not supported", |
| "else": "conditional schemas are not supported", |
| "const": "a single fixed value has nothing to decide; omit the field", |
| "dependentSchemas": "conditional schemas are not supported", |
| "dependentRequired": "conditional requirements are not supported", |
| "dependencies": "conditional requirements are not supported", |
| "patternProperties": "objects must list their properties explicitly", |
| "propertyNames": "objects must list their properties explicitly", |
| "unevaluatedProperties": "objects must list their properties explicitly", |
| "minProperties": "objects must list their properties explicitly", |
| "maxProperties": "objects must list their properties explicitly", |
| "unevaluatedItems": "arrays must have one homogeneous items schema", |
| "prefixItems": "tuple arrays are not supported; use an object", |
| "additionalItems": "tuple arrays are not supported; use an object", |
| "contains": "arrays must have one homogeneous items schema", |
| "minContains": "arrays must have one homogeneous items schema", |
| "maxContains": "arrays must have one homogeneous items schema", |
| "$dynamicRef": "only local $ref ('#/...') is supported", |
| "$dynamicAnchor": "only local $ref ('#/...') is supported", |
| "$recursiveRef": "only local $ref ('#/...') is supported", |
| "$recursiveAnchor": "only local $ref ('#/...') is supported", |
| "contentEncoding": "encoded content is not supported", |
| "contentMediaType": "encoded content is not supported", |
| "contentSchema": "encoded content is not supported", |
| } |
|
|
|
|
| class SchemaError(ValueError): |
| """Invalid or unsupported schema; ``errors`` is a list of ``{loc, msg}``.""" |
|
|
| def __init__(self, errors: list[dict[str, Any]], status: int = 422) -> None: |
| super().__init__("; ".join(f"{'/'.join(map(str, e['loc']))}: {e['msg']}" for e in errors)) |
| self.errors = errors |
| self.status = status |
|
|
| def body(self) -> dict[str, Any]: |
| return {"error": {"type": "invalid_schema", "status": self.status, "detail": self.errors}} |
|
|
|
|
| Path = tuple[str | None, ...] |
|
|
|
|
| def pointer(path: Path) -> str: |
| """Render a plan path; ``ITEM`` -> ``*``; ``~``, ``/``, ``*`` escaped as ~0, ~1, ~2.""" |
| parts = [] |
| for part in path: |
| if part is ITEM: |
| parts.append("*") |
| else: |
| parts.append(part.replace("~", "~0").replace("/", "~1").replace("*", "~2")) |
| return "/" + "/".join(parts) if parts else "" |
|
|
|
|
| def canonical_json(value: Any) -> str: |
| return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
|
|
|
|
| def _digest(value: Any) -> str: |
| return hashlib.sha256(canonical_json(value).encode()).hexdigest() |
|
|
|
|
| @dataclass(frozen=True, eq=True) |
| class FieldPlan: |
| """One leaf to fill. ``constraints`` holds the JSON Schema keywords the value must meet. |
| |
| * ``options``: ``decide.choice`` outcomes (JSON values, schema order); with |
| ``option_descriptions`` from ``oneOf``/``anyOf`` const branches when present. |
| * ``levels``: ``decide.score`` outcomes (the admissible integers, at most 11). |
| * ``level_labels``: per-level label text from ``x-levels`` (aligned with ``levels``). |
| * ``null_outcome``: decide heads get an extra "null" outcome (nullable score fields). |
| * ``max_new_tokens``: a generate field's own decode budget (``x-max-tokens``). |
| * ``on_null``: what an extract field does when nothing is found: ``"null"`` (nullable), |
| ``"omit"`` (optional, non-nullable) or ``"forbid"`` (required, non-nullable: the |
| extractor must return its best span). |
| """ |
|
|
| path: Path |
| mode: Mode |
| json_type: str |
| required: bool |
| nullable: bool |
| title: str | None = None |
| description: str | None = None |
| options: tuple[Any, ...] | None = None |
| option_descriptions: tuple[str | None, ...] | None = None |
| levels: tuple[int, ...] | None = None |
| level_labels: tuple[str | None, ...] | None = None |
| null_outcome: bool = False |
| max_new_tokens: int | None = None |
| format: str | None = None |
| constraints: dict[str, Any] = field(default_factory=dict) |
|
|
| @property |
| def pointer(self) -> str: |
| return pointer(self.path) |
|
|
| @property |
| def on_null(self) -> Literal["null", "omit", "forbid"]: |
| if self.nullable: |
| return "null" |
| return "forbid" if self.required else "omit" |
|
|
| @property |
| def phase(self) -> Literal["parallel", "generate"]: |
| return "generate" if self.mode == "generate" else "parallel" |
|
|
| def to_dict(self) -> dict[str, Any]: |
| out: dict[str, Any] = { |
| "pointer": self.pointer, |
| "path": list(self.path), |
| "mode": self.mode, |
| "type": self.json_type, |
| "required": self.required, |
| "nullable": self.nullable, |
| "on_null": self.on_null, |
| } |
| for key in ("title", "description", "format"): |
| if getattr(self, key) is not None: |
| out[key] = getattr(self, key) |
| if self.options is not None: |
| out["options"] = list(self.options) |
| if self.option_descriptions is not None: |
| out["option_descriptions"] = list(self.option_descriptions) |
| if self.levels is not None: |
| out["levels"] = list(self.levels) |
| if self.level_labels is not None: |
| out["level_labels"] = list(self.level_labels) |
| if self.max_new_tokens is not None: |
| out["max_new_tokens"] = self.max_new_tokens |
| if self.null_outcome: |
| out["null_outcome"] = True |
| if self.constraints: |
| out["constraints"] = dict(self.constraints) |
| return out |
|
|
| @property |
| def fingerprint(self) -> str: |
| """sha256 of the leaf's canonical spec: identical wherever this leaf appears.""" |
| return _digest(self.to_dict()) |
|
|
|
|
| @dataclass(frozen=True, eq=True) |
| class ArrayPlan: |
| """An array node. Its length is a runtime count bounded by min/max items.""" |
|
|
| path: Path |
| min_items: int |
| max_items: int | None |
| unique_items: bool |
| required: bool |
|
|
| @property |
| def pointer(self) -> str: |
| return pointer(self.path) |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return { |
| "pointer": self.pointer, |
| "path": list(self.path), |
| "min_items": self.min_items, |
| "max_items": self.max_items, |
| "unique_items": self.unique_items, |
| "required": self.required, |
| } |
|
|
|
|
| @dataclass(frozen=True) |
| class LeafNode: |
| field: FieldPlan |
|
|
|
|
| @dataclass(frozen=True) |
| class ArrayNode: |
| array: ArrayPlan |
| item: Node |
|
|
|
|
| @dataclass(frozen=True) |
| class ObjectNode: |
| path: Path |
| properties: tuple[tuple[str, Node], ...] |
|
|
|
|
| Node = LeafNode | ArrayNode | ObjectNode |
|
|
|
|
| @dataclass(frozen=True) |
| class FillPlan: |
| """Compiled schema. ``fields`` are leaves in schema order (depth-first).""" |
|
|
| root: ObjectNode |
| fields: tuple[FieldPlan, ...] |
| arrays: tuple[ArrayPlan, ...] |
| schema: Mapping[str, Any] |
|
|
| @property |
| def parallel_fields(self) -> tuple[FieldPlan, ...]: |
| """decide.* and extract.*: independent of each other and of order.""" |
| return tuple(f for f in self.fields if f.phase == "parallel") |
|
|
| @property |
| def generate_fields(self) -> tuple[FieldPlan, ...]: |
| """generate: run after the parallel phase, in this (schema) order.""" |
| return tuple(f for f in self.fields if f.phase == "generate") |
|
|
| def field(self, ptr: str) -> FieldPlan: |
| for f in self.fields: |
| if f.pointer == ptr: |
| return f |
| raise KeyError(ptr) |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return { |
| "fields": [f.to_dict() for f in self.fields], |
| "arrays": [a.to_dict() for a in self.arrays], |
| "fingerprint": self.fingerprint, |
| } |
|
|
| @property |
| def fingerprint(self) -> str: |
| """Order-insensitive over the parallel phase, order-sensitive over generate fields. |
| |
| Reordering decide/extract properties keeps the fingerprint; reordering generate |
| fields changes it, because each generated field conditions on the ones before it. |
| """ |
| return _digest( |
| { |
| "parallel": sorted(f.fingerprint for f in self.parallel_fields), |
| "generate": [f.fingerprint for f in self.generate_fields], |
| "arrays": sorted(_digest(a.to_dict()) for a in self.arrays), |
| } |
| ) |
|
|
|
|
| |
|
|
|
|
| def _is_number(value: Any) -> bool: |
| return isinstance(value, (int, float)) and not isinstance(value, bool) |
|
|
|
|
| def _type_matches(value: Any, json_type: str) -> bool: |
| if json_type == "null": |
| return value is None |
| if json_type == "boolean": |
| return isinstance(value, bool) |
| if json_type == "integer": |
| return _is_number(value) and float(value).is_integer() |
| if json_type == "number": |
| return _is_number(value) and math.isfinite(value) |
| if json_type == "string": |
| return isinstance(value, str) |
| return False |
|
|
|
|
| class _Compiler: |
| def __init__(self, root: Mapping[str, Any]) -> None: |
| self.root = root |
| self.errors: list[dict[str, Any]] = [] |
| self.fields: list[FieldPlan] = [] |
| self.arrays: list[ArrayPlan] = [] |
|
|
| def error(self, loc: Sequence[Any], msg: str) -> None: |
| self.errors.append({"loc": list(loc), "msg": msg}) |
|
|
| |
|
|
| def resolve( |
| self, schema: Any, loc: list[Any], refs: tuple[str, ...] |
| ) -> tuple[dict[str, Any] | None, tuple[str, ...]]: |
| """Follow local ``$ref`` (annotation siblings override the target's).""" |
| while isinstance(schema, dict) and "$ref" in schema: |
| ref = schema["$ref"] |
| if not isinstance(ref, str) or not (ref == "#" or ref.startswith("#/")): |
| self.error([*loc, "$ref"], f"only local $ref ('#/...') is supported, got {ref!r}") |
| return None, refs |
| if ref in refs: |
| self.error([*loc, "$ref"], f"recursive $ref {ref!r} is not supported") |
| return None, refs |
| extra = set(schema) - {"$ref"} - ANNOTATIONS - EXTENSIONS |
| if extra: |
| self.error(loc, f"$ref siblings must be annotations, got {sorted(extra)}") |
| return None, refs |
| target: Any = self.root |
| for raw in ref[2:].split("/") if ref != "#" else []: |
| part = raw.replace("~1", "/").replace("~0", "~") |
| if isinstance(target, dict) and part in target: |
| target = target[part] |
| elif isinstance(target, list) and part.isdigit() and int(part) < len(target): |
| target = target[int(part)] |
| else: |
| self.error([*loc, "$ref"], f"unresolvable $ref {ref!r}") |
| return None, refs |
| if not isinstance(target, dict): |
| self.error([*loc, "$ref"], f"$ref {ref!r} must point at a schema object") |
| return None, refs |
| siblings = {k: v for k, v in schema.items() if k != "$ref"} |
| schema = {**target, **siblings} |
| refs = (*refs, ref) |
| if not isinstance(schema, dict): |
| self.error(loc, "schema must be an object (boolean schemas are not supported)") |
| return None, refs |
| return schema, refs |
|
|
| def flatten_union( |
| self, schema: dict[str, Any], loc: list[Any], refs: tuple[str, ...] |
| ) -> tuple[dict[str, Any] | None, tuple[str, ...], list[str | None] | None]: |
| """anyOf/oneOf allowed only as (a) nullable ``[X, {type: null}]`` or (b) an enum |
| written as const/enum branches (branch descriptions become option descriptions).""" |
| key = "anyOf" if "anyOf" in schema else "oneOf" if "oneOf" in schema else None |
| if key is None: |
| return schema, refs, None |
| if "anyOf" in schema and "oneOf" in schema: |
| self.error(loc, "anyOf and oneOf together are not supported") |
| return None, refs, None |
| extra = set(schema) - {key} - ANNOTATIONS - {"type"} |
| if extra: |
| self.error(loc, f"{key} siblings must be annotations, got {sorted(extra)}") |
| return None, refs, None |
| raw_branches = schema[key] |
| if not isinstance(raw_branches, list) or not raw_branches: |
| self.error([*loc, key], f"{key} must be a non-empty array") |
| return None, refs, None |
| branches = [] |
| for index, raw in enumerate(raw_branches): |
| branch, _ = self.resolve(raw, [*loc, key, index], refs) |
| if branch is None: |
| return None, refs, None |
| branches.append(branch) |
| annotations = {k: v for k, v in schema.items() if k in ANNOTATIONS} |
|
|
| def is_null(b: dict[str, Any]) -> bool: |
| return b.get("type") == "null" and not (set(b) - ANNOTATIONS - {"type"}) |
|
|
| nulls = [b for b in branches if is_null(b)] |
| others = [b for b in branches if not is_null(b)] |
| if len(nulls) == 1 and len(others) == 1: |
| other = others[0] |
| if any(k in other for k in ("anyOf", "oneOf")): |
| self.error([*loc, key], f"nested {key} is not supported") |
| return None, refs, None |
| merged = {**other, **annotations} |
| base = other.get("type") |
| if isinstance(base, str): |
| merged["type"] = [base, "null"] |
| elif "enum" in other and isinstance(other["enum"], list): |
| merged["enum"] = [*other["enum"], *([] if None in other["enum"] else [None])] |
| else: |
| self.error([*loc, key], "nullable branch must declare a single type or enum") |
| return None, refs, None |
| return merged, refs, None |
| options: list[Any] = [] |
| descriptions: list[str | None] = [] |
| for index, branch in enumerate(branches): |
| values = None |
| if "const" in branch and not (set(branch) - ANNOTATIONS - {"const", "type"}): |
| values = [branch["const"]] |
| elif "enum" in branch and not (set(branch) - ANNOTATIONS - {"enum", "type"}): |
| values = branch["enum"] if isinstance(branch["enum"], list) else None |
| elif is_null(branch): |
| values = [None] |
| if values is None: |
| self.error( |
| [*loc, key, index], |
| f"{key} is supported only for nullable [X, {{type: null}}] or for " |
| "branches that are all const/enum (an enum with descriptions)", |
| ) |
| return None, refs, None |
| desc = branch.get("description", branch.get("title")) |
| for value in values: |
| options.append(value) |
| descriptions.append(desc if isinstance(desc, str) else None) |
| merged = {**annotations, "enum": options} |
| if "type" in schema: |
| merged["type"] = schema["type"] |
| if all(d is None for d in descriptions): |
| return merged, refs, None |
| return merged, refs, descriptions |
|
|
| |
|
|
| def node( |
| self, |
| raw: Any, |
| path: Path, |
| loc: list[Any], |
| *, |
| required: bool, |
| generate: bool, |
| refs: tuple[str, ...] = (), |
| max_tokens: int | None = None, |
| ) -> Node | None: |
| schema, refs = self.resolve(raw, loc, refs) |
| if schema is None: |
| return None |
| schema, refs, option_descriptions = self.flatten_union(schema, loc, refs) |
| if schema is None: |
| return None |
| before = len(self.errors) |
| for key in schema: |
| if key in UNSUPPORTED: |
| self.error([*loc, key], f"unsupported keyword {key!r}: {UNSUPPORTED[key]}") |
| elif key.startswith("x-") and key not in EXTENSIONS: |
| self.error([*loc, key], f"unknown extension {key!r} (known: " |
| f"{', '.join(sorted(EXTENSIONS))})") |
| if len(self.errors) > before: |
| return None |
|
|
| flag = schema.get(GENERATE_KEY, False) |
| if not isinstance(flag, bool): |
| self.error([*loc, GENERATE_KEY], "x-generate must be a boolean") |
| return None |
| generate = generate or flag |
| own_budget = schema.get(MAX_TOKENS_KEY) |
| if own_budget is not None: |
| if not generate: |
| self.error([*loc, MAX_TOKENS_KEY], "x-max-tokens applies only to x-generate " |
| "fields") |
| return None |
| if not isinstance(own_budget, int) or isinstance(own_budget, bool) \ |
| or own_budget < 1: |
| self.error([*loc, MAX_TOKENS_KEY], "x-max-tokens must be an integer >= 1") |
| return None |
| max_tokens = own_budget |
|
|
| json_type, nullable = self.parse_type(schema, loc) |
| if json_type is None: |
| return None |
| allowed = COMMON_KEYWORDS | ANNOTATIONS | TYPE_KEYWORDS.get(json_type, frozenset()) |
| unknown = sorted(set(schema) - allowed) |
| for key in unknown: |
| known = any(key in kws for kws in TYPE_KEYWORDS.values()) |
| msg = ( |
| f"keyword {key!r} does not apply to type {json_type!r}" |
| if known |
| else f"unknown keyword {key!r}" |
| ) |
| self.error([*loc, key], msg) |
| if unknown: |
| return None |
|
|
| if generate and not (json_type == "string" or json_type == "array"): |
| self.error( |
| [*loc, GENERATE_KEY], |
| f"x-generate is allowed only on string or array of string, not {json_type!r}" |
| + (" (enum fields are decided)" if "enum" in schema else ""), |
| ) |
| return None |
|
|
| if json_type in ("object", "array") and nullable: |
| self.error([*loc, "type"], "nullable objects/arrays are not supported") |
| return None |
| if json_type == "object": |
| if "enum" in schema: |
| self.error([*loc, "enum"], "enum of objects is not supported") |
| return None |
| return self.object_node(schema, path, loc, refs) |
| if json_type == "array": |
| if "enum" in schema: |
| self.error([*loc, "enum"], "enum of arrays is not supported") |
| return None |
| return self.array_node(schema, path, loc, required=required, generate=generate, |
| refs=refs, max_tokens=max_tokens) |
| leaf = self.leaf(schema, json_type, nullable, path, loc, required=required, |
| generate=generate, option_descriptions=option_descriptions) |
| if leaf is None: |
| return None |
| if max_tokens is not None: |
| leaf = replace(leaf, max_new_tokens=max_tokens) |
| if LEVELS_KEY in schema: |
| leaf = self.level_labels(leaf, schema[LEVELS_KEY], [*loc, LEVELS_KEY]) |
| if leaf is None: |
| return None |
| self.fields.append(leaf) |
| return LeafNode(leaf) |
|
|
| def parse_type(self, schema: dict[str, Any], loc: list[Any]) -> tuple[str | None, bool]: |
| raw = schema.get("type") |
| if raw is None: |
| enum = schema.get("enum") |
| if isinstance(enum, list) and enum: |
| kinds = {_enum_kind(v) for v in enum if v is not None} |
| if len(kinds) == 1 and None not in kinds: |
| return kinds.pop(), None in enum |
| if not kinds: |
| self.error([*loc, "enum"], "enum must contain a non-null value") |
| return None, False |
| return "enum", None in enum |
| self.error([*loc, "type"], "type is required (or an enum)") |
| return None, False |
| types = [raw] if isinstance(raw, str) else raw |
| if not isinstance(types, list) or not all(isinstance(t, str) for t in types): |
| self.error([*loc, "type"], "type must be a string or array of strings") |
| return None, False |
| nullable = "null" in types |
| rest = [t for t in types if t != "null"] |
| if len(rest) != 1 or len(types) != len(set(types)): |
| self.error( |
| [*loc, "type"], |
| f"union types are not supported (only [X, 'null']), got {types!r}", |
| ) |
| return None, False |
| if rest[0] not in TYPE_KEYWORDS: |
| self.error([*loc, "type"], f"unknown type {rest[0]!r}") |
| return None, False |
| return rest[0], nullable |
|
|
| def object_node( |
| self, schema: dict[str, Any], path: Path, loc: list[Any], refs: tuple[str, ...] |
| ) -> ObjectNode | None: |
| props = schema.get("properties", {}) |
| if not isinstance(props, dict): |
| self.error([*loc, "properties"], "properties must be an object") |
| return None |
| required = schema.get("required", []) |
| if not isinstance(required, list) or not all(isinstance(r, str) for r in required): |
| self.error([*loc, "required"], "required must be an array of strings") |
| return None |
| if len(set(required)) != len(required): |
| self.error([*loc, "required"], "required must not repeat names") |
| for name in required: |
| if name not in props: |
| self.error([*loc, "required"], f"required property {name!r} is not in properties") |
| extra = schema.get("additionalProperties", False) |
| if not isinstance(extra, bool): |
| self.error( |
| [*loc, "additionalProperties"], |
| "additionalProperties must be true or false (a schema cannot be filled: the " |
| "property names are unknown); deg never emits undeclared properties", |
| ) |
| children: list[tuple[str, Node]] = [] |
| for name, sub in props.items(): |
| child = self.node( |
| sub, (*path, name), [*loc, "properties", name], |
| required=name in required, generate=False, refs=refs, |
| ) |
| if child is not None: |
| children.append((name, child)) |
| return ObjectNode(path, tuple(children)) |
|
|
| def array_node( |
| self, |
| schema: dict[str, Any], |
| path: Path, |
| loc: list[Any], |
| *, |
| required: bool, |
| generate: bool, |
| refs: tuple[str, ...], |
| max_tokens: int | None = None, |
| ) -> ArrayNode | None: |
| if "items" not in schema: |
| self.error([*loc, "items"], "arrays need an items schema") |
| return None |
| if isinstance(schema["items"], list): |
| self.error([*loc, "items"], "tuple arrays (items: [...]) are not supported") |
| return None |
| min_items = schema.get("minItems", 0) |
| max_items = schema.get("maxItems") |
| ok = True |
| for key, value in (("minItems", min_items), ("maxItems", max_items)): |
| if value is not None and ( |
| not isinstance(value, int) or isinstance(value, bool) or value < 0 |
| ): |
| self.error([*loc, key], f"{key} must be a non-negative integer") |
| ok = False |
| if ok and max_items is not None and max_items < min_items: |
| self.error([*loc, "maxItems"], "maxItems must be >= minItems") |
| ok = False |
| unique = schema.get("uniqueItems", False) |
| if not isinstance(unique, bool): |
| self.error([*loc, "uniqueItems"], "uniqueItems must be a boolean") |
| ok = False |
| if not ok: |
| return None |
| item_path = (*path, ITEM) |
| if generate: |
| item_schema, _ = self.resolve(schema["items"], [*loc, "items"], refs) |
| if item_schema is None: |
| return None |
| item_type = item_schema.get("type") |
| if item_type != "string" or "enum" in item_schema or "format" in item_schema: |
| self.error( |
| [*loc, GENERATE_KEY], |
| "x-generate is allowed only on string or array of string " |
| "(items must be plain type: string)", |
| ) |
| return None |
| array = ArrayPlan(path, min_items, max_items, unique, required) |
| self.arrays.append(array) |
| item = self.node( |
| schema["items"], item_path, [*loc, "items"], |
| required=True, generate=generate, refs=refs, max_tokens=max_tokens, |
| ) |
| if item is None: |
| return None |
| return ArrayNode(array, item) |
|
|
| def level_labels(self, leaf: FieldPlan, raw: Any, loc: list[Any]) -> FieldPlan | None: |
| """``x-levels``: {"<level>": "label"} for a decide.score field. Each label becomes |
| that level's option line (``<level>: <label>``), the layout the score head was |
| trained on (jev levels); unlabeled levels keep the bare number.""" |
| if leaf.mode != "decide.score" or leaf.levels is None: |
| self.error(loc, "x-levels applies only to integer score fields (bounded integers " |
| f"with <= {MAX_SCORE_LEVELS} values)") |
| return None |
| if not isinstance(raw, dict) or not raw: |
| self.error(loc, "x-levels must be a non-empty object {\"<level>\": \"label\"}") |
| return None |
| labels: dict[int, str] = {} |
| for key, label in raw.items(): |
| try: |
| level = int(key) |
| except ValueError: |
| level = None |
| if level is None or str(level) != key or level not in leaf.levels: |
| self.error([*loc, key], f"{key!r} is not one of the levels {list(leaf.levels)}") |
| return None |
| if not isinstance(label, str) or not label.strip(): |
| self.error([*loc, key], "a level label must be a non-empty string") |
| return None |
| labels[level] = label |
| return replace(leaf, level_labels=tuple(labels.get(v) for v in leaf.levels)) |
|
|
| def leaf( |
| self, |
| schema: dict[str, Any], |
| json_type: str, |
| nullable: bool, |
| path: Path, |
| loc: list[Any], |
| *, |
| required: bool, |
| generate: bool, |
| option_descriptions: list[str | None] | None, |
| ) -> FieldPlan | None: |
| notes = {} |
| for key in ("title", "description"): |
| value = schema.get(key) |
| if value is not None and not isinstance(value, str): |
| self.error([*loc, key], f"{key} must be a string") |
| return None |
| notes[key] = value |
| base = dict(path=path, json_type=json_type, required=required, nullable=nullable, |
| **notes) |
|
|
| if "enum" in schema: |
| if generate: |
| self.error([*loc, GENERATE_KEY], |
| "x-generate is allowed only on string or array of string " |
| "(enum fields are decided)") |
| return None |
| return self.choice(schema, base, loc, json_type, option_descriptions) |
|
|
| if json_type == "boolean": |
| if nullable: |
| return FieldPlan(mode="decide.choice", options=(True, False, None), **base) |
| return FieldPlan(mode="decide.noul", **base) |
|
|
| if json_type in ("integer", "number"): |
| return self.numeric(schema, base, loc, json_type, nullable) |
|
|
| |
| constraints = self.string_constraints(schema, loc) |
| if constraints is None: |
| return None |
| fmt = schema.get("format") |
| if fmt is not None: |
| if generate: |
| self.error([*loc, GENERATE_KEY], |
| "x-generate cannot be combined with format (formatted values are " |
| "extracted from the context)") |
| return None |
| if fmt in DATE_FORMATS: |
| return FieldPlan(mode="extract.date", format=fmt, constraints=constraints, **base) |
| if fmt in SPAN_FORMATS: |
| return FieldPlan(mode="extract.span", format=fmt, constraints=constraints, **base) |
| self.error([*loc, "format"], |
| f"unsupported format {fmt!r}; supported: " |
| f"{sorted(DATE_FORMATS | SPAN_FORMATS)}") |
| return None |
| mode: Mode = "generate" if generate else "extract.span" |
| return FieldPlan(mode=mode, constraints=constraints, **base) |
|
|
| def choice( |
| self, |
| schema: dict[str, Any], |
| base: dict[str, Any], |
| loc: list[Any], |
| json_type: str, |
| option_descriptions: list[str | None] | None, |
| ) -> FieldPlan | None: |
| enum = schema["enum"] |
| if not isinstance(enum, list): |
| self.error([*loc, "enum"], "enum must be an array") |
| return None |
| seen: list[str] = [] |
| for index, value in enumerate(enum): |
| key = canonical_json(value) |
| if key in seen: |
| self.error([*loc, "enum", index], f"duplicate enum value {value!r}") |
| return None |
| seen.append(key) |
| if isinstance(value, (dict, list)): |
| self.error([*loc, "enum", index], "enum values must be scalars") |
| return None |
| if value is None: |
| if not base["nullable"] and "type" in schema: |
| self.error([*loc, "enum", index], "null in enum needs a nullable type") |
| return None |
| continue |
| if json_type != "enum" and not _type_matches(value, json_type): |
| self.error([*loc, "enum", index], |
| f"enum value {value!r} does not match type {json_type!r}") |
| return None |
| if not MIN_OPTIONS <= len(enum) <= MAX_OPTIONS: |
| self.error([*loc, "enum"], |
| f"enum needs {MIN_OPTIONS}..{MAX_OPTIONS} values " |
| "(a single value has nothing to decide)") |
| return None |
| extra = sorted(set(schema) & (STRING_KEYWORDS | NUMBER_KEYWORDS)) |
| if extra: |
| self.error([*loc, extra[0]], f"{extra} cannot be combined with enum") |
| return None |
| nullable = None in enum |
| base = {**base, "nullable": nullable} |
| return FieldPlan( |
| mode="decide.choice", |
| options=tuple(enum), |
| option_descriptions=tuple(option_descriptions) if option_descriptions else None, |
| **base, |
| ) |
|
|
| def numeric( |
| self, |
| schema: dict[str, Any], |
| base: dict[str, Any], |
| loc: list[Any], |
| json_type: str, |
| nullable: bool, |
| ) -> FieldPlan | None: |
| constraints: dict[str, Any] = {} |
| for key in NUMBER_KEYWORDS: |
| if key in schema: |
| value = schema[key] |
| if not _is_number(value) or not math.isfinite(value): |
| self.error([*loc, key], f"{key} must be a finite number") |
| return None |
| constraints[key] = value |
| if "multipleOf" in constraints and constraints["multipleOf"] <= 0: |
| self.error([*loc, "multipleOf"], "multipleOf must be > 0") |
| return None |
| lo_in, hi_in = constraints.get("minimum"), constraints.get("maximum") |
| lo_ex, hi_ex = constraints.get("exclusiveMinimum"), constraints.get("exclusiveMaximum") |
| if json_type == "integer": |
| lows = [math.ceil(lo_in)] if lo_in is not None else [] |
| lows += [math.floor(lo_ex) + 1] if lo_ex is not None else [] |
| highs = [math.floor(hi_in)] if hi_in is not None else [] |
| highs += [math.ceil(hi_ex) - 1] if hi_ex is not None else [] |
| if lows and highs: |
| lo, hi = max(lows), min(highs) |
| step = constraints.get("multipleOf") |
| |
| if (hi - lo) / (step or 1) > MAX_SCORE_LEVELS * 2: |
| return FieldPlan(mode="extract.number", constraints=constraints, **base) |
| levels = [v for v in range(lo, hi + 1) |
| if step is None or _multiple(v, step)] |
| if not levels: |
| self.error(loc, "integer bounds admit no value") |
| return None |
| if len(levels) == 1: |
| self.error(loc, "integer bounds admit a single value; nothing to decide") |
| return None |
| if len(levels) <= MAX_SCORE_LEVELS: |
| return FieldPlan(mode="decide.score", levels=tuple(levels), |
| null_outcome=nullable, constraints=constraints, **base) |
| return FieldPlan(mode="extract.number", constraints=constraints, **base) |
| |
| lows = [v for v in (lo_in, lo_ex) if v is not None] |
| highs = [v for v in (hi_in, hi_ex) if v is not None] |
| if lows and highs: |
| lo, hi = max(lows), min(highs) |
| empty = lo > hi or (lo == hi and (lo_ex is not None and lo_ex >= hi |
| or hi_ex is not None and hi_ex <= lo)) |
| if empty: |
| self.error(loc, "number bounds admit no value") |
| return None |
| return FieldPlan(mode="extract.number", constraints=constraints, **base) |
|
|
| def string_constraints(self, schema: dict[str, Any], loc: list[Any]) -> dict[str, Any] | None: |
| constraints: dict[str, Any] = {} |
| for key in ("minLength", "maxLength"): |
| if key in schema: |
| value = schema[key] |
| if not isinstance(value, int) or isinstance(value, bool) or value < 0: |
| self.error([*loc, key], f"{key} must be a non-negative integer") |
| return None |
| constraints[key] = value |
| if constraints.get("maxLength", math.inf) < constraints.get("minLength", 0): |
| self.error([*loc, "maxLength"], "maxLength must be >= minLength") |
| return None |
| if "pattern" in schema: |
| if not isinstance(schema["pattern"], str): |
| self.error([*loc, "pattern"], "pattern must be a string") |
| return None |
| try: |
| re.compile(schema["pattern"]) |
| except re.error as exc: |
| self.error([*loc, "pattern"], f"invalid pattern: {exc}") |
| return None |
| constraints["pattern"] = schema["pattern"] |
| if "format" in schema and not isinstance(schema["format"], str): |
| self.error([*loc, "format"], "format must be a string") |
| return None |
| return constraints |
|
|
|
|
| def _enum_kind(value: Any) -> str: |
| if isinstance(value, bool): |
| return "boolean" |
| if isinstance(value, int): |
| return "integer" |
| if isinstance(value, float): |
| return "number" |
| if isinstance(value, str): |
| return "string" |
| return "enum" |
|
|
|
|
| def _multiple(value: float, step: float) -> bool: |
| quotient = value / step |
| return math.isclose(quotient, round(quotient), rel_tol=0.0, abs_tol=1e-9) |
|
|
|
|
| def check_schema(schema: Any) -> None: |
| """Raise :class:`SchemaError` unless ``schema`` is a valid JSON Schema document.""" |
| if not isinstance(schema, dict): |
| raise SchemaError([{"loc": ["schema"], "msg": "schema must be a JSON object"}]) |
| cls = js_validators.validator_for(schema, default=js_validators.Draft202012Validator) |
| try: |
| cls.check_schema(schema) |
| except js_exceptions.SchemaError as exc: |
| loc = ["schema", *exc.path] |
| raise SchemaError([{"loc": loc, "msg": f"invalid JSON Schema: {exc.message}"}]) from exc |
|
|
|
|
| def compile_schema(schema: Any) -> FillPlan: |
| """Validate ``schema`` and compile it into a :class:`FillPlan`. |
| |
| The root must be an object. Raises :class:`SchemaError` listing every problem found. |
| """ |
| check_schema(schema) |
| compiler = _Compiler(schema) |
| root = compiler.node(schema, (), ["schema"], required=True, generate=False) |
| if compiler.errors: |
| raise SchemaError(compiler.errors) |
| if not isinstance(root, ObjectNode): |
| raise SchemaError([{"loc": ["schema", "type"], "msg": "the root schema must be " |
| "type: object"}]) |
| return FillPlan(root, tuple(compiler.fields), tuple(compiler.arrays), schema) |
|
|
