diff --git a/pyatlan/generator/generate_apps.py b/pyatlan/generator/generate_apps.py index 556b09196..0a1bb7f58 100644 --- a/pyatlan/generator/generate_apps.py +++ b/pyatlan/generator/generate_apps.py @@ -44,7 +44,15 @@ # Modules with a hand-polished builder — the generator leaves these untouched. # databricks_crawler has a hand-written multi-mode asset_selection (the configmap # can't express its include/exclude × hierarchy/regex widget). -_HAND_WRITTEN = {"bigquery_crawler", "databricks_crawler", "kafka_confluent"} +# standard_lineage re-scopes an EXISTING workflow (an update against a slug, which +# no generated builder does) and must JSON-encode a list into a contract field +# declared `str`; neither is expressible from a configmap. +_HAND_WRITTEN = { + "bigquery_crawler", + "databricks_crawler", + "kafka_confluent", + "standard_lineage", +} # Apps to generate even when not currently deployed/running on the tenant # (configmaps are served per app-id, so live discovery alone misses these). diff --git a/pyatlan/model/apps/__init__.py b/pyatlan/model/apps/__init__.py index f5aab1529..998f9f3e5 100644 --- a/pyatlan/model/apps/__init__.py +++ b/pyatlan/model/apps/__init__.py @@ -7,8 +7,6 @@ """ from pyatlan.model.apps._base import AppBuilder, AppInput -from pyatlan.model.apps.bigquery_crawler import BigqueryCrawler, BigqueryCrawlerInputs -from pyatlan.model.apps.bigquery_miner import BigqueryMiner, BigqueryMinerInputs from pyatlan.model.apps.anaplan import Anaplan, AnaplanInputs from pyatlan.model.apps.atlan_athena import AtlanAthena, AtlanAthenaInputs from pyatlan.model.apps.atlan_dbt import AtlanDbt, AtlanDbtInputs @@ -27,6 +25,8 @@ from pyatlan.model.apps.atlan_sigma import AtlanSigma, AtlanSigmaInputs from pyatlan.model.apps.atlan_tableau import AtlanTableau, AtlanTableauInputs from pyatlan.model.apps.atlan_trino import AtlanTrino, AtlanTrinoInputs +from pyatlan.model.apps.bigquery_crawler import BigqueryCrawler, BigqueryCrawlerInputs +from pyatlan.model.apps.bigquery_miner import BigqueryMiner, BigqueryMinerInputs from pyatlan.model.apps.csa_uber_asset_export_basic import ( CsaUberAssetExportBasic, CsaUberAssetExportBasicInputs, @@ -39,10 +39,7 @@ from pyatlan.model.apps.hive_crawler import HiveCrawler, HiveCrawlerInputs from pyatlan.model.apps.kafka_apache import KafkaApache, KafkaApacheInputs from pyatlan.model.apps.kafka_confluent import KafkaConfluent, KafkaConfluentInputs -from pyatlan.model.apps.mongodbatlas_atlas import ( - MongodbAtlas, - MongodbAtlasInputs, -) +from pyatlan.model.apps.mongodbatlas_atlas import MongodbAtlas, MongodbAtlasInputs from pyatlan.model.apps.oracle_crawler import OracleCrawler, OracleCrawlerInputs from pyatlan.model.apps.oracle_miner import OracleMiner, OracleMinerInputs from pyatlan.model.apps.postgres_crawler import PostgresCrawler, PostgresCrawlerInputs @@ -54,6 +51,7 @@ SnowflakeCrawlerInputs, ) from pyatlan.model.apps.snowflake_miner import SnowflakeMiner, SnowflakeMinerInputs +from pyatlan.model.apps.standard_lineage import StandardLineage, StandardLineageInputs from pyatlan.model.apps.teradata_crawler import TeradataCrawler, TeradataCrawlerInputs from pyatlan.model.apps.teradata_miner import TeradataMiner, TeradataMinerInputs @@ -124,6 +122,8 @@ "SnowflakeCrawlerInputs", "SnowflakeMiner", "SnowflakeMinerInputs", + "StandardLineage", + "StandardLineageInputs", "TeradataCrawler", "TeradataCrawlerInputs", "TeradataMiner", diff --git a/pyatlan/model/apps/standard_lineage.py b/pyatlan/model/apps/standard_lineage.py new file mode 100644 index 000000000..2c26abc64 --- /dev/null +++ b/pyatlan/model/apps/standard_lineage.py @@ -0,0 +1,287 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +"""Standard Lineage app — typed inputs + fluent builder. + +Hand-written rather than generated, for two reasons the configmap cannot express: + +* ``cross_connection_qualified_names`` is declared ``str`` in the app's input + contract but *means* a list of connection qualified names. The generator would + emit a bare ``str`` field and every caller would have to remember to + ``json.dumps`` it — sending a native list fails contract validation server-side. + :meth:`StandardLineage.connections` takes a ``List[str]`` and encodes it. +* Standard Lineage's defining operation is **re-scoping an existing workflow** + (adding a connection as it is onboarded), not creating a new one. That is an + update against a slug, and it has to preserve the workflow's own connection + entity — see :meth:`StandardLineage.add_connections`. + +Sourced from the app's UI configmap and input contract: + * inputs form : /api/service/configmaps/atlan-standard-lineage?entrypoint=standard-lineage + * input contract : /api/service/v1/apps/atlan-standard-lineage/inputs?entrypoint=standard-lineage + +There is no credential: the app reads query history already extracted by each +in-scope connection's own miner, so the contract's ``credential_ref`` and +``agent_json`` are optional and stay null. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, ClassVar, Dict, Iterable, List, Optional, Sequence, Union + +from ._base import AppBuilder, AppInput + +#: A connection qualified name is ``default/{connector}/{epoch}``. +_CONNECTION_QN = re.compile(r"^default/([^/]+)/\d+$") + +#: The connector of the workflow's *own* connection — distinct from the +#: ``connector`` input, which names the connector of the connections in scope. +_OWN_CONNECTOR = "standard-lineage" + + +def _connector_of(qualified_name: str) -> str: + """Return the connector segment of a connection qualified name.""" + match = _CONNECTION_QN.match(qualified_name) + if not match: + raise ValueError( + f"{qualified_name!r} is not a connection qualified name " + "(expected 'default/{connector}/{epoch}')" + ) + return match.group(1) + + +def _validate_scope(qualified_names: Sequence[str]) -> str: + """Validate a scope list and return the single connector it covers. + + The app requires a non-empty, same-connector scope and fails the run when that + does not hold, so this is checked client-side where the error is actionable. + """ + if not qualified_names: + raise ValueError( + "Standard Lineage needs at least one connection in scope; " + "to stop processing a connection, remove it and leave the rest, " + "or delete the workflow" + ) + connectors = {_connector_of(qn) for qn in qualified_names} + if len(connectors) > 1: + raise ValueError( + "every connection in scope must belong to the same connector, got " + f"{sorted(connectors)}; use one Standard Lineage workflow per connector" + ) + connector = connectors.pop() + if connector == _OWN_CONNECTOR: + raise ValueError( + "the scope must list the SOURCE connections to build lineage across " + f"(e.g. 'default/bigquery/1700000000'), not the workflow's own " + f"{_OWN_CONNECTOR!r} connection" + ) + return connector + + +def _parse_scope(value: Any) -> List[str]: + """Read a scope back out of a persisted workflow. + + The value is a JSON-encoded string on the wire but a native list once the + Automation Engine has rendered it into the DAG, so both shapes occur. + """ + if value is None: + return [] + if isinstance(value, str): + if not value.strip(): + return [] + try: + parsed = json.loads(value) + except json.JSONDecodeError: + # A bare single qualified name, tolerated rather than crashed on. + return [value.strip()] + return [str(v) for v in parsed] if isinstance(parsed, list) else [str(parsed)] + if isinstance(value, Iterable): + return [str(v) for v in value] + return [str(value)] + + +class StandardLineageInputs(AppInput): + """Typed inputs for the ``atlan-standard-lineage`` / ``standard-lineage`` app.""" + + _APP_ID: ClassVar[str] = "atlan-standard-lineage" + _ENTRYPOINT: ClassVar[Optional[str]] = "standard-lineage" + + # Step 1 · Connection — the workflow's OWN connection, under the + # ``standard-lineage`` connector. Created by the workflow on first run. + connection: Optional[Any] = None + + # Step 3 · Metadata + connector: str = "" + """Connector of the connections in scope (e.g. ``bigquery``).""" + cross_connection_qualified_names: str = "" + """JSON-encoded list of source connection qualified names. Prefer + :meth:`StandardLineage.connections`, which encodes a ``List[str]`` for you.""" + run_role: str = "standard-lineage" + """Fixed — this is what makes the lineage app take its cross-connection path.""" + + +class StandardLineage(AppBuilder): + """Fluent builder for the Standard Lineage (cross-connection lineage) app. + + Standard Lineage builds lineage *across* several connections of one connector, + using query history their own miners already extracted. One workflow owns a set + of connections; onboarding a new connection means adding it to that set. + + Create a workflow:: + + resp = ( + StandardLineage(client) + .connection(name="bigquery-cross-connection") + .connections([ + "default/bigquery/1700000000", + "default/bigquery/1700000001", + ]) + .run() + ) + + Add a connection to an existing workflow — the common case, and idempotent:: + + resp = StandardLineage(client).add_connections( + resp.slug, ["default/bigquery/1700000002"] + ) + + Also available: :meth:`remove_connections`, :meth:`set_connections` and the + read-only :meth:`get_connections`. + """ + + _APP_ID: ClassVar[str] = "atlan-standard-lineage" + _ENTRYPOINT: ClassVar[Optional[str]] = "standard-lineage" + #: The workflow's own connection is minted under this connector; the connector + #: of the connections in scope is the separate ``connector`` input. + _CONNECTOR_NAME: ClassVar[str] = _OWN_CONNECTOR + _CONNECTOR_CONFIG: ClassVar[str] = "" + _INPUTS_CLASS = StandardLineageInputs + _HIDDEN_DEFAULTS: ClassVar[Dict[str, Any]] = {"run_role": "standard-lineage"} + #: No credential — nothing is extracted from a source system directly. + _EXTRACTION_METHOD: ClassVar[str] = "" + + # ── Step 3 · Metadata ────────────────────────────────────────────────── + def connections( + self, + qualified_names: Union[Sequence[str], str], + *, + connector: Optional[str] = None, + ): + """Set the connections to build lineage across (create-time scope). + + :param qualified_names: source connection qualified names, e.g. + ``["default/bigquery/1700000000", ...]``. All must belong to the same + connector. A pre-encoded JSON string is accepted as-is. + :param connector: the scope's connector; derived from + ``qualified_names`` when omitted. + :raises ValueError: on an empty scope, a malformed qualified name, or a + scope spanning more than one connector. + + To change the scope of a workflow that already exists, use + :meth:`add_connections` / :meth:`remove_connections` / :meth:`set_connections` + — those preserve the workflow's own connection, which this does not know about. + """ + scope = _parse_scope(qualified_names) + derived = _validate_scope(scope) + self._metadata["connector"] = connector or derived + # The contract declares this field as `str`, so it goes over JSON-encoded; + # Heracles parses it back into a list for the manifest placeholder. + self._metadata["cross_connection_qualified_names"] = json.dumps(scope) + self._metadata["run_role"] = "standard-lineage" + return self + + # ── Re-scoping an existing workflow (network) ─────────────────────────── + def get_connections(self, slug: str) -> List[str]: + """Return the connections currently in scope for ``slug``. Read-only.""" + return _parse_scope( + self._persisted_args(slug).get("cross_connection_qualified_names") + ) + + def add_connections(self, slug: str, qualified_names: Union[Sequence[str], str]): + """Add connections to an existing workflow's scope, keeping the rest. + + Idempotent: connections already in scope are ignored, and when nothing + would change no version is published and ``None`` is returned. This is the + onboarding call — adding each new connection as it is created. + """ + current = self.get_connections(slug) + additions = [qn for qn in _parse_scope(qualified_names) if qn not in current] + if not additions: + return None + return self.set_connections(slug, current + additions) + + def remove_connections(self, slug: str, qualified_names: Union[Sequence[str], str]): + """Remove connections from an existing workflow's scope, keeping the rest. + + Idempotent in the same way as :meth:`add_connections`. Removing every + connection raises — the app cannot run on an empty scope. + + The next run hands each removed connection back to its own miner and + crawler, by flipping the per-connection standard-lineage marker off. + """ + current = self.get_connections(slug) + removals = set(_parse_scope(qualified_names)) + remaining = [qn for qn in current if qn not in removals] + if len(remaining) == len(current): + return None + return self.set_connections(slug, remaining) + + def set_connections( + self, + slug: str, + qualified_names: Union[Sequence[str], str], + *, + connector: Optional[str] = None, + ): + """Replace an existing workflow's scope with exactly ``qualified_names``. + + Preserves the workflow's own connection, its ``run_role`` and its identity + by reading them back from the persisted workflow — ``client.app.update`` is + a full replace, so anything omitted from the payload would be dropped from + the new version, and the workflow's connection entity is republished on + every run (a partial one would strip its admins). + """ + scope = _parse_scope(qualified_names) + derived = _validate_scope(scope) + args = self._persisted_args(slug) + own_connection = args.get("connection") + if not own_connection: + raise ValueError( + f"workflow {slug!r} has no connection on its extract node; refusing " + "to update, because a partial connection would be republished over " + "the real one" + ) + inputs = self._INPUTS_CLASS( + connection=own_connection, + connector=connector or args.get("connector") or derived, + cross_connection_qualified_names=json.dumps(scope), + run_role=args.get("run_role") or "standard-lineage", + ) + return self._client.app.update( + slug=slug, inputs=inputs, entrypoint=self._ENTRYPOINT + ) + + # ── internals ────────────────────────────────────────────────────────── + def _persisted_args(self, slug: str) -> Dict[str, Any]: + """Read the persisted workflow's extract-node args. + + ``GET /v1/app/{slug}`` returns the rendered DAG; ``AppSummary`` tolerates + unmodelled fields, so it arrives as an extra attribute. + """ + summary = self._client.app.get(slug) + dag = getattr(summary, "dag", None) + if not isinstance(dag, dict): + raise ValueError( + f"workflow {slug!r} returned no DAG to read the scope from" + ) + node = dag.get("extract") + args = ((node or {}).get("inputs") or {}).get("args") if node else None + if not isinstance(args, dict): + raise ValueError( + f"workflow {slug!r} has no extract node args; is it a Standard " + "Lineage workflow?" + ) + return args + + +__all__ = ["StandardLineage", "StandardLineageInputs"] diff --git a/tests/unit/apps/test_standard_lineage.py b/tests/unit/apps/test_standard_lineage.py new file mode 100644 index 000000000..1eff2f112 --- /dev/null +++ b/tests/unit/apps/test_standard_lineage.py @@ -0,0 +1,223 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +import json +from unittest.mock import Mock + +import pytest + +from pyatlan.model.apps import StandardLineage, StandardLineageInputs + +SLUG = "atlan-standard-lineage-1700000000-Abcd1234" +BQ1 = "default/bigquery/1700000001" +BQ2 = "default/bigquery/1700000002" +BQ3 = "default/bigquery/1700000003" + +# The workflow's own connection, as it comes back on the persisted DAG. Every +# attribute matters: the create-connection node republishes this entity, so a +# partial copy would strip the connection's name and admins in Atlan. +OWN_CONNECTION = { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/standard-lineage/1700000000", + "name": "bq-cross-connection", + "connectorName": "standard-lineage", + "category": "lineage", + "adminUsers": ["someone"], + "adminRoles": ["role-guid"], + "adminGroups": [], + "rowLimit": 10000, + }, +} + + +def _client(scope, *, connection=OWN_CONNECTION, connector="bigquery"): + """A mock client whose app.get() returns a persisted Standard Lineage DAG.""" + args = {"connector": connector, "run_role": "standard-lineage"} + if connection is not None: + args["connection"] = connection + if scope is not None: + args["cross_connection_qualified_names"] = scope + client = Mock() + client.app.get.return_value = Mock(dag={"extract": {"inputs": {"args": args}}}) + client.app.update.return_value = Mock(slug=SLUG, version=1700000009) + return client + + +def _sent_inputs(client): + """The inputs dict actually handed to client.app.update().""" + return client.app.update.call_args.kwargs["inputs"].to_inputs() + + +# ── inputs model ──────────────────────────────────────────────────────────── +def test_inputs_defaults(): + i = StandardLineageInputs() + assert StandardLineageInputs._APP_ID == "atlan-standard-lineage" + assert StandardLineageInputs._ENTRYPOINT == "standard-lineage" + assert i.connector == "" + assert i.cross_connection_qualified_names == "" + assert i.run_role == "standard-lineage" + + +def test_builder_class_vars(): + assert StandardLineage._APP_ID == "atlan-standard-lineage" + assert StandardLineage._ENTRYPOINT == "standard-lineage" + # The workflow's OWN connection lives under standard-lineage; the connections + # in scope are a different connector entirely. + assert StandardLineage._CONNECTOR_NAME == "standard-lineage" + + +# ── create-time scope ─────────────────────────────────────────────────────── +def test_connections_json_encodes_and_derives_connector(): + out = ( + StandardLineage(Mock()) + .connection(name="bq-cross-connection") + .connections([BQ1, BQ2]) + .preview() + ) + # Declared `str` in the contract — a native list fails validation server-side. + assert isinstance(out["cross_connection_qualified_names"], str) + assert json.loads(out["cross_connection_qualified_names"]) == [BQ1, BQ2] + assert out["connector"] == "bigquery" + assert out["run_role"] == "standard-lineage" + assert out["connection"]["attributes"]["connectorName"] == "standard-lineage" + + +def test_connections_explicit_connector_wins(): + out = ( + StandardLineage(Mock()) + .connections([BQ1], connector="bigquery-custom") + .preview() + ) + assert out["connector"] == "bigquery-custom" + + +def test_connections_accepts_a_preencoded_json_string(): + out = StandardLineage(Mock()).connections(json.dumps([BQ1, BQ2])).preview() + assert json.loads(out["cross_connection_qualified_names"]) == [BQ1, BQ2] + + +# ── validation ────────────────────────────────────────────────────────────── +def test_empty_scope_is_rejected(): + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(Mock()).connections([]) + + +def test_mixed_connector_scope_is_rejected(): + with pytest.raises(ValueError, match="same connector"): + StandardLineage(Mock()).connections([BQ1, "default/snowflake/1700000004"]) + + +def test_own_connection_as_scope_is_rejected(): + """A natural mistake: passing the workflow's own connection as its scope.""" + with pytest.raises(ValueError, match="not the workflow's own"): + StandardLineage(Mock()).connections(["default/standard-lineage/1700000000"]) + + +def test_malformed_qualified_name_is_rejected(): + with pytest.raises(ValueError, match="not a connection qualified name"): + StandardLineage(Mock()).connections(["bigquery/1700000001"]) + + +# ── reading an existing workflow ──────────────────────────────────────────── +def test_get_connections_reads_a_native_list(): + """Once the Automation Engine renders the DAG the value is a real list.""" + assert StandardLineage(_client([BQ1, BQ2])).get_connections(SLUG) == [BQ1, BQ2] + + +def test_get_connections_reads_a_json_string(): + """On the wire it is JSON-encoded, so both shapes have to be readable.""" + assert StandardLineage(_client(json.dumps([BQ1, BQ2]))).get_connections(SLUG) == [ + BQ1, + BQ2, + ] + + +def test_get_connections_on_empty_scope(): + assert StandardLineage(_client("")).get_connections(SLUG) == [] + + +def test_missing_extract_args_raises(): + client = Mock() + client.app.get.return_value = Mock(dag={"publish": {}}) + with pytest.raises(ValueError, match="no extract node args"): + StandardLineage(client).get_connections(SLUG) + + +# ── re-scoping ────────────────────────────────────────────────────────────── +def test_set_connections_preserves_the_persisted_connection_verbatim(): + """The whole reason set_connections reads before it writes. + + ``client.app.update`` is a full replace, and the connection entity is + republished on every run — so sending a rebuilt or partial connection would + overwrite the real one in Atlan. + """ + client = _client([BQ1, BQ2]) + StandardLineage(client).set_connections(SLUG, [BQ1, BQ2, BQ3]) + sent = _sent_inputs(client) + assert sent["connection"] == OWN_CONNECTION + assert sent["run_role"] == "standard-lineage" + assert json.loads(sent["cross_connection_qualified_names"]) == [BQ1, BQ2, BQ3] + assert client.app.update.call_args.kwargs["entrypoint"] == "standard-lineage" + assert client.app.update.call_args.kwargs["slug"] == SLUG + + +def test_set_connections_refuses_when_the_workflow_has_no_connection(): + client = _client([BQ1], connection=None) + with pytest.raises(ValueError, match="no connection on its extract node"): + StandardLineage(client).set_connections(SLUG, [BQ1, BQ2]) + client.app.update.assert_not_called() + + +def test_set_connections_refuses_to_empty_the_scope(): + client = _client([BQ1]) + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(client).set_connections(SLUG, []) + client.app.update.assert_not_called() + + +def test_add_connections_appends_and_keeps_the_rest(): + client = _client([BQ1, BQ2]) + StandardLineage(client).add_connections(SLUG, [BQ3]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ2, + BQ3, + ] + + +def test_add_connections_is_idempotent(): + """The onboarding portal may replay; a no-op must not publish a version.""" + client = _client([BQ1, BQ2]) + assert StandardLineage(client).add_connections(SLUG, [BQ2]) is None + client.app.update.assert_not_called() + + +def test_add_connections_adds_only_the_new_ones(): + client = _client([BQ1]) + StandardLineage(client).add_connections(SLUG, [BQ1, BQ2]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ2, + ] + + +def test_remove_connections_keeps_the_rest(): + client = _client([BQ1, BQ2, BQ3]) + StandardLineage(client).remove_connections(SLUG, [BQ2]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ3, + ] + + +def test_remove_connections_not_in_scope_is_a_noop(): + client = _client([BQ1, BQ2]) + assert StandardLineage(client).remove_connections(SLUG, [BQ3]) is None + client.app.update.assert_not_called() + + +def test_remove_last_connection_is_refused(): + client = _client([BQ1]) + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(client).remove_connections(SLUG, [BQ1]) + client.app.update.assert_not_called() diff --git a/tests/unit/test_app_generated_inputs.py b/tests/unit/test_app_generated_inputs.py index 06888672c..ed6d5757c 100644 --- a/tests/unit/test_app_generated_inputs.py +++ b/tests/unit/test_app_generated_inputs.py @@ -88,6 +88,9 @@ def test_no_internal_fields_leak(cls): "BigqueryCrawlerInputs", "DatabricksCrawlerInputs", "KafkaConfluentInputs", + # Standard Lineage re-scopes an EXISTING workflow and JSON-encodes a list into a + # contract field declared `str`; neither is expressible from a configmap. + "StandardLineageInputs", }