Coverage for src/lexigram/admin/clusters/registry.py: 0%
33 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Cluster registry — declarative, multi-cluster support.
3The built-in *infrastructure* cluster is registered by default; additional
4clusters can be registered in code or declared via
5``AdminConfig.clusters.extra``. Consumer code never hard-codes a single
6cluster: navigation helpers and the generic cluster center controller read
7the active cluster from this registry.
8"""
10from __future__ import annotations
12from lexigram.admin.clusters.base import Cluster
14INFRASTRUCTURE_CLUSTER = Cluster(
15 name="infrastructure",
16 slug="infrastructure",
17 group="infrastructure",
18 label="Infrastructure",
19 icon="server",
20 description=(
21 "Monitor and manage the services powering your application: "
22 "web, data, and runtime areas."
23 ),
24)
26__all__ = ["INFRASTRUCTURE_CLUSTER", "ClusterRegistry"]
29class ClusterRegistry:
30 """Registry of named clusters, resolved by slug, group, or path.
32 Example:
33 ```python
34 registry = ClusterRegistry()
35 registry.register(cluster)
36 active = registry.for_path("/admin/content/posts")
37 ```
38 """
40 def __init__(self) -> None:
41 """Create an empty registry — no self-registration."""
42 self._clusters: dict[str, Cluster] = {}
43 self._by_group: dict[str, Cluster] = {}
45 @classmethod
46 def with_defaults(cls) -> ClusterRegistry:
47 """Create a registry pre-populated with the built-in clusters."""
48 registry = cls()
49 registry.register(INFRASTRUCTURE_CLUSTER)
50 return registry
52 def register(self, cluster: Cluster) -> None:
53 """Register a cluster, resolving empty slug/group to its name.
55 Args:
56 cluster: Cluster to register.
57 """
58 slug = cluster.slug or cluster.name
59 group = cluster.group or cluster.name
60 resolved = Cluster(
61 name=cluster.name,
62 label=cluster.label,
63 icon=cluster.icon,
64 order=cluster.order,
65 collapsible=cluster.collapsible,
66 collapsed_by_default=cluster.collapsed_by_default,
67 slug=slug,
68 group=group,
69 description=cluster.description,
70 resources=list(cluster.resources),
71 pages=list(cluster.pages),
72 )
73 self._clusters[slug] = resolved
74 self._by_group[group] = resolved
76 def all(self) -> list[Cluster]:
77 """Return all registered clusters, ordered by ``order`` then name."""
78 return sorted(
79 self._clusters.values(),
80 key=lambda c: (c.order, c.name),
81 )
83 def by_slug(self, slug: str) -> Cluster | None:
84 """Look up a cluster by its URL slug.
86 Args:
87 slug: URL slug (e.g. ``"infrastructure"``).
89 Returns:
90 The cluster, or ``None`` when unknown.
91 """
92 return self._clusters.get(slug)
94 def by_group(self, group: str) -> Cluster | None:
95 """Look up a cluster by its navigation group name.
97 Args:
98 group: Contributor navigation group (e.g. ``"infrastructure"``).
100 Returns:
101 The cluster, or ``None`` when unknown.
102 """
103 return self._by_group.get(group)
105 def for_path(self, path: str | None) -> Cluster | None:
106 """Return the cluster whose center namespace contains the path.
108 Matches ``/admin/{slug}`` and anything below it, regardless of the
109 admin mount prefix.
111 Args:
112 path: Request path (e.g. ``"/admin/infrastructure/web"``).
114 Returns:
115 The matching cluster, or ``None`` when the path is outside
116 every cluster center.
117 """
118 if not path:
119 return None
120 for cluster in self.all():
121 prefix = f"/admin/{cluster.slug}"
122 if path == prefix or path.startswith(prefix + "/"):
123 return cluster
124 return None