pudl_diff.report_schema ¶

def main(schema_path: Path = SCHEMA_PATH, docs_path: Path = DOCS_PATH) -> int:
    """Write the report's JSON Schema and its documentation, if they have changed.

    Returns:
        `1` if either file was out of date and had to be updated, so that this can
        be used as a pre-commit hook, and `0` otherwise.
    """
    updated = [
        _write_if_changed(schema_path, report_json_schema_text()),
        _write_if_changed(docs_path, report_docs_text()),
    ]
    return int(any(updated))

def report_docs_text() -> str:
    """The text of the documentation of every field of the report, as Markdown."""
    parts = [_DOCS_INTRO]
    for model in schema_models(report_json_schema(), _link):
        parts.append(f"## {model['name']}\n\n{model['description']}\n")
        for field in model["fields"]:
            derived = " *Derived from the other fields.*" if field["derived"] else ""
            parts.append(
                f"### `{field['name']}`\n\n*Type:* {field['type']}.{derived}\n\n"
                f"{field['description']}\n"
            )
    return "\n".join(parts)

def report_json_schema() -> JsonSchemaValue:
    """The JSON Schema of the JSON report, which every `PudlDiffReport` conforms to.

    Generated from the report's models, so that the schema and the descriptions of
    what each field means always match the code. It describes the report as it is
    written (its `serialization` schema): fields that have a default value are
    still always present, and fields derived from others, like
    `PudlDiffReport.is_identical`, are listed and marked read-only.
    """
    schema = PudlDiffReport.model_json_schema(mode="serialization")

    def clean(node: object) -> None:
        if isinstance(node, dict):
            if isinstance(node.get("description"), str):
                node["description"] = _plain_description(node["description"])
            for child in node.values():
                clean(child)
        elif isinstance(node, list):
            for child in node:
                clean(child)

    clean(schema)
    return schema

def report_json_schema_text() -> str:
    """`report_json_schema()` as the text of the committed schema file."""
    return json.dumps(report_json_schema(), indent=2) + "\n"

def schema_models(
    schema: JsonSchemaValue, link: Callable[[str], str] = str
) -> list[JsonSchemaValue]:
    """Flatten a report schema into a list of models, for rendering as documentation.

    The report's own model comes first, followed by the models it refers to, in the
    order that they're first mentioned. Each is a dictionary with its `name`,
    `description`, and `fields`: a list of dictionaries with each field's
    `name`, `type`, `description` and whether it is `derived` from other fields.
    """
    models = {_ROOT_MODEL: schema, **schema.get("$defs", {})}
    order = [_ROOT_MODEL]
    for name in order:
        for spec in models[name].get("properties", {}).values():
            for node in _walk(spec):
                if "$ref" in node and (ref := _model_name(node["$ref"])) not in order:
                    order.append(ref)
    return [
        {
            "name": name,
            "description": models[name].get("description", ""),
            "fields": [
                {
                    "name": field,
                    "type": schema_type(spec, link),
                    "description": spec.get("description", ""),
                    "derived": bool(spec.get("readOnly")),
                }
                for field, spec in models[name].get("properties", {}).items()
            ],
        }
        for name in order
    ]

def schema_type(spec: JsonSchemaValue, link: Callable[[str], str] = str) -> str:
    """Describe the type of a property of the schema in a few words.

    Args:
        spec: The property's schema.
        link: How to write the name of a model that the property refers to.
    """
    if "$ref" in spec:
        return link(_model_name(spec["$ref"]))
    if "const" in spec:
        return json.dumps(spec["const"])
    if "enum" in spec:
        return " or ".join(json.dumps(value) for value in spec["enum"])
    for key in ("anyOf", "oneOf"):
        if key in spec:
            label = " or ".join(schema_type(option, link) for option in spec[key])
            if "discriminator" in spec:
                field = spec["discriminator"]["propertyName"]
                label += f", told apart by `{field}`"
            return label
    kind = spec.get("type")
    if kind == "array":
        if "prefixItems" in spec:
            items = ", ".join(schema_type(item, link) for item in spec["prefixItems"])
            return f"[{items}]"
        return f"array of {schema_type(spec.get('items', {}), link)}"
    if kind == "object" and "additionalProperties" in spec:
        values = schema_type(spec["additionalProperties"], link)
        return f"object mapping names to {values}"
    return str(kind or "any")