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

backend/infrahub/api/schema.py

@@ -3,9 +3,18 @@
 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,
@@ -75,35 +84,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(NodeSchemaRead, APISchemaMixin):
+    pass
 
-class APINodeSchema(NodeSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
 
+class APIGenericSchema(GenericSchemaRead, APISchemaMixin):
+    pass
 
-class APIGenericSchema(GenericSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
 
+class APIProfileSchema(ProfileSchemaRead, APISchemaMixin):
+    pass
 
-class APIProfileSchema(ProfileSchema, APISchemaMixin):
-    api_kind: str | None = Field(default=None, alias="kind", validate_default=True)
-    hash: str
 
-
-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 +115,35 @@ class SchemaReadAPI(BaseModel):
     namespaces: list[SchemaNamespace] = Field(default_factory=list)
 
 
-class SchemaLoadAPI(SchemaRoot):
+class SchemaLoadAPI(InfrahubSchemaWrite):
     version: str
+    _internal_schema: SchemaRoot = PrivateAttr()
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_write_contract(cls, data: Any) -> Any:
+        # The user-facing "write" contract lives in the SDK; each submitted node/generic is
+        # validated against it so that fields the user may not set (read-level, internal, or
+        # unknown) and out-of-range constrained values are rejected at the boundary with a
+        # field-level message, before the payload is coerced into the richer internal models.
+        if isinstance(data, dict):
+            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 +197,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 +384,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 +399,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 +477,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/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"