Add menu and object cli commands

dgarros · dgarros · commit 1588fe7027eb · 2024-10-12T17:06:48.000+02:00
diff --git a/infrahub_sdk/ctl/cli_commands.py b/infrahub_sdk/ctl/cli_commands.py
@@ -22,6 +22,8 @@
 from infrahub_sdk.ctl.client import initialize_client, initialize_client_sync
 from infrahub_sdk.ctl.exceptions import QueryNotFoundError
 from infrahub_sdk.ctl.generator import run as run_generator
+from infrahub_sdk.ctl.menu import app as menu_app
+from infrahub_sdk.ctl.object import app as object_app
 from infrahub_sdk.ctl.render import list_jinja2_transforms
 from infrahub_sdk.ctl.repository import app as repository_app
 from infrahub_sdk.ctl.repository import get_repository_config
@@ -50,14 +52,17 @@
 app.add_typer(schema_app, name="schema")
 app.add_typer(validate_app, name="validate")
 app.add_typer(repository_app, name="repository")
+app.add_typer(menu_app, name="menu")
+app.add_typer(object_app, name="object")
+
 app.command(name="dump")(dump)
 app.command(name="load")(load)
 
 console = Console()
 
 
 @app.command(name="check")
-@catch_exception(console=console)
+# @catch_exception(console=console)
 def check(
     check_name: str = typer.Argument(default="", help="Name of the Python check"),
     branch: Optional[str] = None,
diff --git a/infrahub_sdk/ctl/menu.py b/infrahub_sdk/ctl/menu.py
@@ -0,0 +1,52 @@
+from pathlib import Path
+
+import typer
+from rich.console import Console
+
+from infrahub_sdk.async_typer import AsyncTyper
+from infrahub_sdk.ctl.client import initialize_client
+from infrahub_sdk.ctl.utils import catch_exception, init_logging
+from infrahub_sdk.spec.menu import MenuFile
+
+from .parameters import CONFIG_PARAM
+from .utils import load_yamlfile_from_disk_and_exit
+
+app = AsyncTyper()
+console = Console()
+
+
+@app.callback()
+def callback() -> None:
+    """
+    Manage the menu in a remote Infrahub instance.
+    """
+
+
+@app.command()
+@catch_exception(console=console)
+async def load(
+    menus: list[Path],
+    debug: bool = False,
+    branch: str = typer.Option("main", help="Branch on which to load the menu."),
+    _: str = CONFIG_PARAM,
+) -> None:
+    """Load one or multiple menu files into Infrahub."""
+
+    init_logging(debug=debug)
+
+    files = load_yamlfile_from_disk_and_exit(paths=menus, file_type=MenuFile, console=console)
+    client = await initialize_client()
+
+    default_kind = "CoreMenuItem"
+
+    for file in files:
+        file.validate_content()
+        if not file.spec.kind:
+            file.spec.kind = default_kind
+
+        schema = await client.schema.get(kind=file.spec.kind, branch=branch)
+
+        for item in file.spec.data:
+            await file.spec.create_node(
+                client=client, schema=schema, data=item, branch=branch, default_schema_kind=default_kind
+            )
diff --git a/infrahub_sdk/ctl/object.py b/infrahub_sdk/ctl/object.py
@@ -0,0 +1,48 @@
+from pathlib import Path
+
+import typer
+from rich.console import Console
+
+from infrahub_sdk.async_typer import AsyncTyper
+from infrahub_sdk.ctl.client import initialize_client
+from infrahub_sdk.ctl.exceptions import FileNotValidError
+from infrahub_sdk.ctl.utils import catch_exception, init_logging
+from infrahub_sdk.spec.object import ObjectFile
+
+from .parameters import CONFIG_PARAM
+from .utils import load_yamlfile_from_disk_and_exit
+
+app = AsyncTyper()
+console = Console()
+
+
+@app.callback()
+def callback() -> None:
+    """
+    Manage objects in a remote Infrahub instance.
+    """
+
+
+@app.command()
+@catch_exception(console=console)
+async def load(
+    paths: list[Path],
+    debug: bool = False,
+    branch: str = typer.Option("main", help="Branch on which to load the objects."),
+    _: str = CONFIG_PARAM,
+) -> None:
+    """Load one or multiple objects files into Infrahub."""
+
+    init_logging(debug=debug)
+
+    files = load_yamlfile_from_disk_and_exit(paths=paths, file_type=ObjectFile, console=console)
+    client = await initialize_client()
+
+    for file in files:
+        file.validate_content()
+        if not file.spec.kind:
+            raise FileNotValidError(name=str(file.location), message="kind must be specified.")
+        schema = await client.schema.get(kind=file.spec.kind, branch=branch)
+
+        for item in file.spec.data:
+            await file.spec.create_node(client=client, schema=schema, data=item, branch=branch)
diff --git a/infrahub_sdk/spec/__init__.py b/infrahub_sdk/spec/__init__.py
diff --git a/infrahub_sdk/spec/menu.py b/infrahub_sdk/spec/menu.py
@@ -0,0 +1,29 @@
+from typing import Optional
+
+from infrahub_sdk.yaml import InfrahubFile, InfrahubFileKind
+
+from .object import InfrahubObjectFileData
+
+
+class InfrahubMenuFileData(InfrahubObjectFileData):
+    @classmethod
+    def enrich_node(cls, data: dict) -> dict:
+        if "kind" in data and "path" not in data:
+            data["path"] = "/objects/" + data["kind"]
+        return data
+
+
+class MenuFile(InfrahubFile):
+    _spec: Optional[InfrahubMenuFileData] = None
+
+    @property
+    def spec(self) -> InfrahubMenuFileData:
+        if not self._spec:
+            self._spec = InfrahubMenuFileData(**self.data.spec)
+        return self._spec
+
+    def validate_content(self) -> None:
+        super().validate_content()
+        if self.kind != InfrahubFileKind.MENU:
+            raise ValueError("File is not an Infrahub Menu file")
+        self._spec = InfrahubMenuFileData(**self.data.spec)
diff --git a/infrahub_sdk/spec/object.py b/infrahub_sdk/spec/object.py
@@ -0,0 +1,121 @@
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+from infrahub_sdk.client import InfrahubClient
+from infrahub_sdk.schema import MainSchemaTypes
+from infrahub_sdk.yaml import InfrahubFile, InfrahubFileKind
+
+
+class InfrahubObjectFileData(BaseModel):
+    kind: str | None = None
+    data: list[dict[str, Any]] = Field(default_factory=list)
+
+    @classmethod
+    def enrich_node(cls, data: dict) -> dict:
+        return data
+
+    @classmethod
+    async def create_node(
+        cls,
+        client: InfrahubClient,
+        schema: MainSchemaTypes,
+        data: dict,
+        context: Optional[dict] = None,
+        branch: Optional[str] = None,
+        default_schema_kind: Optional[str] = None,
+    ) -> None:
+        # First validate of all mandatory fields are present
+        for element in schema.mandatory_attribute_names + schema.mandatory_relationship_names:
+            if element not in data.keys():
+                raise ValueError(f"{element} is mandatory")
+
+        clean_data: dict[str, Any] = {}
+
+        remaining_rels = []
+        for key, value in data.items():
+            if key in schema.attribute_names:
+                # NOTE we could validate the format of the data but the API will do it as well
+                clean_data[key] = value
+
+            if key in schema.relationship_names:
+                rel_schema = schema.get_relationship(name=key)
+
+                if not isinstance(value, dict) and "data" in value:
+                    raise ValueError(f"relationship {key} must be a dict with 'data'")
+
+                if rel_schema.cardinality == "one" or rel_schema.optional is False:
+                    raise ValueError(
+                        "Not supported yet, we need to have a way to define connect object before they exist"
+                    )
+                    # clean_data[key] = value[data]
+                remaining_rels.append(key)
+
+        if context:
+            clean_data.update(context)
+
+        clean_data = cls.enrich_node(data=clean_data)
+
+        node = await client.create(kind=schema.kind, branch=branch, data=clean_data)
+        await node.save(allow_upsert=True)
+        print(f"Created Node: {node.get_human_friendly_id_as_string()}")
+
+        for rel in remaining_rels:
+            # identify what is the name of the relationship on the other side
+            if not isinstance(data[rel], dict) and "data" in data[rel]:
+                raise ValueError(f"relationship {rel} must be a dict with 'data'")
+
+            rel_schema = schema.get_relationship(name=rel)
+            peer_kind = data[rel].get("kind", default_schema_kind) or rel_schema.peer
+            peer_schema = await client.schema.get(kind=peer_kind, branch=branch)
+
+            if rel_schema.identifier is None:
+                raise ValueError("identifier must be defined")
+
+            peer_rel = peer_schema.get_relationship_by_identifier(id=rel_schema.identifier)
+
+            rel_data = data[rel]["data"]
+            context = {}
+            if peer_rel:
+                context[peer_rel.name] = node.id
+
+            if rel_schema.cardinality == "one" and isinstance(rel_data, dict):
+                await cls.create_node(
+                    client=client,
+                    schema=peer_schema,
+                    data=rel_data,
+                    context=context,
+                    branch=branch,
+                    default_schema_kind=default_schema_kind,
+                )
+
+            elif rel_schema.cardinality == "many" and isinstance(rel_data, list):
+                for peer_data in rel_data:
+                    await cls.create_node(
+                        client=client,
+                        schema=peer_schema,
+                        data=peer_data,
+                        context=context,
+                        branch=branch,
+                        default_schema_kind=default_schema_kind,
+                    )
+            else:
+                raise ValueError(
+                    f"Relationship {rel_schema.name} doesn't have the right format {rel_schema.cardinality} / {type(rel_data)}"
+                )
+
+
+class ObjectFile(InfrahubFile):
+    _spec: Optional[InfrahubObjectFileData] = None
+
+    @property
+    def spec(self) -> InfrahubObjectFileData:
+        if not self._spec:
+            self._spec = InfrahubObjectFileData(**self.data.spec)
+        return self._spec
+
+    def validate_content(self) -> None:
+        super().validate_content()
+        if self.kind != InfrahubFileKind.OBJECT:
+            raise ValueError("File is not an Infrahub Object file")
+        self._spec = InfrahubObjectFileData(**self.data.spec)
