Process schema deprecation warnings

changelog/+f6791a3d.added.md

@@ -0,0 +1 @@
+Added deprecation warnings when loading or checking schemas

infrahub_sdk/ctl/schema.py

@@ -14,6 +14,7 @@
 from ..ctl.client import initialize_client
 from ..ctl.utils import catch_exception, init_logging
 from ..queries import SCHEMA_HASH_SYNC_STATUS
+from ..schema import SchemaWarning
 from ..yaml import SchemaFile
 from .parameters import CONFIG_PARAM
 from .utils import load_yamlfile_from_disk_and_exit
@@ -152,6 +153,8 @@ async def load(
 
     console.print(f"[green] {len(schemas_data)} {schema_definition} processed in {loading_time:.3f} seconds.")
 
+    _display_schema_warnings(console=console, warnings=response.warnings)
+
     if response.schema_updated and wait:
         waited = 0
         continue_waiting = True
@@ -187,12 +190,24 @@ async def check(
 
     success, response = await client.schema.check(schemas=[item.payload for item in schemas_data], branch=branch)
 
-    if not success:
+    if not success or not response:
         display_schema_load_errors(response=response or {}, schemas_data=schemas_data)
+        return
+
+    for schema_file in schemas_data:
+        console.print(f"[green] schema '{schema_file.location}' is Valid!")
+
+    warnings = response.pop("warnings", [])
+    schema_warnings = [SchemaWarning.model_validate(warning) for warning in warnings]
+    _display_schema_warnings(console=console, warnings=schema_warnings)
+    if response == {"diff": {"added": {}, "changed": {}, "removed": {}}}:
+        print("No diff")
     else:
-        for schema_file in schemas_data:
-            console.print(f"[green] schema '{schema_file.location}' is Valid!")
-        if response == {"diff": {"added": {}, "changed": {}, "removed": {}}}:
-            print("No diff")
-        else:
-            print(yaml.safe_dump(data=response, indent=4))
+        print(yaml.safe_dump(data=response, indent=4))
+
+
+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])}]"
+        )

infrahub_sdk/schema/__init__.py

@@ -90,6 +90,26 @@ class EnumMutation(str, Enum):
 ]
 
 
+class SchemaWarningType(Enum):
+    DEPRECATION = "deprecation"
+
+
+class SchemaWarningKind(BaseModel):
+    kind: str = Field(..., description="The kind impacted by the warning")
+    field: str | None = Field(default=None, description="The attribute or relationship impacted by the warning")
+
+    @property
+    def display(self) -> str:
+        suffix = f".{self.field}" if self.field else ""
+        return f"{self.kind}{suffix}"
+
+
+class SchemaWarning(BaseModel):
+    type: SchemaWarningType = Field(..., description="The type of warning")
+    kinds: list[SchemaWarningKind] = Field(default_factory=list, description="The kinds impacted by the warning")
+    message: str = Field(..., description="The message that describes the warning")
+
+
 class InfrahubSchemaBase:
     client: InfrahubClient | InfrahubClientSync
     cache: dict[str, BranchSchema]
@@ -169,7 +189,9 @@ def generate_payload_create(
     def _validate_load_schema_response(response: httpx.Response) -> SchemaLoadResponse:
         if response.status_code == httpx.codes.OK:
             status = response.json()
-            return SchemaLoadResponse(hash=status["hash"], previous_hash=status["previous_hash"])
+            return SchemaLoadResponse(
+                hash=status["hash"], previous_hash=status["previous_hash"], warnings=status.get("warnings") or []
+            )
 
         if response.status_code in [
             httpx.codes.BAD_REQUEST,
@@ -802,6 +824,7 @@ class SchemaLoadResponse(BaseModel):
     hash: str = Field(default="", description="The new hash for the entire schema")
     previous_hash: str = Field(default="", description="The previous hash for the entire schema")
     errors: dict = Field(default_factory=dict, description="Errors reported by the server")
+    warnings: list[SchemaWarning] = Field(default_factory=list, description="Warnings reported by the server")
 
     @property
     def schema_updated(self) -> bool: