Cont working on the new menu in the backend

Author: dgarros

backend/infrahub/api/menu.py

@@ -3,251 +3,35 @@
 from typing import TYPE_CHECKING
 
 from fastapi import APIRouter, Depends
-from pydantic import BaseModel, Field
 
 from infrahub.api.dependencies import get_branch_dep, get_current_user, get_db
 from infrahub.core import registry
 from infrahub.core.branch import Branch  # noqa: TCH001
-from infrahub.core.constants import InfrahubKind
 from infrahub.core.protocols import CoreMenuItem
-from infrahub.core.schema import NodeSchema
 from infrahub.log import get_logger
 from infrahub.menu.generator import generate_menu
 from infrahub.menu.models import Menu  # noqa: TCH001
 
 if TYPE_CHECKING:
     from infrahub.auth import AccountSession
-    from infrahub.core.schema import MainSchemaTypes
     from infrahub.database import InfrahubDatabase
 
 
 log = get_logger()
 router = APIRouter(prefix="/menu")
 
 
-class InterfaceMenu(BaseModel):
-    title: str = Field(..., description="Title of the menu item")
-    path: str = Field(default="", description="URL endpoint if applicable")
-    icon: str = Field(default="", description="The icon to show for the current view")
-    children: list[InterfaceMenu] = Field(default_factory=list, description="Child objects")
-    kind: str = Field(default="")
-
-    def __lt__(self, other: object) -> bool:
-        if not isinstance(other, InterfaceMenu):
-            raise NotImplementedError
-        return self.title < other.title
-
-    def list_title(self) -> str:
-        return f"All {self.title}(s)"
-
-
-def add_to_menu(structure: dict[str, list[InterfaceMenu]], menu_item: InterfaceMenu) -> None:
-    all_items = InterfaceMenu(title=menu_item.list_title(), path=menu_item.path, icon=menu_item.icon)
-    menu_item.path = ""
-    menu_item.icon = ""
-    for child in structure[menu_item.kind]:
-        menu_item.children.append(child)
-        if child.kind in structure:
-            add_to_menu(structure, child)
-    menu_item.children.sort()
-    menu_item.children.insert(0, all_items)
-
-
-def _extract_node_icon(model: MainSchemaTypes) -> str:
-    if not model.icon:
-        return ""
-    return model.icon
-
-
 @router.get("")
-async def get_menu(branch: Branch = Depends(get_branch_dep)) -> list[InterfaceMenu]:
-    log.info("menu_request", branch=branch.name)
-
-    full_schema = registry.schema.get_full(branch=branch, duplicate=False)
-    objects = InterfaceMenu(title="Objects", children=[])
-
-    structure: dict[str, list[InterfaceMenu]] = {}
-
-    ipam = InterfaceMenu(
-        title="IPAM",
-        children=[
-            InterfaceMenu(
-                title="Namespaces",
-                path=f"/objects/{InfrahubKind.IPNAMESPACE}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.IPNAMESPACE]),
-            ),
-            InterfaceMenu(
-                title="IP Prefixes", path="/ipam/prefixes", icon=_extract_node_icon(full_schema[InfrahubKind.IPPREFIX])
-            ),
-            InterfaceMenu(
-                title="IP Addresses",
-                path="/ipam/addresses?ipam-tab=ip-details",
-                icon=_extract_node_icon(full_schema[InfrahubKind.IPADDRESS]),
-            ),
-        ],
-    )
-
-    has_ipam = False
-
-    for key in full_schema.keys():
-        model = full_schema[key]
-
-        if isinstance(model, NodeSchema) and (
-            InfrahubKind.IPADDRESS in model.inherit_from or InfrahubKind.IPPREFIX in model.inherit_from
-        ):
-            has_ipam = True
-
-        if not model.include_in_menu:
-            continue
-
-        menu_name = model.menu_placement or "base"
-        if menu_name not in structure:
-            structure[menu_name] = []
-
-        structure[menu_name].append(
-            InterfaceMenu(title=model.menu_title, path=f"/objects/{model.kind}", icon=model.icon or "", kind=model.kind)
-        )
-
-    for menu_item in structure["base"]:
-        if menu_item.kind in structure:
-            add_to_menu(structure, menu_item)
-
-        objects.children.append(menu_item)
-
-    objects.children.sort()
-    groups = InterfaceMenu(
-        title="Object Management",
-        children=[
-            InterfaceMenu(
-                title="All Groups",
-                path=f"/objects/{InfrahubKind.GENERICGROUP}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.GENERICGROUP]),
-            ),
-            InterfaceMenu(
-                title="All Profiles",
-                path=f"/objects/{InfrahubKind.PROFILE}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.PROFILE]),
-            ),
-            InterfaceMenu(
-                title="Resource Manager",
-                path="/resource-manager",
-                icon=_extract_node_icon(full_schema[InfrahubKind.RESOURCEPOOL]),
-            ),
-        ],
-    )
-
-    unified_storage = InterfaceMenu(
-        title="Unified Storage",
-        children=[
-            InterfaceMenu(title="Schema", path="/schema", icon="mdi:file-code"),
-            InterfaceMenu(
-                title="Repository",
-                path=f"/objects/{InfrahubKind.GENERICREPOSITORY}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.GENERICREPOSITORY]),
-            ),
-            InterfaceMenu(
-                title="GraphQL Query",
-                path=f"/objects/{InfrahubKind.GRAPHQLQUERY}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.GRAPHQLQUERY]),
-            ),
-        ],
-    )
-    change_control = InterfaceMenu(
-        title="Change Control",
-        children=[
-            InterfaceMenu(title="Branches", path="/branches", icon="mdi:layers-triple"),
-            InterfaceMenu(
-                title="Proposed Changes",
-                path="/proposed-changes",
-                icon=_extract_node_icon(full_schema[InfrahubKind.PROPOSEDCHANGE]),
-            ),
-            InterfaceMenu(
-                title="Check Definition",
-                path=f"/objects/{InfrahubKind.CHECKDEFINITION}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.CHECKDEFINITION]),
-            ),
-            InterfaceMenu(title="Tasks", path="/tasks", icon="mdi:shield-check"),
-        ],
-    )
-    deployment = InterfaceMenu(
-        title="Deployment",
-        children=[
-            InterfaceMenu(
-                title="Artifact",
-                path=f"/objects/{InfrahubKind.ARTIFACT}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.ARTIFACT]),
-            ),
-            InterfaceMenu(
-                title="Artifact Definition",
-                path=f"/objects/{InfrahubKind.ARTIFACTDEFINITION}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.ARTIFACTDEFINITION]),
-            ),
-            InterfaceMenu(
-                title="Generator Definition",
-                path=f"/objects/{InfrahubKind.GENERATORDEFINITION}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.GENERATORDEFINITION]),
-            ),
-            InterfaceMenu(
-                title="Generator Instance",
-                path=f"/objects/{InfrahubKind.GENERATORINSTANCE}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.GENERATORINSTANCE]),
-            ),
-            InterfaceMenu(
-                title="Transformation",
-                path=f"/objects/{InfrahubKind.TRANSFORM}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.TRANSFORM]),
-            ),
-        ],
-    )
-
-    admin = InterfaceMenu(
-        title="Admin",
-        children=[
-            InterfaceMenu(
-                title="Role Management",
-                path="/role-management",
-                icon=_extract_node_icon(full_schema[InfrahubKind.BASEPERMISSION]),
-            ),
-            InterfaceMenu(
-                title="Credentials",
-                path=f"/objects/{InfrahubKind.CREDENTIAL}",
-                icon=_extract_node_icon(full_schema[InfrahubKind.CREDENTIAL]),
-            ),
-            InterfaceMenu(
-                title="Webhooks",
-                children=[
-                    InterfaceMenu(
-                        title="Webhook",
-                        path=f"/objects/{InfrahubKind.STANDARDWEBHOOK}",
-                        icon=_extract_node_icon(full_schema[InfrahubKind.STANDARDWEBHOOK]),
-                    ),
-                    InterfaceMenu(
-                        title="Custom Webhook",
-                        path=f"/objects/{InfrahubKind.CUSTOMWEBHOOK}",
-                        icon=_extract_node_icon(full_schema[InfrahubKind.CUSTOMWEBHOOK]),
-                    ),
-                ],
-            ),
-        ],
-    )
-
-    menu_items = [objects]
-    if has_ipam:
-        menu_items.append(ipam)
-    menu_items.extend([groups, unified_storage, change_control, deployment, admin])
-
-    return menu_items
-
-
-@router.get("/new")
-async def get_new_menu(
+async def get_menu(
     db: InfrahubDatabase = Depends(get_db),
     branch: Branch = Depends(get_branch_dep),
     account_session: AccountSession = Depends(get_current_user),
 ) -> Menu:
-    log.info("new_menu_request", branch=branch.name)
+    log.info("menu_request", branch=branch.name)
 
-    menu_items = await registry.manager.query(db=db, schema=CoreMenuItem, branch=branch, prefetch_relationships=True)
+    menu_items = await registry.manager.query(
+        db=db, schema=CoreMenuItem, branch=branch
+    )  # , prefetch_relationships=True)
     menu = await generate_menu(db=db, branch=branch, account=account_session, menu_items=menu_items)
 
     return menu.to_rest()

backend/infrahub/core/initialization.py

@@ -20,15 +20,15 @@
 from infrahub.core.node.resource_manager.ip_address_pool import CoreIPAddressPool
 from infrahub.core.node.resource_manager.ip_prefix_pool import CoreIPPrefixPool
 from infrahub.core.node.resource_manager.number_pool import CoreNumberPool
-from infrahub.core.protocols import CoreAccount, CoreMenuItem
+from infrahub.core.protocols import CoreAccount
 from infrahub.core.root import Root
 from infrahub.core.schema import SchemaRoot, core_models, internal_schema
 from infrahub.core.schema.manager import SchemaManager
 from infrahub.database import InfrahubDatabase
 from infrahub.exceptions import DatabaseError
 from infrahub.log import get_logger
 from infrahub.menu.menu import default_menu
-from infrahub.menu.models import MenuItemDefinition
+from infrahub.menu.utils import create_menu_children
 from infrahub.permissions import PermissionBackend
 from infrahub.storage import InfrahubObjectStorage
 from infrahub.utils import format_label
@@ -307,14 +307,6 @@ async def create_initial_permission(db: InfrahubDatabase) -> Node:
     return permission
 
 
-async def create_menu_children(db: InfrahubDatabase, parent: CoreMenuItem, children: list[MenuItemDefinition]) -> None:
-    for child in children:
-        obj = await child.to_node(db=db, parent=parent)
-        await obj.save(db=db)
-        if child.children:
-            await create_menu_children(db=db, parent=obj, children=child.children)
-
-
 async def create_default_menu(db: InfrahubDatabase) -> None:
     for item in default_menu:
         obj = await item.to_node(db=db)

backend/infrahub/core/protocols.py

@@ -137,6 +137,7 @@ class CoreMenu(CoreNode):
     namespace: String
     name: String
     label: StringOptional
+    kind: StringOptional
     path: StringOptional
     description: StringOptional
     icon: StringOptional

backend/infrahub/core/schema/definitions/core.py

@@ -67,11 +67,12 @@
     "description": "Base node for the menu",
     "label": "Menu Item",
     "hierarchical": True,
-    "uniqueness_constraints": [["namespace__value", "name__value"]],
+    "human_friendly_id": ["namespace__value", "name__value"],
     "attributes": [
         {"name": "namespace", "kind": "Text", "regex": NAMESPACE_REGEX, "order_weight": 1000},
         {"name": "name", "kind": "Text", "order_weight": 1000},
         {"name": "label", "kind": "Text", "optional": True, "order_weight": 2000},
+        {"name": "kind", "kind": "Text", "optional": True, "order_weight": 2500},
         {"name": "path", "kind": "Text", "optional": True, "order_weight": 2500},
         {"name": "description", "kind": "Text", "optional": True, "order_weight": 3000},
         {"name": "icon", "kind": "Text", "optional": True, "order_weight": 4000},

backend/infrahub/core/schema/schema_branch.py

@@ -486,7 +486,6 @@ def process_pre_validation(self) -> None:
 
     def process_validate(self) -> None:
         self.validate_names()
-        self.validate_menu_placements()
         self.validate_kinds()
         self.validate_default_values()
         self.validate_count_against_cardinality()
@@ -899,28 +898,6 @@ def validate_names(self) -> None:
                 ):
                     raise ValueError(f"{node.kind}: {rel.name} isn't allowed as a relationship name.")
 
-    def validate_menu_placements(self) -> None:
-        menu_placements: dict[str, str] = {}
-
-        for name in list(self.nodes.keys()) + list(self.generics.keys()):
-            node = self.get(name=name, duplicate=False)
-            if node.menu_placement:
-                try:
-                    placement_node = self.get(name=node.menu_placement, duplicate=False)
-                except SchemaNotFoundError as exc:
-                    raise SchemaNotFoundError(
-                        branch_name=self.name,
-                        identifier=node.menu_placement,
-                        message=f"{node.kind} refers to an invalid menu placement node: {node.menu_placement}.",
-                    ) from exc
-                if node == placement_node:
-                    raise ValueError(f"{node.kind}: cannot be placed under itself in the menu") from None
-
-                if menu_placements.get(placement_node.kind) == node.kind:
-                    raise ValueError(f"{node.kind}: cyclic menu placement with {placement_node.kind}") from None
-
-                menu_placements[node.kind] = placement_node.kind
-
     def validate_kinds(self) -> None:
         for name in list(self.nodes.keys()):
             node = self.get_node(name=name, duplicate=False)

backend/infrahub/menu/generator.py

@@ -32,31 +32,29 @@ async def generate_menu(
     full_schema = registry.schema.get_full(branch=branch, duplicate=False)
 
     already_processed = []
-    havent_been_processed = []
 
     # Process the parent first
     for item in menu_items:
         full_name = get_full_name(item)
-        parent = await item.parent.get_peer(db=db, peer_type=CoreMenuItem)
-        if parent:
-            havent_been_processed.append(full_name)
+        parent1 = await item.parent.get_peer(db=db, peer_type=CoreMenuItem)
+        if parent1:
             continue
         structure.data[full_name] = MenuItemDict.from_node(obj=item)
         already_processed.append(full_name)
 
     # Process the children
-    havent_been_processed = []
+    havent_been_processed: list[str] = []
     for item in menu_items:
         full_name = get_full_name(item)
         if full_name in already_processed:
             continue
 
-        parent = await item.parent.get_peer(db=db, peer_type=CoreMenuItem)
-        if not parent:
+        parent2 = await item.parent.get_peer(db=db, peer_type=CoreMenuItem)
+        if not parent2:
             havent_been_processed.append(full_name)
             continue
 
-        parent_full_name = get_full_name(parent)
+        parent_full_name = get_full_name(parent2)
         menu_item = structure.find_item(name=parent_full_name)
         if menu_item:
             child_item = MenuItemDict.from_node(obj=item)
@@ -65,8 +63,8 @@ async def generate_menu(
             log.warning(
                 "new_menu_request: unable to find the parent menu item",
                 branch=branch.name,
-                menu_item=item.name.value,
-                parent_item=parent.name.value,
+                menu_item=child_item.identifier,
+                parent_item=parent_full_name,
             )
 
     default_menu = structure.find_item(name=FULL_DEFAULT_MENU)