Coverage for src/lexigram/admin/navigation/manager.py: 81%
111 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Per-request navigation mount for the admin shell.
3:class:`NavigationManager` is the single per-request entry point for
4navigation state: it resolves the primary nav (resource items + assembler
5contributions with dedup and per-request active state), the cluster
6secondary sidebar for the active cluster center, and the user-menu entries
7(cluster landings + settings/plugins) as :class:`MenuItem` values.
9Everything is derived from request-scoped app state, so a fresh manager is
10built per request — the mount lifecycle.
11"""
13from __future__ import annotations
15from typing import Any
17from lexigram.admin.clusters import ClusterRegistry
18from lexigram.admin.navigation.types import MenuItem
20__all__ = ["NavigationManager"]
22_MENU_PROFILE = MenuItem(label="Profile", href="/admin/profile", icon="user-circle")
23_MENU_SETTINGS = MenuItem(label="Settings", href="/admin/settings", icon="settings")
24_MENU_PLUGINS = MenuItem(label="Plugins", href="/admin/plugins", icon="plugins")
27class NavigationManager:
28 """Per-request navigation state for the admin panel.
30 Example:
31 ```python
32 nav_items, system_items, cluster_nav = NavigationManager(
33 request
34 ).resolve_nav()
35 menu = NavigationManager(request).user_menu_items()
36 ```
37 """
39 def __init__(self, request: Any) -> None:
40 """Read app state relevant to navigation from the request.
42 Args:
43 request: Current Starlette request (its app carries the admin
44 state: nav builder, assembler groups, cluster registry).
45 """
46 self._request = request
47 state = getattr(request, "app", None) if request else None
48 self._state = getattr(state, "state", None) if state else None
49 self._nav_builder = (
50 getattr(self._state, "nav_builder", None) if self._state else None
51 )
52 self._assembler_nav_items: list[dict] = (
53 list(getattr(self._state, "assembler_nav_items", None) or [])
54 if self._state
55 else []
56 )
57 self._assembler_groups: dict[str, Any] | None = (
58 getattr(self._state, "assembler_groups", None) or None
59 if self._state
60 else None
61 )
62 registry = (
63 getattr(self._state, "cluster_registry", None) if self._state else None
64 )
65 self._cluster_registry = registry or ClusterRegistry.with_defaults()
67 # ------------------------------------------------------------------
68 # Clusters
69 # ------------------------------------------------------------------
71 def clusters(self) -> list[Any]:
72 """Return all registered clusters in registry order.
74 Returns:
75 List of :class:`~lexigram.admin.clusters.Cluster` instances.
76 """
77 return self._cluster_registry.all()
79 def active_cluster(self) -> Any | None:
80 """Return the cluster whose center the current path belongs to.
82 Returns:
83 The matching cluster, or ``None`` outside every cluster center.
84 """
85 path = self._current_path()
86 return self._cluster_registry.for_path(path)
88 def _current_path(self) -> str | None:
89 if not self._request or not hasattr(self._request, "url"):
90 return None
91 return str(self._request.url.path)
93 # ------------------------------------------------------------------
94 # Primary nav resolution
95 # ------------------------------------------------------------------
97 def resolve_nav(self) -> tuple[list, list, list | None]:
98 """Resolve the full navigation state for this request.
100 Merges NavItemBuilder resource items with NavigationAssembler
101 contributor items, computes per-request active states, collapses
102 every cluster group from the primary sidebar, and — when the path
103 belongs to a cluster center — returns its secondary nav.
105 Returns:
106 A tuple of (nav_items, system_menu_items, secondary_nav).
107 """
108 if self._nav_builder is None:
109 return [], [], None
111 from lexigram.admin.navigation.clusters import (
112 build_secondary_nav,
113 cluster_items,
114 collapse_cluster_in_primary,
115 is_cluster_path,
116 )
118 current_path = self._current_path()
119 assembler_nav_items = list(self._assembler_nav_items)
121 cluster_nav: list | None = None
122 items_by_cluster: dict[Any, list] = {}
123 for cluster in self._cluster_registry.all():
124 items = cluster_items(self._assembler_groups, cluster=cluster)
125 if not items:
126 continue
127 items_by_cluster[cluster] = items
128 if cluster_nav is None and is_cluster_path(
129 current_path, items, cluster=cluster
130 ):
131 cluster_nav = build_secondary_nav(items, current_path, cluster=cluster)
132 for cluster, items in items_by_cluster.items():
133 assembler_nav_items = collapse_cluster_in_primary(
134 assembler_nav_items,
135 current_path,
136 items,
137 cluster=cluster,
138 )
140 builder_items = self._nav_builder.build_nav_items(current_path=current_path)
141 system_menu_items = self._nav_builder.build_system_menu_items()
143 merged = list(builder_items)
144 seen_hrefs: set[str] = set()
145 group_labels: dict[str, set[str]] = {}
146 current_group = ""
148 for item in merged:
149 if not isinstance(item, dict):
150 continue
151 if item.get("is_group"):
152 current_group = item.get("label", "") or ""
153 group_labels.setdefault(current_group, set())
154 else:
155 href = (item.get("href", "") or "").strip()
156 if href:
157 seen_hrefs.add(href)
158 label = (item.get("label", "") or "").strip()
159 if label:
160 group_labels.setdefault(current_group, set()).add(label)
162 top_items: list[dict] = []
163 for item in assembler_nav_items:
164 if not isinstance(item, dict):
165 continue
166 if item.get("is_group"):
167 break
168 href = (item.get("href", "") or "").strip()
169 label = (item.get("label", "") or "").strip()
170 if current_path is not None and href:
171 item["active"] = current_path == href or current_path.startswith(
172 href + "/"
173 )
174 if href:
175 seen_hrefs.add(href)
176 if label:
177 group_labels.setdefault("", set()).add(label)
178 top_items.append(item)
180 merged = top_items + merged
182 current_group = ""
183 for item in assembler_nav_items:
184 if not isinstance(item, dict):
185 merged.append(item)
186 continue
188 if item.get("is_group"):
189 group_label = (item.get("label", "") or "").strip()
190 current_group = group_label
191 if group_label in group_labels:
192 continue
193 group_labels.setdefault(current_group, set())
194 merged.append(item)
195 continue
197 href = (item.get("href", "") or "").strip()
198 label = (item.get("label", "") or "").strip()
200 if href and href in seen_hrefs:
201 continue
202 if label and label in group_labels.get(current_group, set()):
203 continue
205 item["active"] = (
206 current_path is not None
207 and href
208 and (current_path == href or current_path.startswith(href + "/"))
209 )
211 if href:
212 seen_hrefs.add(href)
213 if label:
214 group_labels.setdefault(current_group, set()).add(label)
215 merged.append(item)
217 return merged, system_menu_items, cluster_nav
219 # ------------------------------------------------------------------
220 # User menu
221 # ------------------------------------------------------------------
223 def user_menu_items(
224 self, include_plugins: bool = True
225 ) -> list[dict[str, str | None]]:
226 """Build the shell user-menu entries for this request.
228 The Profile entry comes first, then cluster centers (one entry per
229 registered cluster), then Plugins and Settings.
231 Args:
232 include_plugins: Include the Plugins landing entry (skipped by
233 the placeholder/under-construction shell).
235 Returns:
236 Shell-compatible menu entry dicts (label, href, icon).
237 """
238 entries: list[MenuItem] = [_MENU_PROFILE]
239 entries.extend(
240 MenuItem(
241 label=cluster.label,
242 href=f"/admin/{cluster.slug}",
243 icon=cluster.icon or "box",
244 )
245 for cluster in self._cluster_registry.all()
246 )
247 if include_plugins:
248 entries.append(_MENU_PLUGINS)
249 entries.append(_MENU_SETTINGS)
250 return [entry.to_dict() for entry in entries]