diff --git a/changelog/+f6791a3d.added.md b/changelog/+f6791a3d.added.md new file mode 100644 index 00000000..4a78b2eb --- /dev/null +++ b/changelog/+f6791a3d.added.md @@ -0,0 +1 @@ +Added deprecation warnings when loading or checking schemas diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index cb876569..5a977b59 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/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])}]" + ) diff --git a/infrahub_sdk/schema/__init__.py b/infrahub_sdk/schema/__init__.py index 4c2b9c79..a89d7dc5 100644 --- a/infrahub_sdk/schema/__init__.py +++ b/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: