feat(db): rewrite inventory service part 1 - helpers and tree

backend/app/services/inventory.py

@@ -1,101 +1,129 @@
+import json
 from collections import defaultdict
 
+from sqlalchemy.orm import Session
+
+from app.models import NetObjectModel, RelationModel
 from app.schemas import InventoryTreeNode, NetObject, NodeContext, NodeContextRelations, Relation
-from app.services.store import OBJECTS, RELATIONS
 
 
-def get_object_map() -> dict[str, NetObject]:
+def _to_net_object(m: NetObjectModel) -> NetObject:
-    return {item.id: item for item in OBJECTS}
+    return NetObject(
+        id=m.id, name=m.name, slug=m.slug, type=m.type,
+        parent_id=m.parent_id, status=m.status, vendor=m.vendor,
+        model=m.model, serial=m.serial, notes=m.notes,
+        created_at=m.created_at, updated_at=m.updated_at,
+    )
 
 
-def get_children_map() -> dict[str, list[NetObject]]:
+def _to_relation(m: RelationModel) -> Relation:
-    object_map = get_object_map()
+    return Relation(
+        id=m.id, type=m.type, layer=m.layer,
+        source_id=m.source_id, target_id=m.target_id,
+        direction=m.direction, source=m.source,
+        confidence=m.confidence, status=m.status,
+        attributes=json.loads(m.attributes_json or "{}"),
+        created_at=m.created_at, updated_at=m.updated_at,
+    )
+
+
+def get_object_map(db: Session) -> dict[str, NetObject]:
+    return {r.id: _to_net_object(r) for r in db.query(NetObjectModel).all()}
+
+
+def get_children_map(db: Session) -> dict[str, list[NetObject]]:
+    object_map = get_object_map(db)
+    rels = db.query(RelationModel).filter(
+        RelationModel.type == "contains",
+        RelationModel.status == "active",
+    ).all()
     children: dict[str, list[NetObject]] = defaultdict(list)
-    for rel in RELATIONS:
+    for rel in rels:
-        if rel.type == 'contains' and rel.status == 'active' and rel.target_id in object_map:
+        if rel.target_id in object_map:
             children[rel.source_id].append(object_map[rel.target_id])
     return children
 
 
-def get_badges(object_id: str) -> list[str]:
+def get_badges(db: Session, object_id: str) -> list[str]:
-    obj = get_object_map()[object_id]
+    obj = get_object_map(db).get(object_id)
+    if not obj:
+        return []
     badges: list[str] = []
-    if obj.type == 'physical_host':
+    if obj.type == "physical_host":
-        badges.append('virtualization-host')
+        badges.append("virtualization-host")
-    if obj.type in {'vm', 'router', 'firewall'}:
+    if obj.type in {"vm", "router", "firewall"}:
-        badges.append('network-node')
+        badges.append("network-node")
-    outgoing = [r for r in RELATIONS if r.source_id == object_id and r.status == 'active']
+    outgoing = db.query(RelationModel).filter(
-    if any(r.type == 'routes_via' for r in outgoing):
+        RelationModel.source_id == object_id,
-        badges.append('gateway-dependent')
+        RelationModel.status == "active",
-    if any(r.type in {'routes_via', 'bridges_to'} for r in outgoing) or obj.name.lower().startswith('openwrt'):
+    ).all()
-        badges.append('gateway')
+    if any(r.type == "routes_via" for r in outgoing):
+        badges.append("gateway-dependent")
+    if any(r.type in {"routes_via", "bridges_to"} for r in outgoing) \
+            or obj.name.lower().startswith("openwrt"):
+        badges.append("gateway")
     return badges
 
 
-def build_tree_node(object_id: str, depth: int | None = None) -> InventoryTreeNode:
+def build_tree_node(db: Session, object_id: str, depth: int | None = None) -> InventoryTreeNode:
-    object_map = get_object_map()
+    om = get_object_map(db)
-    children_map = get_children_map()
+    cm = get_children_map(db)
-    obj = object_map[object_id]
+    obj = om[object_id]
-    child_objects = children_map.get(object_id, [])
+    ch = cm.get(object_id, [])
     child_nodes = []
     if depth is None or depth > 0:
-        next_depth = None if depth is None else depth - 1
+        nd = None if depth is None else depth - 1
-        child_nodes = [build_tree_node(child.id, next_depth) for child in child_objects]
+        child_nodes = [build_tree_node(db, c.id, nd) for c in ch]
     return InventoryTreeNode(
-        id=obj.id,
+        id=obj.id, name=obj.name, type=obj.type, status=obj.status,
-        name=obj.name,
+        badges=get_badges(db, obj.id),
-        type=obj.type,
+        children_count=len(ch),
-        status=obj.status,
-        badges=get_badges(obj.id),
-        children_count=len(child_objects),
         children=child_nodes,
     )
 
 
-def build_inventory_tree(root_id: str | None = None, depth: int | None = None) -> list[InventoryTreeNode]:
+def build_inventory_tree(
-    object_map = get_object_map()
+    db: Session, root_id: str | None = None, depth: int | None = None
+) -> list[InventoryTreeNode]:
+    om = get_object_map(db)
     if root_id:
-        return [build_tree_node(root_id, depth)] if root_id in object_map else []
+        return [build_tree_node(db, root_id, depth)] if root_id in om else []
-    roots = [obj for obj in OBJECTS if obj.parent_id is None]
+    roots = db.query(NetObjectModel).filter(NetObjectModel.parent_id == None).all()  # noqa: E711
-    return [build_tree_node(obj.id, depth) for obj in roots]
+    return [build_tree_node(db, r.id, depth) for r in roots]
 
 
-def get_parent(item: NetObject) -> NetObject | None:
+def build_node_context(db: Session, item_id: str) -> NodeContext | None:
-    object_map = get_object_map()
+    row = db.query(NetObjectModel).filter(NetObjectModel.id == item_id).first()
-    if not item.parent_id:
+    if not row:
-        return None
-    return object_map.get(item.parent_id)
-
-
-def get_children(item_id: str) -> list[NetObject]:
-    return get_children_map().get(item_id, [])
-
-
-def get_relations(item_id: str) -> NodeContextRelations:
-    incoming = [rel for rel in RELATIONS if rel.target_id == item_id and rel.status == 'active']
-    outgoing = [rel for rel in RELATIONS if rel.source_id == item_id and rel.status == 'active']
-    return NodeContextRelations(incoming=incoming, outgoing=outgoing)
-
-
-def build_node_context(item_id: str) -> NodeContext | None:
-    object_map = get_object_map()
-    item = object_map.get(item_id)
-    if not item:
         return None
+    item = _to_net_object(row)
+    parent = None
+    if item.parent_id:
+        p = db.query(NetObjectModel).filter(NetObjectModel.id == item.parent_id).first()
+        parent = _to_net_object(p) if p else None
+    children = get_children_map(db).get(item_id, [])
+    inc = [_to_relation(r) for r in db.query(RelationModel).filter(
+        RelationModel.target_id == item_id, RelationModel.status == "active").all()]
+    out = [_to_relation(r) for r in db.query(RelationModel).filter(
+        RelationModel.source_id == item_id, RelationModel.status == "active").all()]
     return NodeContext(
-        item=item,
+        item=item, parent=parent, children=children,
-        parent=get_parent(item),
+        relations=NodeContextRelations(incoming=inc, outgoing=out),
-        children=get_children(item_id),
-        relations=get_relations(item_id),
     )
 
 
-def get_node_relations(item_id: str, layer: str | None = None, rel_type: str | None = None, direction: str = 'both') -> list[Relation]:
+def get_node_relations(
+    db: Session, item_id: str,
+    layer: str | None = None,
+    rel_type: str | None = None,
+    direction: str = "both",
+) -> list[Relation]:
     result: list[Relation] = []
-    if direction in {'incoming', 'both'}:
+    base = db.query(RelationModel).filter(RelationModel.status == "active")
-        result.extend([r for r in RELATIONS if r.target_id == item_id and r.status == 'active'])
+    if direction in {"incoming", "both"}:
-    if direction in {'outgoing', 'both'}:
+        result += [_to_relation(r) for r in base.filter(RelationModel.target_id == item_id).all()]
-        result.extend([r for r in RELATIONS if r.source_id == item_id and r.status == 'active'])
+    if direction in {"outgoing", "both"}:
+        result += [_to_relation(r) for r in base.filter(RelationModel.source_id == item_id).all()]
     if layer:
         result = [r for r in result if r.layer == layer]
     if rel_type: