feat: separate user-facing schema (write/read) from internal model [INFP-234]

backend/infrahub/api/schema.py

@@ -3,18 +3,29 @@
 from typing import TYPE_CHECKING, Any, Sequence
 
 from fastapi import APIRouter, Depends, Query, Request
+from infrahub_sdk.schema.generated.read import (
+    GenericSchemaRead,
+    NodeSchemaRead,
+    ProfileSchemaRead,
+    TemplateSchemaRead,
+)
+from infrahub_sdk.schema.generated.write import InfrahubSchemaWrite
+from infrahub_sdk.schema.validate import validate_schema as validate_write_schema
 from pydantic import (
     BaseModel,
     Field,
+    PrivateAttr,
     computed_field,
     create_model,
     model_validator,
 )
+from pydantic import ValidationError as PydanticValidationError
 from starlette.responses import JSONResponse
 
 from infrahub import lock
 from infrahub.api.dependencies import get_branch_dep, get_context, get_current_user, get_db, get_permission_manager
 from infrahub.api.exceptions import SchemaNotValidError
+from infrahub.api.schema_write_projection import project_to_write_contract
 from infrahub.branch.status_checker import BranchStatusChecker
 from infrahub.core import registry
 from infrahub.core.account import GlobalPermission
@@ -75,35 +86,26 @@ def from_schema(cls, schema: MainSchemaTypes) -> Self:
         data["relationships"] = [
             relationship.model_dump() for relationship in schema.relationships if not relationship.internal_peer
         ]
+        # ``kind`` is a computed field on the generated read model (derived from namespace+name);
+        # only ``hash`` needs to be supplied here since it is computed by the server.
         data["hash"] = schema.get_hash()
         return cls(**data)
 
-    @model_validator(mode="before")
-    @classmethod
-    def set_kind(cls, values: Any) -> Any:
-        if isinstance(values, dict):
-            values["kind"] = f"{values['namespace']}{values['name']}"
-        return values
-
 
-class APINodeSchema(NodeSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
+class APINodeSchema(NodeSchemaRead, APISchemaMixin):
+    pass
 
 
-class APIGenericSchema(GenericSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
+class APIGenericSchema(GenericSchemaRead, APISchemaMixin):
+    pass
 
 
-class APIProfileSchema(ProfileSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
+class APIProfileSchema(ProfileSchemaRead, APISchemaMixin):
+    pass
 
 
-class APITemplateSchema(TemplateSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
+class APITemplateSchema(TemplateSchemaRead, APISchemaMixin):
+    pass
 
 
 class SchemaReadAPI(BaseModel):
@@ -115,8 +117,71 @@ class SchemaReadAPI(BaseModel):
     namespaces: list[SchemaNamespace] = Field(default_factory=list)
 
 
-class SchemaLoadAPI(SchemaRoot):
+def _format_schema_errors(errors: list[Any]) -> list[str]:
+    """Render internal-schema validation errors as field-level messages.
+
+    The dotted path indexes list elements with ``[i]`` (``nodes[0].relationships[0].cardinality``)
+    and appends the offending scalar value so the message names both the field and what was received.
+    """
+    messages: list[str] = []
+    for error in errors:
+        parts: list[str] = []
+        for element in error["loc"]:
+            if isinstance(element, int) and parts:
+                parts[-1] = f"{parts[-1]}[{element}]"
+            else:
+                parts.append(str(element))
+        message = f"{'.'.join(parts)}: {error['msg']}"
+        received = error.get("input")
+        if isinstance(received, (str, int, float, bool)) or received is None:
+            message += f" (received: {received!r})"
+        messages.append(message)
+    return messages
+
+
+class SchemaLoadAPI(InfrahubSchemaWrite):
     version: str
+    _internal_schema: SchemaRoot = PrivateAttr()
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_write_contract(cls, data: Any) -> Any:
+        # A submitted schema may legitimately carry read-only/internal fields — a schema read back
+        # from Infrahub, or a full internal dump. The internal schema accepts those but still
+        # rejects fields that are unknown or invalid for their context (a typo, or a node-only
+        # field on an extension), drawing the tolerate/reject line a uniform extra="forbid" cannot.
+        if isinstance(data, dict):
+            try:
+                SchemaRoot.model_validate(data)
+            except PydanticValidationError as exc:
+                # Only reject genuinely unknown/forbidden fields here. Constraint and enum
+                # violations, reserved keywords, and other invariants are surfaced with their
+                # original messages by the write-contract check below or by build_internal_schema.
+                unknown = [error for error in exc.errors() if error["type"] == "extra_forbidden"]
+                if unknown:
+                    raise ValueError("; ".join(_format_schema_errors(unknown))) from exc
+
+            # Drop the read-only/internal fields so the strict write models accept the payload; the
+            # unknown-field check above has already run, so nothing is hidden. What remains is the
+            # user-facing write contract, validated for out-of-range constrained values.
+            data = project_to_write_contract(data)
+            result = validate_write_schema(schema=data)
+            if not result.valid:
+                raise ValueError("; ".join(result.messages))
+        return data
+
+    @model_validator(mode="after")
+    def build_internal_schema(self) -> Self:
+        # The published request contract only exposes user-settable fields; the processing
+        # engine operates on the richer internal SchemaRoot. Building it here surfaces
+        # internal-only invariants (e.g. reserved keywords) as request-validation errors,
+        # and lets the endpoints feed the pipeline without re-deriving the conversion.
+        self._internal_schema = SchemaRoot.model_validate(self.model_dump(exclude_none=True))
+        return self
+
+    @property
+    def internal_schema(self) -> SchemaRoot:
+        return self._internal_schema
 
 
 class SchemasLoadAPI(BaseModel):
@@ -170,11 +235,11 @@ def _merge_candidate_schemas(schemas: Sequence[SchemaRoot]) -> SchemaRoot:
 
 
 def evaluate_candidate_schemas(
-    branch_schema: SchemaBranch, schemas_to_evaluate: SchemasLoadAPI
+    branch_schema: SchemaBranch, schemas_to_evaluate: Sequence[SchemaRoot]
 ) -> tuple[SchemaBranch, SchemaUpdateValidationResult]:
     candidate_schema = branch_schema.duplicate()
     try:
-        schema = _merge_candidate_schemas(schemas=schemas_to_evaluate.schemas)
+        schema = _merge_candidate_schemas(schemas=schemas_to_evaluate)
 
         candidate_schema.load_schema(schema=schema)
         candidate_schema.process()
@@ -357,10 +422,13 @@ async def load_schema(
 
     errors: list[str] = []
     warnings: list[SchemaWarning] = []
+    candidate_schemas: list[SchemaRoot] = []
     for schema in schemas.schemas:
-        errors += schema.validate_namespaces()
-        errors += schema.validate_reserved_suffixes()
-        warnings += schema.gather_warnings()
+        internal_schema = schema.internal_schema
+        candidate_schemas.append(internal_schema)
+        errors += internal_schema.validate_namespaces()
+        errors += internal_schema.validate_reserved_suffixes()
+        warnings += internal_schema.gather_warnings()
 
     if errors:
         raise SchemaNotValidError(message=", ".join(errors))
@@ -369,7 +437,9 @@ async def load_schema(
         branch_schema = registry.schema.get_schema_branch(name=branch.name)
         original_hash = branch_schema.get_hash()
 
-        candidate_schema, result = evaluate_candidate_schemas(branch_schema=branch_schema, schemas_to_evaluate=schemas)
+        candidate_schema, result = evaluate_candidate_schemas(
+            branch_schema=branch_schema, schemas_to_evaluate=candidate_schemas
+        )
 
         if not result.diff.all:
             return SchemaUpdate(hash=original_hash, previous_hash=original_hash, diff=result.diff)
@@ -445,17 +515,22 @@ async def check_schema(
 
     errors: list[str] = []
     warnings: list[SchemaWarning] = []
+    candidate_schemas: list[SchemaRoot] = []
     for schema in schemas.schemas:
-        errors += schema.validate_namespaces()
-        errors += schema.validate_reserved_suffixes()
-        warnings += schema.gather_warnings()
+        internal_schema = schema.internal_schema
+        candidate_schemas.append(internal_schema)
+        errors += internal_schema.validate_namespaces()
+        errors += internal_schema.validate_reserved_suffixes()
+        warnings += internal_schema.gather_warnings()
 
     if errors:
         raise SchemaNotValidError(message=", ".join(errors))
 
     branch_schema = registry.schema.get_schema_branch(name=branch.name)
 
-    candidate_schema, result = evaluate_candidate_schemas(branch_schema=branch_schema, schemas_to_evaluate=schemas)
+    candidate_schema, result = evaluate_candidate_schemas(
+        branch_schema=branch_schema, schemas_to_evaluate=candidate_schemas
+    )
 
     # ----------------------------------------------------------
     # Validate if the new schema is valid with the content of the database

backend/infrahub/api/schema_write_projection.py

@@ -0,0 +1,95 @@
+"""Project a schema-load payload onto the user-facing write contract.
+
+A load payload may carry read-only and internal fields (a schema read back from Infrahub, or a
+full internal model dump used by tests). Those fields are legitimate — they are validated against
+the internal schema first — but the strict write models reject them. This projection keeps only the
+write-settable fields, resolving the per-kind attribute and computed-attribute discriminated unions
+so each kind keeps the fields valid for it. It runs only after the internal validation has already
+rejected genuinely unknown or forbidden fields, so dropping non-write keys never hides a mistake.
+"""
+
+from __future__ import annotations
+
+import functools
+import types
+import typing
+from typing import Any
+
+from infrahub_sdk.schema.generated.write import InfrahubSchemaWrite
+from pydantic import BaseModel
+from pydantic.fields import FieldInfo
+
+_UNION_ORIGINS = {typing.Union, types.UnionType}
+_SEQUENCE_ORIGINS = {list, tuple}
+
+
+def project_to_write_contract(schema: dict[str, Any]) -> dict[str, Any]:
+    """Return a copy of ``schema`` containing only fields accepted by the write contract."""
+    if not isinstance(schema, dict):
+        return schema
+    return _project_model(schema, InfrahubSchemaWrite)
+
+
+@functools.cache
+def _resolved_hints(model: type[BaseModel]) -> dict[str, Any]:
+    # ``model_fields[...].annotation`` leaves module-level Annotated aliases (the discriminated
+    # unions) as unresolved forward references; ``get_type_hints`` resolves them with the metadata.
+    return typing.get_type_hints(model, include_extras=True)
+
+
+def _project_model(data: Any, model: type[BaseModel]) -> Any:
+    if not isinstance(data, dict):
+        return data
+    hints = _resolved_hints(model)
+    projected: dict[str, Any] = {}
+    for name, info in model.model_fields.items():
+        key = name if name in data else (info.alias if info.alias in data else None)
+        if key is None:
+            continue
+        discriminator = info.discriminator if isinstance(info.discriminator, str) else None
+        projected[key] = _project_value(data[key], hints.get(name, info.annotation), discriminator)
+    return projected
+
+
+def _project_value(value: Any, annotation: Any, discriminator: str | None = None) -> Any:
+    # Annotated[...] (used for the discriminated unions): unwrap to the base + its discriminator.
+    if hasattr(annotation, "__metadata__"):
+        nested = next(
+            (
+                m.discriminator
+                for m in annotation.__metadata__
+                if isinstance(m, FieldInfo) and isinstance(m.discriminator, str)
+            ),
+            None,
+        )
+        return _project_value(value, annotation.__origin__, nested or discriminator)
+
+    origin = typing.get_origin(annotation)
+    args = typing.get_args(annotation)
+
+    if origin in _UNION_ORIGINS:
+        members = [a for a in args if a is not type(None)]
+        if discriminator and isinstance(value, dict):
+            member = _pick_variant(members, discriminator, value)
+            if member is not None:
+                return _project_model(value, member)
+        return _project_value(value, members[0]) if len(members) == 1 else value
+
+    if origin in _SEQUENCE_ORIGINS and args and isinstance(value, list):
+        return [_project_value(item, args[0]) for item in value]
+
+    if isinstance(annotation, type) and issubclass(annotation, BaseModel):
+        return _project_model(value, annotation)
+
+    return value
+
+
+def _pick_variant(members: list[Any], discriminator: str, value: dict[str, Any]) -> type[BaseModel] | None:
+    target = str(getattr(value.get(discriminator), "value", value.get(discriminator)))
+    for member in members:
+        if not (isinstance(member, type) and issubclass(member, BaseModel)):
+            continue
+        field = member.model_fields.get(discriminator)
+        if field is not None and target in {str(getattr(a, "value", a)) for a in typing.get_args(field.annotation)}:
+            return member
+    return None

backend/infrahub/core/constants/__init__.py

@@ -6,7 +6,14 @@
 from infrahub.exceptions import ValidationError
 from infrahub.utils import InfrahubNumberEnum, InfrahubStringEnum
 
-from .schema import FlagProperty, NodeProperty, SchemaElementPathType, UpdateSupport, UpdateValidationErrorType
+from .schema import (
+    FlagProperty,
+    NodeProperty,
+    SchemaElementPathType,
+    UpdateSupport,
+    UpdateValidationErrorType,
+    Visibility,
+)
 
 __all__ = [
     "FlagProperty",
@@ -16,6 +23,7 @@
     "UpdateSupport",
     "UpdateValidationErrorType",
     "ValidationError",
+    "Visibility",
 ]
 
 

backend/infrahub/core/constants/schema.py

@@ -1,4 +1,4 @@
-from enum import Enum, Flag, auto
+from enum import Enum, Flag, IntEnum, auto
 
 PARENT_CHILD_IDENTIFIER = "parent__child"
 RESOURCE_POOL_REL_SUFFIX = "_from_resource_pool"
@@ -21,6 +21,18 @@ class UpdateSupport(Enum):
     NOT_APPLICABLE = "not_applicable"
 
 
+class Visibility(IntEnum):
+    """Ordinal classification of who may see or set a schema field.
+
+    The ordering is meaningful: a field is included in a given model variant when the
+    variant's minimum level is at or below the field's visibility (write ⊆ read ⊆ internal).
+    """
+
+    INTERNAL = 0
+    READ = 1
+    WRITE = 2
+
+
 class UpdateValidationErrorType(Enum):
     NOT_SUPPORTED = "not_supported"
     VALIDATOR_FAILED = "validator_failed"