diff --git a/pyegeria/omvs/asset_catalog.py b/pyegeria/omvs/asset_catalog.py index 18dd2a89..e5f155c1 100644 --- a/pyegeria/omvs/asset_catalog.py +++ b/pyegeria/omvs/asset_catalog.py @@ -24,6 +24,74 @@ from typing import Any, Optional from pyegeria.core.async_runner import run_sync +# Report spec for the hop-aware lineage listing (one row per lineage node). +LINEAGE_HOPS_REPORT_SPEC = "Asset-Lineage-Hops" + + +def _lineage_nodes(graph: dict) -> list[dict]: + """The starting asset followed by its linked elements, once each.""" + nodes, seen = [], set() + for node in [graph] + list(graph.get("linkedAssets") or []): + guid = (node.get("elementHeader") or {}).get("guid") + if guid and guid in seen: + continue + seen.add(guid) + nodes.append(node) + return nodes + + +def lineage_hop_summary(graph: dict, max_hops: Optional[int] = None) -> dict: + """Say whether an asset lineage graph was cut by Egeria's hop limit. + + Egeria 6.3 stops the lineage walk after ``maxHops`` relationships (default 8) and sets + ``hopLimitReached`` on each element whose lineage continues past the limit. A platform older + than 6.3 sends neither ``hopsFromStart`` nor ``hopLimitReached``, so its graphs are reported as + of unknown completeness, never as complete. + + Parameters + ---------- + graph : dict + The ``element`` returned by the as-lineage-graph request (an AssetLineageGraph). + max_hops : int, optional + The ``maxHops`` the request sent, if any. Used only to word the summary. + + Returns + ------- + dict + ``complete`` (True, False, or None when unknown), ``nodeCount``, ``maxHopsFromStart``, + ``hopLimitReachedCount``, ``cutAtHops`` (None unless cut) and a one-line ``summary``. + """ + nodes = _lineage_nodes(graph) + hops = [n["hopsFromStart"] for n in nodes if isinstance(n.get("hopsFromStart"), int)] + if not hops: + return { + "complete": None, + "nodeCount": len(nodes), + "maxHopsFromStart": None, + "hopLimitReachedCount": None, + "cutAtHops": None, + "summary": (f"{len(nodes)} elements · hop information not reported (Egeria platform " + f"older than 6.3), so it is unknown whether the graph is complete"), + } + cut = [n for n in nodes if n.get("hopLimitReached") is True] + deepest = max(hops) + if cut: + cut_at = max_hops if max_hops else max(n.get("hopsFromStart") or 0 for n in cut) + noun = "element continues" if len(cut) == 1 else "elements continue" + summary = f"Graph cut at {cut_at} hops · {len(cut)} {noun} past the limit · {len(nodes)} elements shown" + else: + cut_at = None + summary = f"Complete graph · {len(nodes)} elements · up to {deepest} hops from the start" + return { + "complete": not cut, + "nodeCount": len(nodes), + "maxHopsFromStart": deepest, + "hopLimitReachedCount": len(cut), + "cutAtHops": cut_at, + "summary": summary, + } + + class AssetCatalog(ServerClient): """ Client for the Asset Catalog View Service. @@ -82,6 +150,9 @@ def _extract_asset_properties(self, element: dict, columns_struct: dict) -> dict col["value"] = proc_status elif key == "stepCount": col["value"] = step_count + elif key in ("hopsFromStart", "hopLimitReached"): + # Top-level on AssetLineageGraphNode (Egeria 6.3+); left empty, not 0/False, when absent. + col["value"] = element.get(key) col_data = get_required_relationships(element, col_data) mermaid_val = element.get("mermaidGraph", "") or "" for col in columns_list: @@ -451,6 +522,9 @@ async def _async_get_asset_lineage_graph_by_guid( graph_query_depth: int = 5, max_mermaid_node_count: int = 10, for_lineage: Optional[bool] = None, + max_hops: Optional[int] = None, + skip_relationships: Optional[list[str]] = None, + timeout: Optional[int] = None, output_format: str = "DICT", report_spec: str | dict = "Common-Mermaid", **kwargs, @@ -468,7 +542,13 @@ async def _async_get_asset_lineage_graph_by_guid( as_of_time : str, optional As-of time to query on. If not specified, the current time is used. relationship_types : list[str], optional - Relationship types to include in the lineage graph. If not specified, all relationship types are included. + Lineage relationship types to follow (sent as ``includeOnlyRelationships``). The list replaces + Egeria's defaults rather than adding to them, and a supertype name such as ``LineageRelationship`` + or ``DataLineageRelationship`` includes all of its subtypes. If not specified, Egeria's default + list applies; from Egeria 6.3 that is DataFlow, ControlFlow, LineageMapping, ProcessCall, + DerivedSchemaTypeQueryTarget, DataSetContent, DataMapping and ActionRequester. + RequestForActionTarget and AssignmentScope are no longer followed by default: name them here + to follow them. limit_to_isc_q_name : str, optional If specified, filters results to only include information supply chains with the given qualified name. hilight_isc_q_name : str, optional @@ -481,22 +561,39 @@ async def _async_get_asset_lineage_graph_by_guid( The maximum number of items to fetch. graph_query_depth : int, default 5 How many hops of the lineage graph to traverse. - max_mermaid_node_count : int, default 5 + max_mermaid_node_count : int, default 10 Maximum number of nodes to include in the generated mermaid graph. for_lineage : bool, optional Set True to include elements only visible to lineage requests (e.g. ones classified as Promise or Memento). If not specified, Egeria's default (False) applies. + max_hops : int, optional + Egeria 6.3+: how many lineage relationships to follow away from the asset in each direction + (sent as ``maxHops``). 0 means no limit, which can take minutes on a large estate (raise + ``timeout`` too). If not specified, nothing is sent and Egeria's default (8 on 6.3) applies. + Elements whose lineage continues past the limit come back with ``hopLimitReached`` set, and + every element carries ``hopsFromStart``; see ``lineage_hop_summary``. + skip_relationships : list[str], optional + Egeria 6.3+: relationship types (and their subtypes) to remove from the default or requested + list (sent as ``skipRelationships``), for example ``["DataSetContent"]``. Ignored by older + platforms. + timeout : int, optional + Seconds to wait for this request, overriding the client's timeout. Lineage walks are slow on + large estates. output_format : str, default "DICT" The desired output format. report_spec : str | dict, default "Common-Mermaid" - The desired output columns/fields to include. + The desired output columns/fields to include. ``"Asset-Lineage-Hops"`` lists one row per lineage + element (the asset first) with ``hopsFromStart`` and ``hopLimitReached``, headed by the + ``lineage_hop_summary`` line that says whether, and where, the graph was cut. **kwargs : dict, optional Additional query parameters. Returns ------- str | dict - A dictionary of the asset graph that includes a mermaid markdown string. + A dictionary of the asset graph that includes a mermaid markdown string. With + ``report_spec="Asset-Lineage-Hops"``, string formats start with the summary line and list + formats return ``{"lineageSummary": {...}, "nodes": [...]}``. Raises ------ @@ -533,14 +630,27 @@ async def _async_get_asset_lineage_graph_by_guid( "graphQueryDepth": graph_query_depth, "maxMermaidNodeCount": max_mermaid_node_count, "forLineage": for_lineage, + # Egeria 6.3+. None is slimmed away, so a 6.2 platform sees the same body as before. + "maxHops": max_hops, + "skipRelationships": skip_relationships, } - response = await self._async_make_request("POST", url, body_slimmer(body)) + request_kwargs = {"timeout": timeout} if timeout is not None else {} + response = await self._async_make_request("POST", url, body_slimmer(body), **request_kwargs) element = response.json().get("element", NO_ASSETS_FOUND) if isinstance(element, str): logger.info(NO_ELEMENTS_FOUND) return NO_ELEMENTS_FOUND + if output_format != "JSON" and report_spec == LINEAGE_HOPS_REPORT_SPEC: + summary = lineage_hop_summary(element, max_hops) + rows = self._generate_asset_output( + _lineage_nodes(element), None, "Asset", output_format, report_spec + ) + if isinstance(rows, str): + return f"{summary['summary']}\n\n{rows}" + return {"lineageSummary": summary, "nodes": rows} + if output_format != "JSON": # return a simplified markdown representation logger.info( f"Found elements, output format: {output_format} and report_spec: {report_spec}" @@ -564,6 +674,9 @@ def get_asset_lineage_graph_by_guid( graph_query_depth: int = 5, max_mermaid_node_count: int = 10, for_lineage: Optional[bool] = None, + max_hops: Optional[int] = None, + skip_relationships: Optional[list[str]] = None, + timeout: Optional[int] = None, output_format: str = "DICT", report_spec: str | dict = "Common-Mermaid", **kwargs, @@ -579,7 +692,13 @@ def get_asset_lineage_graph_by_guid( as_of_time : str, optional As-of time to query on. If not specified, the current time is used. relationship_types : list[str], optional - Relationship types to include in the lineage graph. If not specified, all relationship types are included. + Lineage relationship types to follow (sent as ``includeOnlyRelationships``). The list replaces + Egeria's defaults rather than adding to them, and a supertype name such as ``LineageRelationship`` + or ``DataLineageRelationship`` includes all of its subtypes. If not specified, Egeria's default + list applies; from Egeria 6.3 that is DataFlow, ControlFlow, LineageMapping, ProcessCall, + DerivedSchemaTypeQueryTarget, DataSetContent, DataMapping and ActionRequester. + RequestForActionTarget and AssignmentScope are no longer followed by default: name them here + to follow them. limit_to_isc_q_name : str, optional If specified, filters results to only include information supply chains with the given qualified name. hilight_isc_q_name : str, optional @@ -592,22 +711,39 @@ def get_asset_lineage_graph_by_guid( The maximum number of items to fetch. graph_query_depth : int, default 5 How many hops of the lineage graph to traverse. - max_mermaid_node_count : int, default 5 + max_mermaid_node_count : int, default 10 Maximum number of nodes to include in the generated mermaid graph. for_lineage : bool, optional Set True to include elements only visible to lineage requests (e.g. ones classified as Promise or Memento). If not specified, Egeria's default (False) applies. + max_hops : int, optional + Egeria 6.3+: how many lineage relationships to follow away from the asset in each direction + (sent as ``maxHops``). 0 means no limit, which can take minutes on a large estate (raise + ``timeout`` too). If not specified, nothing is sent and Egeria's default (8 on 6.3) applies. + Elements whose lineage continues past the limit come back with ``hopLimitReached`` set, and + every element carries ``hopsFromStart``; see ``lineage_hop_summary``. + skip_relationships : list[str], optional + Egeria 6.3+: relationship types (and their subtypes) to remove from the default or requested + list (sent as ``skipRelationships``), for example ``["DataSetContent"]``. Ignored by older + platforms. + timeout : int, optional + Seconds to wait for this request, overriding the client's timeout. Lineage walks are slow on + large estates. output_format : str, default "DICT" The desired output format. report_spec : str | dict, default "Common-Mermaid" - The desired output columns/fields to include. + The desired output columns/fields to include. ``"Asset-Lineage-Hops"`` lists one row per lineage + element (the asset first) with ``hopsFromStart`` and ``hopLimitReached``, headed by the + ``lineage_hop_summary`` line that says whether, and where, the graph was cut. **kwargs : dict, optional Additional query parameters. Returns ------- str | dict - A dictionary of the asset graph that includes a mermaid markdown string. + A dictionary of the asset graph that includes a mermaid markdown string. With + ``report_spec="Asset-Lineage-Hops"``, string formats start with the summary line and list + formats return ``{"lineageSummary": {...}, "nodes": [...]}``. Raises ------ @@ -629,6 +765,9 @@ def get_asset_lineage_graph_by_guid( graph_query_depth=graph_query_depth, max_mermaid_node_count=max_mermaid_node_count, for_lineage=for_lineage, + max_hops=max_hops, + skip_relationships=skip_relationships, + timeout=timeout, output_format=output_format, report_spec=report_spec, **kwargs, @@ -650,6 +789,9 @@ def get_asset_lineage_mermaid_graph( graph_query_depth: int = 5, max_mermaid_node_count: int = 10, for_lineage: Optional[bool] = None, + max_hops: Optional[int] = None, + skip_relationships: Optional[list[str]] = None, + timeout: Optional[int] = None, ) -> str: """Return the asset lineage including a mermaid markdown string. @@ -662,7 +804,13 @@ def get_asset_lineage_mermaid_graph( as_of_time : str, optional As-of time to query on. If not specified, the current time is used. relationship_types : list[str], optional - Relationship types to include in the lineage graph. If not specified, all relationship types are included. + Lineage relationship types to follow (sent as ``includeOnlyRelationships``). The list replaces + Egeria's defaults rather than adding to them, and a supertype name such as ``LineageRelationship`` + or ``DataLineageRelationship`` includes all of its subtypes. If not specified, Egeria's default + list applies; from Egeria 6.3 that is DataFlow, ControlFlow, LineageMapping, ProcessCall, + DerivedSchemaTypeQueryTarget, DataSetContent, DataMapping and ActionRequester. + RequestForActionTarget and AssignmentScope are no longer followed by default: name them here + to follow them. limit_to_isc_q_name : str, optional If specified, filters results to only include information supply chains with the given qualified name. hilight_isc_q_name : str, optional @@ -675,11 +823,24 @@ def get_asset_lineage_mermaid_graph( The maximum number of items to fetch. graph_query_depth : int, default 5 How many hops of the lineage graph to traverse. - max_mermaid_node_count : int, default 5 + max_mermaid_node_count : int, default 10 Maximum number of nodes to include in the generated mermaid graph. for_lineage : bool, optional Set True to include elements only visible to lineage requests (e.g. ones classified as Promise or Memento). If not specified, Egeria's default (False) applies. + max_hops : int, optional + Egeria 6.3+: how many lineage relationships to follow away from the asset in each direction + (sent as ``maxHops``). 0 means no limit, which can take minutes on a large estate (raise + ``timeout`` too). If not specified, nothing is sent and Egeria's default (8 on 6.3) applies. + Elements whose lineage continues past the limit come back with ``hopLimitReached`` set, and + every element carries ``hopsFromStart``; see ``lineage_hop_summary``. + skip_relationships : list[str], optional + Egeria 6.3+: relationship types (and their subtypes) to remove from the default or requested + list (sent as ``skipRelationships``), for example ``["DataSetContent"]``. Ignored by older + platforms. + timeout : int, optional + Seconds to wait for this request, overriding the client's timeout. Lineage walks are slow on + large estates. Returns ------- @@ -707,8 +868,17 @@ def get_asset_lineage_mermaid_graph( graph_query_depth=graph_query_depth, max_mermaid_node_count=max_mermaid_node_count, for_lineage=for_lineage, + max_hops=max_hops, + skip_relationships=skip_relationships, + timeout=timeout, output_format="JSON", ) + if not isinstance(asset_graph, dict): + return asset_graph + hop_summary = lineage_hop_summary(asset_graph, max_hops) + if hop_summary["complete"] is False: + # The Mermaid string itself does not mark the cut; don't let it pass as the whole graph. + logger.warning(hop_summary["summary"]) return asset_graph.get("mermaidGraph") diff --git a/pyegeria/view/base_report_formats.py b/pyegeria/view/base_report_formats.py index 9617642d..d6aea247 100644 --- a/pyegeria/view/base_report_formats.py +++ b/pyegeria/view/base_report_formats.py @@ -1419,6 +1419,29 @@ def _user_report_specs_dir() -> str: spec_params={}, ) ), + "Asset-Lineage-Hops": FormatSet( + target_type="Asset", + heading="Asset Lineage Hops", + description=("One row per element of an asset lineage graph, with how many hops it is from the asset and " + "whether its lineage continues past Egeria's hop limit (Egeria 6.3+). " + "AssetCatalog.get_asset_lineage_graph_by_guid heads the output with a line saying whether, " + "and where, the graph was cut."), + annotations={}, + family="AssetCatalog", + formats=[ + Format( + types=["ALL"], + attributes=[ + Column(name='Display Name', key='display_name'), + Column(name="Qualified Name", key='qualified_name'), + Column(name="Type Name", key='type_name'), + Column(name="GUID", key='guid'), + Column(name="Hops From Start", key='hopsFromStart'), + Column(name="Hop Limit Reached", key='hopLimitReached'), + ], + ) + ], + ), "Common-Mermaid": FormatSet( target_type="Referenceable", heading="Common Attributes with Mermaid", diff --git a/tests/micro-tests/test_asset_lineage_max_hops.py b/tests/micro-tests/test_asset_lineage_max_hops.py new file mode 100644 index 00000000..75417269 --- /dev/null +++ b/tests/micro-tests/test_asset_lineage_max_hops.py @@ -0,0 +1,178 @@ +""" +AssetCatalog lineage-graph parameters added for Egeria 6.3 (odpi/egeria#9420): +max_hops -> "maxHops" and skip_relationships -> "skipRelationships" are sent +only when set, so a call to a 6.2 platform is byte-identical to before; a +per-call timeout reaches _async_make_request; and the Asset-Lineage-Hops +output says when the graph was cut instead of presenting it as complete. +""" +import json + +import pytest + +from pyegeria.omvs.asset_catalog import ( + AssetCatalog, + LINEAGE_HOPS_REPORT_SPEC, + lineage_hop_summary, +) + + +def _node(guid, name, hops=None, cut=None): + node = { + "elementHeader": { + "guid": guid, + "type": {"typeName": "DataSet"}, + "origin": {"homeMetadataCollectionId": "mc-1"}, + "versions": {"createdBy": "user", "version": 1}, + }, + "properties": {"displayName": name, "qualifiedName": f"DataSet::{name}"}, + } + if hops is not None: + node["hopsFromStart"] = hops + if cut is not None: + node["hopLimitReached"] = cut + return node + + +def _graph_63(cut: bool): + graph = _node("g0", "start", 0, False) + graph["mermaidGraph"] = "graph TD" + graph["linkedAssets"] = [_node("g1", "one", 1, False), _node("g2", "two", 2, cut)] + return graph + + +def _graph_62(): + graph = _node("g0", "start") + graph["mermaidGraph"] = "graph TD" + graph["linkedAssets"] = [_node("g1", "one"), _node("g2", "two")] + return graph + + +class _Response: + def __init__(self, element): + self._element = element + + def json(self): + return {"element": self._element} + + +def _client_capturing(calls: list, monkeypatch, element=None) -> AssetCatalog: + # No live server: skip the constructor's connection check, and use a port + # nothing listens on so any real network call would fail loudly. + monkeypatch.setattr(AssetCatalog, "check_connection", lambda self: "test") + client = AssetCatalog("view-server", "https://localhost:1", "user", "pwd") + + async def fake_make_request(method, url, body=None, *args, **kwargs): + calls.append({"body": json.loads(body) if isinstance(body, str) else body, "kwargs": kwargs}) + return _Response(element if element is not None else {"mermaidGraph": "graph TD"}) + + client._async_make_request = fake_make_request + return client + + +@pytest.mark.asyncio +async def test_unset_fields_are_omitted(monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch) + + await client._async_get_asset_lineage_graph_by_guid("guid-1", output_format="JSON") + + body = calls[0]["body"] + assert "maxHops" not in body + assert "skipRelationships" not in body + assert "timeout" not in calls[0]["kwargs"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("max_hops", [0, 3]) +async def test_max_hops_sent_including_zero(max_hops, monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch) + + await client._async_get_asset_lineage_graph_by_guid("guid-1", max_hops=max_hops, output_format="JSON") + + assert calls[0]["body"]["maxHops"] == max_hops + + +@pytest.mark.asyncio +async def test_skip_relationships_and_timeout_passed(monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch) + + await client._async_get_asset_lineage_graph_by_guid( + "guid-1", + relationship_types=["LineageRelationship"], + skip_relationships=["DataSetContent"], + timeout=300, + output_format="JSON", + ) + + assert calls[0]["body"]["includeOnlyRelationships"] == ["LineageRelationship"] + assert calls[0]["body"]["skipRelationships"] == ["DataSetContent"] + assert calls[0]["kwargs"]["timeout"] == 300 + + +def test_sync_and_mermaid_wrappers_pass_through(monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch) + + client.get_asset_lineage_graph_by_guid("guid-1", max_hops=2, timeout=90, output_format="JSON") + client.get_asset_lineage_mermaid_graph("guid-1", max_hops=0, skip_relationships=["DataMapping"]) + + assert calls[0]["body"]["maxHops"] == 2 + assert calls[0]["kwargs"]["timeout"] == 90 + assert calls[1]["body"]["maxHops"] == 0 + assert calls[1]["body"]["skipRelationships"] == ["DataMapping"] + + +def test_summary_reports_cut(): + summary = lineage_hop_summary(_graph_63(cut=True), max_hops=2) + + assert summary["complete"] is False + assert summary["hopLimitReachedCount"] == 1 + assert summary["cutAtHops"] == 2 + assert summary["summary"].startswith("Graph cut at 2 hops · 1 element continues past the limit") + + +def test_summary_reports_complete(): + summary = lineage_hop_summary(_graph_63(cut=False)) + + assert summary["complete"] is True + assert summary["maxHopsFromStart"] == 2 + assert summary["summary"].startswith("Complete graph") + + +def test_summary_on_pre_63_platform_is_unknown_not_complete(): + summary = lineage_hop_summary(_graph_62()) + + assert summary["complete"] is None + assert summary["hopLimitReachedCount"] is None + assert "unknown" in summary["summary"] + + +@pytest.mark.asyncio +async def test_hops_report_spec_lists_nodes_with_summary(monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch, element=_graph_63(cut=True)) + + result = await client._async_get_asset_lineage_graph_by_guid( + "guid-1", max_hops=2, output_format="DICT", report_spec=LINEAGE_HOPS_REPORT_SPEC + ) + + assert result["lineageSummary"]["complete"] is False + nodes = result["nodes"] + assert [n["GUID"] for n in nodes] == ["g0", "g1", "g2"] + assert [n["Hops From Start"] for n in nodes] == [0, 1, 2] + assert nodes[2]["Hop Limit Reached"] is True + + +@pytest.mark.asyncio +async def test_hops_report_spec_markdown_starts_with_summary(monkeypatch): + calls: list = [] + client = _client_capturing(calls, monkeypatch, element=_graph_63(cut=True)) + + result = await client._async_get_asset_lineage_graph_by_guid( + "guid-1", max_hops=2, output_format="MD", report_spec=LINEAGE_HOPS_REPORT_SPEC + ) + + assert isinstance(result, str) + assert result.startswith("Graph cut at 2 hops")