Post-process flatc jsonschema output into rich schema

score/flatbuffers/bazel/BUILD

@@ -22,3 +22,12 @@ py_binary(
     tags = ["manual"],
     visibility = ["//visibility:public"],
 )
+
+py_binary(
+    name = "schema_generator",
+    srcs = ["schema_generator.py"],
+    # "manual" excludes this target from wildcard builds so it is only built for the exec
+    # (host) platform when referenced as a tool via cfg = "exec" in the starlark rules.
+    tags = ["manual"],
+    visibility = ["//visibility:public"],
+)

score/flatbuffers/bazel/schema_generator.py

@@ -0,0 +1,274 @@
+# *******************************************************************************
+# Copyright (c) 2026 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+"""Post-process flatc --jsonschema output into a rich JSON Schema.
+
+This tool reads the raw JSON schema output from flatc --jsonschema and post-processes it
+into a rich JSON schema in draft-2020-12 format with extracted metadata.
+
+The Bazel rule handles flatc invocation; this script only does post-processing.
+
+Input: Raw JSON schema file from ``flatc --jsonschema`` (draft 2019-09 format with definitions + $ref)
+Output: Rich JSON schema in draft-2020-12 format
+
+Post-processing steps:
+  * split ``@title:`` / ``@default:`` / ``@min:`` / ``@max:`` / ``@required`` token lines
+    out of each ``description`` into proper JSON-schema attributes;
+  * strip flatc's type-based ``minimum`` / ``maximum``, re-adding only from ``@min`` / ``@max``;
+  * inline every ``$ref`` except tables marked ``@shared: <name>`` in the .fbs, which
+    are lifted into ``$defs/<name>`` and referenced (used for tables shared by 2+ parents).
+
+The result is deterministic (fixed key ordering, ``json.dumps(indent=4)``), so the checked-in
+schema is simply this tool's committed output and a drift test can compare byte-for-byte.
+"""
+
+import argparse
+import json
+import re
+import sys
+
+_TOKEN_RE = re.compile(r"^@(title|default|min|max|required|shared)\b\s*:?\s*(.*)$")
+
+_DRAFT_2020_12 = "https://json-schema.org/draft/2020-12/schema"
+
+
+class GenerationError(RuntimeError):
+    """Raised when schema generation fails."""
+
+
+def _parse_default(raw):
+    """Parse an ``@default:`` token value into its JSON type (bool / int / string)."""
+    if raw == "true":
+        return True
+    if raw == "false":
+        return False
+    try:
+        return int(raw)
+    except ValueError:
+        return raw
+
+
+def _parse_description(text):
+    """Split a flatc ``description`` string into (metadata, clean_description).
+
+    ``@token`` lines are extracted into a dict; the remaining lines form the human-readable
+    description (joined with ``\\n``, preserving the original multi-line layout).
+    """
+    meta = {
+        "title": None,
+        "default": None,
+        "min": None,
+        "max": None,
+        "required": False,
+        "shared": None,
+    }
+    desc_lines = []
+    for line in text.split("\n"):
+        match = _TOKEN_RE.match(line)
+        if match is None:
+            desc_lines.append(line)
+            continue
+        key, value = match.group(1), match.group(2).strip()
+        if key == "required":
+            meta["required"] = True
+        elif key in ("min", "max"):
+            meta[key] = int(value)
+        elif key == "default":
+            meta["default"] = _parse_default(value)
+        elif key == "shared":
+            meta["shared"] = value
+        else:  # title
+            meta["title"] = value
+    return meta, "\n".join(desc_lines)
+
+
+class _Enricher:
+    def __init__(self, definitions, shared_defs):
+        self._defs = definitions
+        # Maps a full (namespaced) def name -> the ``$defs`` key it is lifted into.
+        self._shared_defs = shared_defs
+
+    def _ref_name(self, node):
+        return node["$ref"].split("/")[-1]
+
+    def _is_enum(self, def_name):
+        return "enum" in self._defs[def_name]
+
+    def _enum_values(self, def_name):
+        return self._defs[def_name]["enum"]
+
+    def build_field(self, node):
+        """Build an enriched schema node for a property. Returns (schema_node, required)."""
+        if "$ref" in node:
+            ref = self._ref_name(node)
+            if ref in self._shared_defs:
+                meta, _ = _parse_description(node.get("description", ""))
+                return {"$ref": "#/$defs/%s" % self._shared_defs[ref]}, meta["required"]
+            if self._is_enum(ref):
+                meta, desc = _parse_description(node.get("description", ""))
+                out = {"type": "string"}
+                if meta["title"] is not None:
+                    out["title"] = meta["title"]
+                if desc:
+                    out["description"] = desc
+                out["enum"] = self._enum_values(ref)
+                if meta["default"] is not None:
+                    out["default"] = meta["default"]
+                return out, meta["required"]
+            # Non-shared table reference -> inline it. Such fields are authored bare
+            # (metadata lives on the referenced table), so they are never required here.
+            return self.build_object(self._defs[ref]), False
+
+        node_type = node.get("type")
+        if node_type == "array":
+            meta, desc = _parse_description(node.get("description", ""))
+            out = {"type": "array"}
+            if meta["title"] is not None:
+                out["title"] = meta["title"]
+            if desc:
+                out["description"] = desc
+            out["items"] = self._build_items(node["items"])
+            return out, meta["required"]
+
+        # Scalar / string / bool leaf.
+        meta, desc = _parse_description(node.get("description", ""))
+        out = {"type": node_type}
+        if meta["title"] is not None:
+            out["title"] = meta["title"]
+        if desc:
+            out["description"] = desc
+        if "enum" in node:  # inline enum defined directly on the node (not via $ref)
+            out["enum"] = node["enum"]
+        if meta["default"] is not None:
+            out["default"] = meta["default"]
+        if meta["min"] is not None:
+            out["minimum"] = meta["min"]
+        if meta["max"] is not None:
+            out["maximum"] = meta["max"]
+        if node.get("deprecated"):
+            out["deprecated"] = True
+        return out, meta["required"]
+
+    def _build_items(self, items):
+        if "$ref" in items:
+            ref = self._ref_name(items)
+            if ref in self._shared_defs:
+                return {"$ref": "#/$defs/%s" % self._shared_defs[ref]}
+            if self._is_enum(ref):
+                return {"type": "string", "enum": self._enum_values(ref)}
+            return self.build_object(self._defs[ref])
+        # Scalar vector items: drop flatc's type-based min/max (schema has none for these).
+        return {"type": items["type"]}
+
+    def build_object(self, def_node):
+        meta, desc = _parse_description(def_node.get("description", ""))
+        properties = {}
+        required = []
+        for field_name, field_node in def_node.get("properties", {}).items():
+            built, is_required = self.build_field(field_node)
+            properties[field_name] = built
+            if is_required:
+                required.append(field_name)
+        out = {"type": "object"}
+        if meta["title"] is not None:
+            out["title"] = meta["title"]
+        if desc:
+            out["description"] = desc
+        if required:
+            out["required"] = required
+        out["additionalProperties"] = False
+        out["properties"] = properties
+        return out
+
+
+def _collect_shared_defs(definitions):
+    """Map full def name -> ``$defs`` key for every table marked ``@shared: <name>``.
+
+    Iteration follows ``definitions`` order (i.e. .fbs order), keeping ``$defs`` deterministic.
+    """
+    shared = {}
+    for full_name, def_node in definitions.items():
+        meta, _ = _parse_description(def_node.get("description", ""))
+        name = meta["shared"]
+        if name is None:
+            continue
+        if name in shared.values():
+            raise GenerationError("duplicate @shared name: %s" % name)
+        shared[full_name] = name
+    return shared
+
+
+def generate(raw_schema_json):
+    """Post-process raw flatc JSON schema into rich draft-2020-12 format.
+
+    Args:
+        raw_schema_json: Raw JSON schema dict from flatc --jsonschema (draft 2019-09 format)
+
+    Returns:
+        Rich JSON schema as string in draft-2020-12 format
+    """
+    raw = raw_schema_json
+    definitions = raw["definitions"]
+    shared_defs = _collect_shared_defs(definitions)
+    enricher = _Enricher(definitions, shared_defs)
+
+    root_name = raw["$ref"].split("/")[-1]
+    root_obj = enricher.build_object(definitions[root_name])
+
+    schema = {"$schema": _DRAFT_2020_12}
+    if "title" in root_obj:
+        schema["title"] = root_obj["title"]
+    if "description" in root_obj:
+        schema["description"] = root_obj["description"]
+    schema["type"] = "object"
+    if "required" in root_obj:
+        schema["required"] = root_obj["required"]
+    schema["additionalProperties"] = root_obj["additionalProperties"]
+    schema["properties"] = root_obj["properties"]
+    schema["$defs"] = {
+        name: enricher.build_object(definitions[full])
+        for full, name in shared_defs.items()
+    }
+
+    return json.dumps(schema, indent=4, ensure_ascii=False) + "\n"
+
+
+def main(argv=None):
+    argv = list(sys.argv[1:] if argv is None else argv)
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--raw-schema",
+        required=True,
+        help="Path to the raw JSON schema file from flatc --jsonschema.",
+    )
+    parser.add_argument(
+        "--output",
+        default="-",
+        help="Where to write the schema, or '-' for stdout (default: stdout).",
+    )
+    args = parser.parse_args(argv)
+
+    # Read the raw schema from flatc
+    with open(args.raw_schema, "r", encoding="utf-8") as handle:
+        raw_schema_json = json.load(handle)
+
+    schema = generate(raw_schema_json)
+    if args.output == "-":
+        sys.stdout.write(schema)
+    else:
+        with open(args.output, "w", encoding="utf-8") as handle:
+            handle.write(schema)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

score/flatbuffers/bazel/tools.bzl

@@ -457,23 +457,29 @@ def serialize_multiple_versioned_buffers(
     )
 
 def _generate_json_schema_impl(ctx):
-    """Implementation of the generate_json_schema rule."""
+    """Implementation of the generate_json_schema rule.
+
+    Runs flatc --jsonschema to generate a raw JSON schema, then post-processes it
+    with schema_generator.py to convert to draft-2020-12 format and extract metadata.
+    """
 
     # Input files
     schema_file = ctx.file.schema
 
     # Get the flatc compiler using absolute path from flatbuffers repository
     flatc = ctx.executable._flatc
 
+    # Get the schema generator script
+    schema_generator = ctx.executable._schema_generator
+
     # Collect include directories from included .fbs files
     include_files = ctx.files.includes + [ctx.file._buffer_version_fbs]
     include_dirs = {f.dirname: True for f in include_files}
 
-    # When generating JSON schema, flatc generates a file named after the schema file
+    # Step 1: Run flatc --jsonschema to generate raw schema
     default_name = schema_file.basename.replace(".fbs", ".schema.json")
     temp_subdir = "tmp_{}".format(ctx.label.name)
-    generated_file = ctx.actions.declare_file("{}/{}".format(temp_subdir, default_name))
-    out_schema = ctx.actions.declare_file(ctx.attr.output)
+    raw_schema_file = ctx.actions.declare_file("{}/{}".format(temp_subdir, default_name))
 
     # Options for flatc --jsonschema: Generate a JSON Schema from a FlatBuffer schema.
     # Options that apply only to other modes are not listed.
@@ -495,20 +501,34 @@ def _generate_json_schema_impl(ctx):
     args.add("--jsonschema")
     for inc_dir in include_dirs:
         args.add("-I", inc_dir)
-    args.add("-o", generated_file.dirname)
+    args.add("-o", raw_schema_file.dirname)
     args.add(schema_file.path)
 
     ctx.actions.run(
         inputs = [schema_file] + include_files,
-        outputs = [generated_file],
+        outputs = [raw_schema_file],
         executable = flatc,
         arguments = [args],
         mnemonic = "FlatbuffersJsonSchema",
-        progress_message = "Generating JSON schema from %s" % schema_file.short_path,
+        progress_message = "Generating raw JSON schema from %s" % schema_file.short_path,
     )
 
-    # Symlink to the requested output name
-    ctx.actions.symlink(output = out_schema, target_file = generated_file)
+    # Step 2: Post-process the raw schema with schema_generator.py
+    # Converts to draft-2020-12 format and extracts metadata
+    out_schema = ctx.actions.declare_file(ctx.attr.output)
+
+    generator_args = ctx.actions.args()
+    generator_args.add("--raw-schema", raw_schema_file.path)
+    generator_args.add("--output", out_schema.path)
+
+    ctx.actions.run(
+        inputs = [raw_schema_file],
+        outputs = [out_schema],
+        executable = schema_generator,
+        arguments = [generator_args],
+        mnemonic = "FlatbuffersRichJsonSchema",
+        progress_message = "Post-processing JSON schema from %s" % schema_file.short_path,
+    )
 
     return [DefaultInfo(files = depset([out_schema]))]
 
@@ -535,13 +555,22 @@ generate_json_schema = rule(
             cfg = "exec",
             doc = "The flatc compiler (absolute path from flatbuffers repository)",
         ),
+        "_schema_generator": attr.label(
+            default = "//score/flatbuffers/bazel:schema_generator",
+            executable = True,
+            cfg = "exec",
+            doc = "The schema_generator script for post-processing JSON schema output",
+        ),
         "_buffer_version_fbs": attr.label(
             default = "@score_baselibs//score/flatbuffers/common:buffer_version.fbs",
             allow_single_file = [".fbs"],
             doc = "Automatically included buffer_version.fbs for common buffer version support.",
         ),
     },
-    doc = """Generates a JSON schema from a FlatBuffer schema.
+    doc = """Generates a rich JSON schema from a FlatBuffer schema.
+
+    This rule runs flatc --jsonschema and post-processes the output to convert it to
+    draft-2020-12 format with extracted metadata (@title, @default, @min, @max, etc.).
 
     @score_baselibs//score/flatbuffers/common:buffer_version.fbs is always included
     automatically. The schema must include buffer_version.fbs manually if it uses