Skip to content

Commit 382dfef

Browse files
authored
feat(hey_egeria): add general element register view (any type, grouped by subtype) (#301)
hey_egeria tech show elements register (Click) and a standalone show_register script -- a Rich-based, general-purpose take on browsing any family of Egeria elements, grouped by each element's own real subtype. Takes any Open Metadata Type (default Referenceable); fetches elements of that type + subtypes via get_elements(), groups by elementHeader.type.typeName so one command covers GovernanceDefinition, Project, SolutionComponent, etc. -- not hardcoded to any one domain or data set. Summary panel (element/subtype/classification counts), one table per subtype with classification badges (deterministic color-by-name), --search and --classification filters. Correctly follows Egeria's paging contract -- pages until an empty result, not just a short one (regression-tested for that specifically, since a short-page-means-last-page bug is an easy mistake here). 7 new unit tests in tests/micro-tests/test_list_element_register.py against a fake client (grouping, search filter, classification filter, paging contract, badge/display-name helpers). Full micro-tests suite passes. Signed-off-by: Dan Wolfson <dan.wolfson@pdr-associates.com>
1 parent 188f3e7 commit 382dfef

4 files changed

Lines changed: 429 additions & 1 deletion

File tree

‎commands/cli/egeria.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@
109109
from commands.tech.get_tech_type_template import template_viewer
110110
from commands.tech.list_all_om_type_elements import list_elements
111111
from commands.tech.list_all_om_type_elements_x import list_elements_x
112+
from commands.tech.list_element_register import element_register
112113
from commands.tech.list_all_related_elements import list_related_elements
113114
from commands.tech.list_anchored_elements import display_anchored_elements
114115
from commands.tech.list_asset_types import display_asset_types
@@ -942,6 +943,35 @@ def get_element_info(ctx, om_type):
942943
display_elements(om_type, c.view_server, c.view_server_url, c.userid, c.password, c.jupyter, c.width)
943944

944945

946+
@show_elements.command("register")
947+
@click.pass_context
948+
@click.option("--om_type", default="Referenceable", help="Open Metadata Type to browse (subtypes included)")
949+
@click.option("--search", default=None, help="Filter by display name or qualified name substring")
950+
@click.option("--classification", default=None, help="Filter to elements carrying this classification")
951+
@click.option("--page-size", type=int, default=100, help="Page size used when paging through results")
952+
def show_element_register(ctx, om_type, search, classification, page_size):
953+
"""Display every element of a type family, grouped by its real subtype, with classification badges
954+
955+
A general-purpose register view: point it at any Open Metadata Type
956+
(e.g. "GovernanceDefinition", "Referenceable", "Project",
957+
"SolutionComponent") and it groups whatever comes back by each
958+
element's own type -- not just governance definitions.
959+
"""
960+
c = ctx.obj
961+
element_register(
962+
om_type,
963+
c.view_server,
964+
c.view_server_url,
965+
c.userid,
966+
c.password,
967+
search=search,
968+
classification=classification,
969+
page_size=page_size,
970+
jupyter=c.jupyter,
971+
width=c.width,
972+
)
973+
974+
945975
@show_supply_chains.command("supply-chains")
946976
@click.option("--search-string", default="*", help="Search string")
947977
@click.pass_context
Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
"""
2+
Element Register — a hey_egeria/Rich take on browsing any family of Egeria
3+
elements grouped by their real (sub)type, with badges for classifications
4+
and a simple search/filter.
5+
6+
Deliberately general: takes any Open Metadata Type name (e.g.
7+
"GovernanceDefinition", "Referenceable", "Project", "SolutionComponent")
8+
and groups whatever comes back by `elementHeader.type.typeName` -- so one
9+
command works for every element family, not just governance definitions.
10+
"""
11+
12+
import argparse
13+
import os
14+
import sys
15+
import time
16+
17+
from rich import box
18+
from rich.console import Console, Group
19+
from rich.padding import Padding
20+
from rich.panel import Panel
21+
from rich.table import Table
22+
from rich.text import Text
23+
24+
from pyegeria import (
25+
EgeriaTech,
26+
PyegeriaException,
27+
config_logging,
28+
load_app_config,
29+
print_basic_exception,
30+
settings,
31+
)
32+
33+
app_config = settings.Environment
34+
config_path = os.path.join(app_config.pyegeria_config_directory, app_config.pyegeria_config_file)
35+
36+
EGERIA_USER = os.environ.get("EGERIA_USER", "erinoverview")
37+
EGERIA_USER_PASSWORD = os.environ.get("EGERIA_USER_PASSWORD", "secret")
38+
conf = load_app_config(config_path)
39+
40+
config_logging()
41+
42+
# A small, fixed palette so classification badges are stable across runs --
43+
# the same classification name always lands on the same color, cycling
44+
# deterministically by a hash of the name rather than insertion order.
45+
_BADGE_PALETTE = [
46+
"black on #c9a227", # gold
47+
"black on #4fa5a0", # teal
48+
"black on #7a9b4e", # olive
49+
"black on #4d7ea8", # steel blue
50+
"black on #a8544d", # rust
51+
"black on #8b6bb1", # violet
52+
"black on #b1866b", # tan
53+
"black on #6b8fb1", # blue-grey
54+
]
55+
56+
57+
def _badge_style(label: str) -> str:
58+
return _BADGE_PALETTE[sum(ord(c) for c in label) % len(_BADGE_PALETTE)]
59+
60+
61+
def _badges(labels: list[str]) -> Text:
62+
out = Text()
63+
for i, label in enumerate(labels):
64+
if i:
65+
out.append(" ")
66+
out.append(f" {label} ", style=_badge_style(label))
67+
return out
68+
69+
70+
def _display_name(properties: dict, qualified_name: str) -> str:
71+
for key in ("displayName", "name", "title"):
72+
val = properties.get(key)
73+
if val:
74+
return val
75+
return qualified_name or "---"
76+
77+
78+
def _fetch_all_elements(c_client: EgeriaTech, om_type: str, page_size: int) -> list[dict]:
79+
"""Page through get_elements until an empty page comes back.
80+
81+
Per Egeria's own paging contract, a short page does not mean the last
82+
page -- only an empty result does -- so this keeps asking until that
83+
happens rather than stopping the moment a page comes back under
84+
page_size.
85+
"""
86+
all_elements: list[dict] = []
87+
start_from = 0
88+
while True:
89+
page = c_client.get_elements(om_type, start_from=start_from, page_size=page_size, output_format="JSON")
90+
if not isinstance(page, list) or len(page) == 0:
91+
break
92+
all_elements.extend(page)
93+
start_from += page_size
94+
return all_elements
95+
96+
97+
def element_register(
98+
om_type: str,
99+
server: str,
100+
url: str,
101+
username: str,
102+
password: str,
103+
search: str | None = None,
104+
classification: str | None = None,
105+
page_size: int = 100,
106+
jupyter: bool = settings.Environment.egeria_jupyter,
107+
width: int = settings.Environment.egeria_width,
108+
):
109+
c_client = EgeriaTech(server, url, user_id=username, user_pwd=password)
110+
c_client.create_egeria_bearer_token()
111+
112+
om_typedef = c_client.get_typedef_by_name(om_type)
113+
if isinstance(om_typedef, str):
114+
print(f"The type name '{om_type}' is not known to the Egeria platform at {url} - {server}")
115+
sys.exit(1)
116+
117+
try:
118+
raw_elements = _fetch_all_elements(c_client, om_type, page_size)
119+
120+
rows = []
121+
for element in raw_elements:
122+
header = element.get("elementHeader", {})
123+
properties = element.get("properties", {})
124+
qualified_name = properties.get("qualifiedName", "---")
125+
type_name = header.get("type", {}).get("typeName", om_type)
126+
guid = header.get("guid", "---")
127+
classifications = [
128+
c.get("classificationName", "---")
129+
for c in header.get("otherClassifications", []) or []
130+
if isinstance(c, dict)
131+
]
132+
rows.append({
133+
"display_name": _display_name(properties, qualified_name),
134+
"qualified_name": qualified_name,
135+
"type_name": type_name,
136+
"guid": guid,
137+
"classifications": classifications,
138+
})
139+
140+
search_lower = search.lower() if search else None
141+
if search_lower:
142+
rows = [
143+
r for r in rows
144+
if search_lower in r["display_name"].lower() or search_lower in r["qualified_name"].lower()
145+
]
146+
if classification:
147+
rows = [r for r in rows if classification in r["classifications"]]
148+
149+
groups: dict[str, list[dict]] = {}
150+
for r in rows:
151+
groups.setdefault(r["type_name"], []).append(r)
152+
153+
distinct_classifications = {c for r in rows for c in r["classifications"]}
154+
155+
stats = Table.grid(padding=(0, 3))
156+
stats.add_row(
157+
Text(str(len(rows)), style="bold white"),
158+
Text(str(len(groups)), style="bold white"),
159+
Text(str(len(distinct_classifications)), style="bold white"),
160+
)
161+
stats.add_row(
162+
Text("ELEMENTS", style="dim"),
163+
Text("SUBTYPES", style="dim"),
164+
Text("CLASSIFICATIONS", style="dim"),
165+
)
166+
167+
header_lines = [f"Open Metadata Type: [bold]{om_type}[/bold] · {server} @ {url}"]
168+
if search:
169+
header_lines.append(f"Search: '{search}'")
170+
if classification:
171+
header_lines.append(f"Classification filter: '{classification}'")
172+
header_lines.append(f"{len(rows)} of {len(raw_elements)} shown · {time.asctime()}")
173+
174+
panel = Panel(
175+
Group(stats, Padding(Text("\n".join(header_lines), style="dim"), (1, 0, 0, 0))),
176+
title=f"Element Register",
177+
border_style="bright_blue",
178+
box=box.ROUNDED,
179+
)
180+
181+
renderables = [panel]
182+
for type_name in sorted(groups.keys()):
183+
group_rows = groups[type_name]
184+
table = Table(
185+
title=f"{type_name} [dim]({len(group_rows)})[/dim]",
186+
title_justify="left",
187+
box=box.SIMPLE_HEAVY,
188+
header_style="bold white",
189+
show_lines=False,
190+
expand=True,
191+
)
192+
table.add_column("Display Name", style="bold")
193+
table.add_column("Classifications")
194+
table.add_column("Qualified Name", style="dim", overflow="fold")
195+
196+
for r in sorted(group_rows, key=lambda r: r["display_name"].lower()):
197+
table.add_row(
198+
r["display_name"],
199+
_badges(r["classifications"]) if r["classifications"] else Text("—", style="dim"),
200+
r["qualified_name"],
201+
)
202+
renderables.append(table)
203+
204+
if not rows:
205+
renderables.append(Text("No elements matched.", style="italic dim"))
206+
207+
console = Console(width=width)
208+
with console.pager(styles=True):
209+
for renderable in renderables:
210+
console.print(renderable)
211+
console.print()
212+
213+
except PyegeriaException as e:
214+
print_basic_exception(e)
215+
finally:
216+
c_client.close_session()
217+
218+
219+
def main():
220+
parser = argparse.ArgumentParser()
221+
parser.add_argument("--server", help="Name of the server to display status for")
222+
parser.add_argument("--url", help="URL Platform to connect to")
223+
parser.add_argument("--userid", help="User Id")
224+
parser.add_argument("--password", help="Password")
225+
parser.add_argument("--om-type", default="Referenceable", help="Open Metadata Type to browse")
226+
parser.add_argument("--search", default=None, help="Filter by display name or qualified name substring")
227+
parser.add_argument("--classification", default=None, help="Filter to elements carrying this classification")
228+
parser.add_argument("--page-size", type=int, default=100, help="Page size used when paging through results")
229+
230+
args = parser.parse_args()
231+
232+
server = args.server if args.server is not None else app_config.egeria_view_server
233+
url = args.url if args.url is not None else app_config.egeria_platform_url
234+
userid = args.userid if args.userid is not None else EGERIA_USER
235+
password = args.password if args.password is not None else EGERIA_USER_PASSWORD
236+
237+
try:
238+
element_register(
239+
args.om_type,
240+
server,
241+
url,
242+
userid,
243+
password,
244+
search=args.search,
245+
classification=args.classification,
246+
page_size=args.page_size,
247+
)
248+
except KeyboardInterrupt:
249+
pass
250+
251+
252+
if __name__ == "__main__":
253+
main()

‎pyproject.toml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33

44
[project]
55
name = "pyegeria"
6-
version = "6.1.1"
6+
version = "6.1.2"
77
license = 'Apache-2.0'
88
license-files = ["LICENSE"]
99
authors = [{name = "Dan Wolfson", email = "dan.wolfson@pdr-associates.com"},]
@@ -163,6 +163,7 @@ dependencies = [
163163
list_tech_templates = "commands.tech.list_tech_templates:main"
164164
get_element_info = "commands.tech.get_element_info:main"
165165
list_elements = "commands.tech.list_all_om_type_elements:main"
166+
show_register = "commands.tech.list_element_register:main"
166167
list_elements_x = "commands.tech.list_all_om_type_elements_x:main"
167168
list_elements_by_classification_by_prop_value = "commands.tech.list_elements_by_classification_by_property_value:main"
168169
list_elements_by_prop_value = "commands.tech.list_elements_by_property_value:main"

0 commit comments

Comments
 (0)