feat: report the schema fields the write contract does not apply

infrahub_sdk/ctl/schema.py

@@ -4,26 +4,23 @@
 import time
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Literal
+from typing import Any, Literal
 
 import typer
 import yaml
-from pydantic import ValidationError
 from rich.console import Console
+from rich.markup import escape
 from rich.table import Table
 
 from ..async_typer import AsyncTyper
 from ..ctl.client import initialize_client
 from ..ctl.utils import catch_exception, init_logging
 from ..queries import SCHEMA_HASH_SYNC_STATUS
-from ..schema import NodeSchemaAPI, SchemaWarning
+from ..schema import NodeSchemaAPI, SchemaWarning, validate_schema
 from ..yaml import SchemaFile
 from .parameters import CONFIG_PARAM
 from .utils import load_yamlfile_from_disk_and_exit
 
-if TYPE_CHECKING:
-    from .. import InfrahubClient
-
 SchemaContainer = Literal["nodes", "generics", "relationships"]
 
 app = AsyncTyper()
@@ -35,17 +32,21 @@ def callback() -> None:
     """Manage the schema in a remote Infrahub instance."""
 
 
-def validate_schema_content_and_exit(client: InfrahubClient, schemas: list[SchemaFile]) -> None:
+def validate_schema_content_and_exit(schemas: list[SchemaFile]) -> None:
+    """Report every offline contract violation and exit when at least one schema is invalid.
+
+    Read-only fields are reported by the server on the load/check response, so only errors are
+    rendered here to avoid warning about the same field twice.
+    """
     has_error: bool = False
     for schema_file in schemas:
-        try:
-            client.schema.validate(data=schema_file.payload)
-        except ValidationError as exc:
-            console.print(f"[red]Schema not valid, found '{len(exc.errors())}' error(s) in {schema_file.location}")
-            has_error = True
-            for error in exc.errors():
-                loc_str = [str(item) for item in error["loc"]]
-                console.print(f"  '{'/'.join(loc_str)}' | {error['msg']} ({error['type']})")
+        result = validate_schema(schema=schema_file.payload)
+        if result.valid:
+            continue
+        has_error = True
+        console.print(f"[red]Schema not valid, found '{len(result.errors)}' error(s) in {schema_file.location}")
+        for error in result.errors:
+            console.print(f"  {escape(error.message)}")
 
     if has_error:
         raise typer.Exit(1)
@@ -190,7 +191,7 @@ async def load(
     schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
     schema_definition = "schema" if len(schemas_data) == 1 else "schemas"
     client = initialize_client()
-    validate_schema_content_and_exit(client=client, schemas=schemas_data)
+    validate_schema_content_and_exit(schemas=schemas_data)
 
     start_time = time.time()
     response = await client.schema.load(schemas=[item.payload for item in schemas_data], branch=branch)
@@ -240,7 +241,7 @@ async def check(
 
     schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
     client = initialize_client()
-    validate_schema_content_and_exit(client=client, schemas=schemas_data)
+    validate_schema_content_and_exit(schemas=schemas_data)
 
     success, response = await client.schema.check(schemas=[item.payload for item in schemas_data], branch=branch)
 
@@ -262,9 +263,9 @@ async def check(
 
 def _display_schema_warnings(console: Console, warnings: list[SchemaWarning]) -> None:
     for warning in warnings:
-        console.print(
-            f"[yellow] {warning.type.value}: {warning.message} [{', '.join([kind.display for kind in warning.kinds])}]"
-        )
+        # A warning about a top-level key has no kind to attribute it to.
+        kinds = f" [{', '.join(kind.display for kind in warning.kinds)}]" if warning.kinds else ""
+        console.print(f"[yellow] {warning.type.value}: {escape(warning.message)}{escape(kinds)}")
 
 
 def _default_export_directory() -> Path:

infrahub_sdk/ctl/validate.py

@@ -5,14 +5,15 @@
 
 import typer
 import ujson
-from pydantic import ValidationError
 from rich.console import Console
+from rich.markup import escape
 
 from ..async_typer import AsyncTyper
-from ..ctl.client import initialize_client, initialize_client_sync
+from ..ctl.client import initialize_client_sync
 from ..ctl.exceptions import QueryNotFoundError
 from ..ctl.utils import catch_exception, find_graphql_query, parse_cli_vars
 from ..exceptions import GraphQLError
+from ..schema import validate_schema as validate_schema_offline
 from ..utils import write_to_file
 from ..yaml import SchemaFile
 from .parameters import CONFIG_PARAM
@@ -36,16 +37,16 @@ async def validate_schema(schema: Path, _: str = CONFIG_PARAM) -> None:
         console.print(f"[red]Unable to find {schema}")
         raise typer.Exit(1)
 
-    client = initialize_client()
+    result = validate_schema_offline(schema=schema_data[0].payload)
 
-    try:
-        client.schema.validate(schema_data[0].payload)
-    except ValidationError as exc:
-        console.print(f"[red]Schema not valid, found {len(exc.errors())} error(s)")
-        for error in exc.errors():
-            loc_str = [str(item) for item in error["loc"]]
-            console.print(f"  '{'/'.join(loc_str)}' | {error['msg']} ({error['type']})")
-        raise typer.Exit(1) from None
+    for warning in result.warnings:
+        console.print(f"[yellow]{escape(warning.message)}")
+
+    if not result.valid:
+        console.print(f"[red]Schema not valid, found {len(result.errors)} error(s)")
+        for error in result.errors:
+            console.print(f"  {escape(error.message)}")
+        raise typer.Exit(1)
 
     console.print("[green]Schema is valid !!")
 

infrahub_sdk/schema/__init__.py

@@ -44,7 +44,12 @@
     SchemaRootAPI,
     TemplateSchemaAPI,
 )
-from .validate import SchemaValidationErrorDetail, SchemaValidationResult, validate_schema
+from .validate import (
+    SchemaValidationErrorDetail,
+    SchemaValidationResult,
+    SchemaValidationWarningDetail,
+    validate_schema,
+)
 
 if TYPE_CHECKING:
     from ..client import InfrahubClient, InfrahubClientSync, SchemaType, SchemaTypeSync
@@ -74,6 +79,7 @@
     "SchemaRootAPI",
     "SchemaValidationErrorDetail",
     "SchemaValidationResult",
+    "SchemaValidationWarningDetail",
     "TemplateSchemaAPI",
     "schema_to_export_dict",
     "validate_schema",
@@ -173,10 +179,19 @@ def _build_export_schemas(
                 ns_map[ns].nodes.append(schema_dict)
         return SchemaExport(namespaces=ns_map)
 
-    def validate(self, data: dict[str, Any]) -> None:
-        # Validate against the generated write contract so this matches what /api/schema/load
-        # enforces (unknown keys rejected, attribute kinds discriminated, extensions understood).
-        InfrahubSchemaWrite.model_validate(data)
+    def validate(self, data: dict[str, Any]) -> SchemaValidationResult:
+        """Validate a schema payload against the generated write contract.
+
+        Returns:
+            The verdict, carrying a warning for every read-only field the payload sets.
+
+        Raises:
+            ValueError: When the payload is invalid, joining every field-level message.
+
+        """
+        # Delegating to the offline validator keeps this verdict identical to the one
+        # /api/schema/load reaches, since the server runs the same models.
+        return validate_schema(schema=data, raise_on_error=True)
 
     def validate_data_against_schema(self, schema: MainSchemaTypesAPI, data: dict) -> None:
         for key in data:

infrahub_sdk/schema/generated/__init__.py

@@ -1,4 +1,4 @@
 # Generated by "invoke backend.generate", do not edit directly
-from . import enums, read, write
+from . import contract, enums, read, write
 
-__all__ = ["enums", "read", "write"]
+__all__ = ["contract", "enums", "read", "write"]

infrahub_sdk/schema/generated/contract.py

@@ -0,0 +1,29 @@
+# Generated by "invoke backend.generate", do not edit directly
+"""Read-only fields of the write contract, keyed by generated write class name.
+
+A field listed here is one the contract knows at that location but the user may not set:
+a field the read API returns, the bookkeeping a schema dumped from the internal models
+carries, or a field belonging to a sibling variant of a discriminated union. Submitting one
+is reported as a warning and the value is dropped, where an extra field that is not listed
+is an error. Lookups union the entries of every class in the model's MRO.
+"""
+
+READ_ONLY_FIELDS: dict[str, frozenset[str]] = {
+    "AttributeParametersWrite": frozenset({"id", "state"}),
+    "AttributeSchemaBaseWrite": frozenset({"inherited"}),
+    "BaseNodeSchemaWrite": frozenset({"hash", "kind"}),
+    "ComputedAttributeJinja2Write": frozenset({"id", "state", "transform"}),
+    "ComputedAttributeTransformPythonWrite": frozenset({"id", "jinja2_template", "state"}),
+    "ComputedAttributeUserWrite": frozenset({"id", "jinja2_template", "state", "transform"}),
+    "DropdownChoiceWrite": frozenset({"id", "state"}),
+    "GenericSchemaWrite": frozenset({"used_by"}),
+    "InfrahubSchemaWrite": frozenset({"main", "namespaces", "profiles", "templates"}),
+    "ListAttributeParametersWrite": frozenset({"id", "state"}),
+    "NodeExtensionWrite": frozenset({"id", "state"}),
+    "NodeSchemaWrite": frozenset({"hierarchy"}),
+    "NumberAttributeParametersWrite": frozenset({"id", "state"}),
+    "NumberPoolParametersWrite": frozenset({"id", "state"}),
+    "RelationshipSchemaWrite": frozenset({"hierarchical", "inherited"}),
+    "SchemaExtensionWrite": frozenset({"id", "state"}),
+    "TextAttributeParametersWrite": frozenset({"id", "state"}),
+}