Add infrahubctl menu load command to load menu items from a file

Author: dgarros

infrahub_sdk/ctl/cli_commands.py

@@ -22,13 +22,19 @@
 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
 from infrahub_sdk.ctl.schema import app as schema_app
-from infrahub_sdk.ctl.schema import load_schemas_from_disk_and_exit
 from infrahub_sdk.ctl.transform import list_transforms
-from infrahub_sdk.ctl.utils import catch_exception, execute_graphql_query, parse_cli_vars
+from infrahub_sdk.ctl.utils import (
+    catch_exception,
+    execute_graphql_query,
+    load_yamlfile_from_disk_and_exit,
+    parse_cli_vars,
+)
 from infrahub_sdk.ctl.validate import app as validate_app
 from infrahub_sdk.exceptions import GraphQLError, InfrahubTransformNotFoundError
 from infrahub_sdk.jinja2 import identify_faulty_jinja_code
@@ -39,6 +45,7 @@
 )
 from infrahub_sdk.transforms import get_transform_class_instance
 from infrahub_sdk.utils import get_branch, write_to_file
+from infrahub_sdk.yaml import SchemaFile
 
 from .exporter import dump
 from .importer import load
@@ -50,6 +57,9 @@
 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", hidden=True)
+
 app.command(name="dump")(dump)
 app.command(name="load")(load)
 
@@ -338,7 +348,7 @@ def protocols(  # noqa: PLR0915
     schema: dict[str, MainSchemaTypes] = {}
 
     if schemas:
-        schemas_data = load_schemas_from_disk_and_exit(schemas=schemas)
+        schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
 
         for data in schemas_data:
             data.load_content()

infrahub_sdk/ctl/menu.py

@@ -0,0 +1,60 @@
+import logging
+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)
+
+    logging.getLogger("infrahub_sdk").setLevel(logging.INFO)
+
+    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 idx, item in enumerate(file.spec.data):
+            await file.spec.create_node(
+                client=client,
+                schema=schema,
+                data=item,
+                branch=branch,
+                default_schema_kind=default_kind,
+                context={"list_index": idx},
+            )

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)

infrahub_sdk/ctl/schema.py

@@ -11,13 +11,12 @@
 from infrahub_sdk import InfrahubClient
 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.queries import SCHEMA_HASH_SYNC_STATUS
-from infrahub_sdk.utils import find_files
 from infrahub_sdk.yaml import SchemaFile
 
 from .parameters import CONFIG_PARAM
+from .utils import load_yamlfile_from_disk_and_exit
 
 app = AsyncTyper()
 console = Console()
@@ -30,45 +29,6 @@ def callback() -> None:
     """
 
 
-def load_schemas_from_disk(schemas: list[Path]) -> list[SchemaFile]:
-    schemas_data: list[SchemaFile] = []
-    for schema in schemas:
-        if schema.is_file():
-            schema_file = SchemaFile(location=schema)
-            schema_file.load_content()
-            schemas_data.append(schema_file)
-        elif schema.is_dir():
-            files = find_files(extension=["yaml", "yml", "json"], directory=schema)
-            for item in files:
-                schema_file = SchemaFile(location=item)
-                schema_file.load_content()
-                schemas_data.append(schema_file)
-        else:
-            raise FileNotValidError(name=schema, message=f"Schema path: {schema} does not exist!")
-
-    return schemas_data
-
-
-def load_schemas_from_disk_and_exit(schemas: list[Path]) -> list[SchemaFile]:
-    has_error = False
-    try:
-        schemas_data = load_schemas_from_disk(schemas=schemas)
-    except FileNotValidError as exc:
-        console.print(f"[red]{exc.message}")
-        raise typer.Exit(1) from exc
-
-    for schema_file in schemas_data:
-        if schema_file.valid and schema_file.content:
-            continue
-        console.print(f"[red]{schema_file.error_message} ({schema_file.location})")
-        has_error = True
-
-    if has_error:
-        raise typer.Exit(1)
-
-    return schemas_data
-
-
 def validate_schema_content_and_exit(client: InfrahubClient, schemas: list[SchemaFile]) -> None:
     has_error: bool = False
     for schema_file in schemas:
@@ -153,7 +113,7 @@ async def load(
 
     init_logging(debug=debug)
 
-    schemas_data = load_schemas_from_disk_and_exit(schemas=schemas)
+    schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
     schema_definition = "schema" if len(schemas_data) == 1 else "schemas"
     client = await initialize_client()
     validate_schema_content_and_exit(client=client, schemas=schemas_data)
@@ -203,7 +163,7 @@ async def check(
 
     init_logging(debug=debug)
 
-    schemas_data = load_schemas_from_disk_and_exit(schemas=schemas)
+    schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
     client = await initialize_client()
     validate_schema_content_and_exit(client=client, schemas=schemas_data)
 

infrahub_sdk/ctl/utils.py

@@ -3,7 +3,7 @@
 import traceback
 from functools import wraps
 from pathlib import Path
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, TypeVar, Union
 
 import pendulum
 import typer
@@ -14,7 +14,7 @@
 from rich.logging import RichHandler
 from rich.markup import escape
 
-from infrahub_sdk.ctl.exceptions import QueryNotFoundError
+from infrahub_sdk.ctl.exceptions import FileNotValidError, QueryNotFoundError
 from infrahub_sdk.exceptions import (
     AuthenticationError,
     Error,
@@ -25,9 +25,12 @@
     ServerNotResponsiveError,
 )
 from infrahub_sdk.schema import InfrahubRepositoryConfig
+from infrahub_sdk.yaml import YamlFile
 
 from .client import initialize_client_sync
 
+YamlFileVar = TypeVar("YamlFileVar", bound=YamlFile)
+
 
 def init_logging(debug: bool = False) -> None:
     logging.getLogger("infrahub_sdk").setLevel(logging.CRITICAL)
@@ -179,3 +182,25 @@ def get_fixtures_dir() -> Path:
     """Get the directory which stores fixtures that are common to multiple unit/integration tests."""
     here = Path(__file__).resolve().parent
     return here.parent.parent / "tests" / "fixtures"
+
+
+def load_yamlfile_from_disk_and_exit(
+    paths: list[Path], file_type: type[YamlFileVar], console: Console
+) -> list[YamlFileVar]:
+    has_error = False
+    try:
+        data_files = file_type.load_from_disk(paths=paths)
+    except FileNotValidError as exc:
+        console.print(f"[red]{exc.message}")
+        raise typer.Exit(1) from exc
+
+    for data_file in data_files:
+        if data_file.valid and data_file.content:
+            continue
+        console.print(f"[red]{data_file.error_message} ({data_file.location})")
+        has_error = True
+
+    if has_error:
+        raise typer.Exit(1)
+
+    return data_files

infrahub_sdk/protocols.py

@@ -143,6 +143,7 @@ class CoreMenu(CoreNode):
     namespace: String
     name: String
     label: StringOptional
+    kind: StringOptional
     path: StringOptional
     description: StringOptional
     icon: StringOptional
@@ -607,6 +608,7 @@ class CoreMenuSync(CoreNodeSync):
     namespace: String
     name: String
     label: StringOptional
+    kind: StringOptional
     path: StringOptional
     description: StringOptional
     icon: StringOptional

infrahub_sdk/spec/__init__.py

@@ -0,0 +1,33 @@
+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, context: dict) -> dict:
+        if "kind" in data and "path" not in data:
+            data["path"] = "/objects/" + data["kind"]
+
+        if "list_index" in context and "order_weight" not in data:
+            data["order_weight"] = (context["list_index"] + 1) * 1000
+
+        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)