From 73c341717d289f87f98d67bee010696a21ccf3c4 Mon Sep 17 00:00:00 2001 From: Aryamanz29 Date: Tue, 18 Aug 2026 16:47:28 +0530 Subject: [PATCH 1/8] feat(apps)!: type the BigQuery crawler WIF credential inputs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BREAKING CHANGE: workload_identity_federation() now requires service_account_email, wif_pool_provider_id, atlan_oauth_id and atlan_oauth_secret as typed keyword-only parameters. Previously only project_id was typed and the rest were funnelled through **extra — undiscoverable, and a misspelled key was silently accepted, producing a broken gcp-wif credential with no error. A missing/misspelled field is now a TypeError. **extra is retained for forward-compatibility with fields newer than this signature. Matches the fully-typed sibling service_account(). Also preserve hand-written app tests (test_bigquery_crawler.py) across generate_apps regeneration, mirroring how hand-written modules are kept. --- pyatlan/model/apps/bigquery_crawler.py | 33 ++++++++- tests/unit/apps/test_bigquery_crawler.py | 93 ++++++++++++++++++++++++ tests/unit/test_app_builders.py | 8 +- 3 files changed, 131 insertions(+), 3 deletions(-) create mode 100644 tests/unit/apps/test_bigquery_crawler.py diff --git a/pyatlan/model/apps/bigquery_crawler.py b/pyatlan/model/apps/bigquery_crawler.py index 035e9d8b6..e58bf4207 100644 --- a/pyatlan/model/apps/bigquery_crawler.py +++ b/pyatlan/model/apps/bigquery_crawler.py @@ -132,12 +132,41 @@ def workload_identity_federation( self, *, project_id: str, + service_account_email: str, + wif_pool_provider_id: str, + atlan_oauth_id: str, + atlan_oauth_secret: str, connectivity: str = "public", host: Optional[str] = None, port: Optional[int] = None, **extra: Any, ) -> "BigqueryCrawler": - """Direct extraction with Workload Identity Federation auth.""" + """Direct extraction with Workload Identity Federation (keyless) auth. + + Atlan mints a token from its own OAuth client and exchanges it through the + WIF pool/provider to impersonate ``service_account_email`` — so no Service + Account key is stored. All five values are required by the ``gcp-wif`` + credential; a missing one is a ``TypeError`` rather than a silently broken + credential. Any additional keyword rides through into ``extra`` unchanged + (forward-compatible with fields newer than this signature). + + :param project_id: GCP project ID. + :param service_account_email: Service Account that Atlan impersonates via WIF. + :param wif_pool_provider_id: full Workload Identity Pool *provider* resource id. + :param atlan_oauth_id: Atlan OAuth client id used for the token exchange. + :param atlan_oauth_secret: Atlan OAuth client secret (vaulted, never logged). + :param connectivity: ``public`` (Google's endpoint) or ``private`` (PSC). + :param host: Private Service Connect host (required for ``private``). + """ + wif: Dict[str, Any] = { + "project_id": project_id, + "connect_type": connectivity, + "service_account_email": service_account_email, + "wif_pool_provider_id": wif_pool_provider_id, + "atlan_oauth_id": atlan_oauth_id, + "atlan_oauth_secret": atlan_oauth_secret, + } + wif.update(extra) return self._stage_credential( "credential_guid", Credential( @@ -146,7 +175,7 @@ def workload_identity_federation( auth_type="gcp-wif", host=host or self._DEFAULT_HOST, port=port or self._DEFAULT_PORT, - extra={"project_id": project_id, "connect_type": connectivity, **extra}, + extra=wif, ), ) diff --git a/tests/unit/apps/test_bigquery_crawler.py b/tests/unit/apps/test_bigquery_crawler.py new file mode 100644 index 000000000..dfd9927f2 --- /dev/null +++ b/tests/unit/apps/test_bigquery_crawler.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +# Hand-written: bigquery_crawler is a hand-authored builder (see generate_apps +# _HAND_WRITTEN), so this test is preserved across regeneration. +from unittest.mock import Mock + +import pytest + +from pyatlan.model.apps import BigqueryCrawler, BigqueryCrawlerInputs + +WIF = dict( + project_id="my-project", + service_account_email="svc@my-project.iam.gserviceaccount.com", + wif_pool_provider_id="projects/1/locations/global/workloadIdentityPools/p/providers/pr", + atlan_oauth_id="oauth-client-id", + atlan_oauth_secret="oauth-secret", +) + + +def test_bigquery_crawler_inputs_defaults(): + i = BigqueryCrawlerInputs() + assert BigqueryCrawlerInputs._APP_ID == "bigquery-crawler" + assert BigqueryCrawlerInputs._ENTRYPOINT == "crawler" + assert i.include_filter == "{}" + assert i.exclude_filter == "{}" + assert i.enable_nested_columns is True + assert i.filter_sharded_tables is True + + +def test_bigquery_crawler_builder_payload(): + out = ( + BigqueryCrawler(Mock()) + .service_account( + email="svc@my-project.iam.gserviceaccount.com", + service_account_json="{}", + project_id="my-project", + ) + .connection(name="conn", admin_users=["u"]) + .include({"my-project": ["analytics"]}) + .preview() + ) + assert out["connection"]["attributes"]["connectorName"] == "bigquery" + assert out["extraction_method"] == "direct" + assert out["include_filter"] == '{"^my-project$": ["^analytics$"]}' + + +def test_wif_stages_gcp_wif_credential_with_the_five_typed_keys(): + b = BigqueryCrawler(Mock()).workload_identity_federation(**WIF) + cred = b._raw_creds["credential_guid"] + assert cred.auth_type == "gcp-wif" + assert cred.connector_config_name == "atlan-connectors-bigquery" + # WIF carries no username/password — auth is the token exchange. + assert cred.username is None and cred.password is None + assert cred.extras == { + "project_id": "my-project", + "connect_type": "public", + "service_account_email": "svc@my-project.iam.gserviceaccount.com", + "wif_pool_provider_id": "projects/1/locations/global/workloadIdentityPools/p/providers/pr", + "atlan_oauth_id": "oauth-client-id", + "atlan_oauth_secret": "oauth-secret", + } + + +def test_wif_keys_land_in_the_previewed_credential_payload(): + out = ( + BigqueryCrawler(Mock()) + .workload_identity_federation(**WIF) + .connection(name="conn") + .preview() + ) + cred = out["credential"] + assert cred["authType"] == "gcp-wif" + assert set(cred["extra"]) == { + "project_id", + "connect_type", + "service_account_email", + "wif_pool_provider_id", + "atlan_oauth_id", + "atlan_oauth_secret", + } + + +def test_wif_requires_every_credential_field(): + # Each WIF field is required; a missing one is a TypeError, not a silently + # incomplete credential. A misspelled required field trips the same check. + with pytest.raises(TypeError): + BigqueryCrawler(Mock()).workload_identity_federation(project_id="p") + + +def test_wif_forwards_unknown_extra_keys(): + # Forward-compatible: keys newer than this signature ride through **extra. + b = BigqueryCrawler(Mock()).workload_identity_federation(**WIF, future_flag="x") + assert b._raw_creds["credential_guid"].extras["future_flag"] == "x" diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 3d90ec7a4..48113d9b8 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -158,7 +158,13 @@ def test_service_account_credential_shape(): def test_workload_identity_federation_auth_type(): - b = BigqueryCrawler(Mock()).workload_identity_federation(project_id="proj") + b = BigqueryCrawler(Mock()).workload_identity_federation( + project_id="proj", + service_account_email="svc@proj.iam.gserviceaccount.com", + wif_pool_provider_id="pool/provider", + atlan_oauth_id="oauth-id", + atlan_oauth_secret="oauth-secret", + ) cred = b._raw_creds["credential_guid"] assert cred.auth_type == "gcp-wif" assert cred.extras["project_id"] == "proj" From ddb273d2deb42e4022a0c60e07257abc3c8e9e66 Mon Sep 17 00:00:00 2001 From: Aryamanz29 Date: Wed, 19 Aug 2026 10:02:07 +0530 Subject: [PATCH 2/8] feat(apps): add load()/update() for full-replace app-workflow updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit app.update() is a full replace with no read-back, so callers had to hand-build the entire input set — and silently dropping a server-derived field (notably connection_qualified_name, which create injects but update does not) makes the update return 200 while the next run fails downstream. AppBuilder now offers the same fluent shape as create for updates: BigqueryCrawler(client).load(slug).include({...}).update() load() seeds the builder from the workflow's current inputs (connection, existing credential, hidden defaults, all other fields); update() full-replaces while re-injecting connection_qualified_name, normalizing string-typed contract fields (e.g. control_config returned as an object), and referencing the existing credential by guid — never re-vaulting or rotating it. Generic on the base, so every app builder inherits it. --- pyatlan/model/apps/_base.py | 102 ++++++++++++++++++++++++++++++++ tests/unit/test_app_builders.py | 48 +++++++++++++++ 2 files changed, 150 insertions(+) diff --git a/pyatlan/model/apps/_base.py b/pyatlan/model/apps/_base.py index a2ca6db2c..02071bfa8 100644 --- a/pyatlan/model/apps/_base.py +++ b/pyatlan/model/apps/_base.py @@ -25,6 +25,10 @@ from pyatlan.model.credential import Credential +# Handshake/runtime ids the server injects into a workflow's inputs; they must +# not be echoed back on update (stripped server-side on create). +_RUNTIME_KEYS = frozenset({"user-id", "user_id", "workflow_id", "correlation_id"}) + class AppInput(BaseModel): """A typed, configmap-derived ``inputs`` payload for an app workflow.""" @@ -109,6 +113,7 @@ def __init__(self, client: Any): self._admin_groups: List[str] = [] self._admin_roles: List[str] = [] self._metadata: Dict[str, Any] = {} + self._update_slug: Optional[str] = None # ── Step 1 · Credential ──────────────────────────────────────────────── def _stage_credential(self, field: str, credential: Credential): @@ -267,6 +272,103 @@ def run(self, *, name: Optional[str] = None, schedule: Optional[Any] = None): """Create the workflow **and** submit a run immediately (``run=True``).""" return self._create(name=name, run=True, schedule=schedule) + # ── update (load an existing workflow, change fields, full-replace) ───── + def load(self, slug: str) -> "AppBuilder": + """Seed this builder from an existing workflow's current inputs, so + :meth:`update` can change a few fields and preserve the rest. + + Mirrors the create builders — same fluent methods, one extra step:: + + Builder(client).load(slug).include({...}).update() + + The connection and the existing credential are read from the workflow and + reused: the credential is referenced by guid and **never re-vaulted or + rotated**. Everything you don't change is carried through untouched. + """ + args = ( + self._client.app.get(slug) + .dict() + .get("dag", {}) + .get("extract", {}) + .get("inputs", {}) + .get("args", {}) + ) + self._update_slug = slug + conn = args.get("connection") + conn = json.loads(conn) if isinstance(conn, str) else conn + attrs = (conn or {}).get("attributes", {}) if isinstance(conn, dict) else {} + self._connection_qualified_name = attrs.get("qualifiedName") + self._connection_name = attrs.get("name") + self._admin_users = list(attrs.get("adminUsers") or []) + self._admin_groups = list(attrs.get("adminGroups") or []) + self._admin_roles = list(attrs.get("adminRoles") or []) + self._credential_guid = args.get("credential_guid") or "" + self._extraction_method = args.get("extraction_method") or self._EXTRACTION_METHOD + # Carry every non-structural current field so nothing is dropped on the + # full replace; the structural bits are rebuilt from the state above. + structural = { + "connection", + "connection_qualified_name", + "credential", + "credential_guid", + "agent_json", + "extraction_method", + } | _RUNTIME_KEYS + self._metadata = {k: v for k, v in args.items() if k not in structural} + return self + + def update(self): + """Publish a new version of the loaded workflow (full-replace) with the + builder's current fields. Call :meth:`load` first. + + Re-injects ``connection_qualified_name`` (the read-back omits it, but the + run needs it) and re-encodes string-typed contract fields returned as + objects — so a caller only sets what they want to change. + """ + if not self._update_slug: + raise ValueError("call load(slug) before update()") + # Normalize BEFORE assembly — the typed inputs model rejects a string + # field (e.g. control_config) that the read-back returned as an object. + self._metadata = self._normalize_string_inputs(dict(self._metadata)) + epoch = int(time.time()) + qn = self._connection_qualified_name or f"default/{self._CONNECTOR_NAME}/{epoch}" + payload = self._assemble(qualified_name=qn, epoch=epoch).to_inputs() + payload["connection_qualified_name"] = qn + + def _update_with(entrypoint: Optional[str]): + return self._client.app.update( + slug=self._update_slug, inputs=payload, entrypoint=entrypoint or None + ) + + # Same 1003 fallback as _create(): retry at the default slot if the named + # entrypoint has no registered contract. + ep = self._ENTRYPOINT or None + try: + return _update_with(ep) + except Exception as exc: # noqa: BLE001 + msg = str(exc) + if ep is not None and ("1003" in msg or "unknown entrypoint" in msg): + return _update_with(None) + raise + + def _normalize_string_inputs(self, payload: Dict[str, Any]) -> Dict[str, Any]: + """JSON-encode string-typed contract fields the read-back returned as + objects (e.g. ``control_config``), so the full-replace matches the contract.""" + try: + contract = self._client.app.get_input_contract( + self._APP_ID, self._ENTRYPOINT or None + ) + except Exception: # noqa: BLE001 - best-effort; skip if the contract can't load + return payload + for key, spec in (contract.properties or {}).items(): + if ( + isinstance(spec, dict) + and spec.get("type") == "string" + and isinstance(payload.get(key), (dict, list)) + ): + payload[key] = json.dumps(payload[key]) + return payload + def _vault_credential(self, cred: Credential) -> str: """Vault a raw credential into Atlan's secret store and return its guid. diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 48113d9b8..ac4288f61 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -264,3 +264,51 @@ def test_agent_mode_uses_agent_json_not_credential(client): assert out["agent_json"] == {"name": "my-agent"} assert "credential" not in out assert "credential_guid" not in out + + +# --------------------------------------------------------------------------- # +# update() — load an existing workflow, change a field, full-replace +# --------------------------------------------------------------------------- # +def test_load_update_preserves_reinjects_and_references_credential(): + """load(slug)..update() re-injects connection_qualified_name, keeps + the existing credential (referenced, not rotated), normalizes string-typed + fields, drops runtime keys, and preserves everything else.""" + from types import SimpleNamespace + + client = Mock() + current = { + "connection": {"typeName": "Connection", "attributes": { + "qualifiedName": "default/bigquery/123", "connectorName": "bigquery", + "name": "prod", "adminRoles": ["role-1"]}}, + "credential_guid": "cred-1", + "extraction_method": "direct", + "include_filter": {"^proj$": ["^old$"]}, + "control_config": {}, # object on read-back -> must normalize to "{}" + "user-id": "u1", "workflow_id": "w1", # runtime keys -> must be dropped + "atlas_auth_type": "internal", # non-structural -> preserved + } + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace( + properties={"control_config": {"type": "string"}} + ) + client.app.update.return_value = Mock(version=2) + + BigqueryCrawler(client).load("slug-1").include({"proj": ["new"]}).update() + + kw = client.app.update.call_args.kwargs + inp = kw["inputs"] + assert kw["slug"] == "slug-1" and kw["entrypoint"] == "crawler" + assert inp["connection_qualified_name"] == "default/bigquery/123" # re-injected + assert inp["credential_guid"] == "cred-1" and "credential" not in inp # no rotation + assert inp["include_filter"] == '{"^proj$": ["^new$"]}' # changed + anchored + assert inp["control_config"] == "{}" # normalized object -> string + assert "user-id" not in inp and "workflow_id" not in inp # runtime keys dropped + assert inp["atlas_auth_type"] == "internal" # preserved + assert inp["connection"]["attributes"]["adminRoles"] == ["role-1"] # connection kept + + +def test_update_without_load_raises(): + with pytest.raises(ValueError): + BigqueryCrawler(Mock()).update() From c996f7692a3af374c6daa5275da15011868a03a8 Mon Sep 17 00:00:00 2001 From: Aryamanz29 Date: Wed, 19 Aug 2026 10:23:03 +0530 Subject: [PATCH 3/8] test(apps): cover load()/update() across builders, multi-field changes, 1003 retry - generic across every builder (load().update() re-injects connection_qualified_name, references the existing credential, targets the right slug/entrypoint) - multiple typed field changes in one update (include/exclude/regex/toggles) apply while the rest is preserved and runtime keys are dropped - update() falls back to the default entrypoint on a 1003 (no registered contract) --- tests/unit/test_app_builders.py | 116 ++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index ac4288f61..4c6fcc3ab 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -312,3 +312,119 @@ def test_load_update_preserves_reinjects_and_references_credential(): def test_update_without_load_raises(): with pytest.raises(ValueError): BigqueryCrawler(Mock()).update() + + +@pytest.mark.parametrize("cls", BUILDERS, ids=BUILDER_IDS) +def test_load_update_is_generic_across_builders(cls): + """load()/update() lives on the base, so every app builder can update: it + re-injects connection_qualified_name and references the existing credential + (no rotation), targeting the right slug/entrypoint. The builder's own + preview() is used as the 'current inputs' so the seed is valid for its + typed inputs model.""" + from types import SimpleNamespace + + qn = f"default/{cls._CONNECTOR_NAME}/1700000000" + current = cls(Mock()).connection(qualified_name=qn).credential_guid("cred-x").preview() + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace(properties={}) + client.app.update.return_value = Mock(version=2) + + cls(client).load("slug-x").update() + + kw = client.app.update.call_args.kwargs + inp = kw["inputs"] + assert kw["slug"] == "slug-x" + assert kw["entrypoint"] == (cls._ENTRYPOINT or None) + assert inp["connection_qualified_name"] == qn # re-injected + assert inp["credential_guid"] == "cred-x" # referenced, not rotated + assert "credential" not in inp # no raw credential re-sent + + +def test_update_retries_without_entrypoint_on_1003(): + """update() falls back to the default entrypoint when the named one has no + registered contract (server 1003) — mirroring _create().""" + from types import SimpleNamespace + + current = ( + BigqueryCrawler(Mock()) + .connection(qualified_name="default/bigquery/1") + .credential_guid("g") + .preview() + ) + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace(properties={}) + client.app.update.side_effect = [ + Exception("Server responded 1003: unknown entrypoint"), + Mock(version=3), + ] + + BigqueryCrawler(client).load("slug-x").update() + + assert client.app.update.call_count == 2 + assert client.app.update.call_args_list[0].kwargs["entrypoint"] == "crawler" + assert client.app.update.call_args_list[1].kwargs["entrypoint"] is None + + +def test_load_update_applies_multiple_field_changes(): + """load(slug)..update() applies every change and + preserves + re-injects the rest (connection_qualified_name, credential ref, + hidden defaults), dropping runtime keys.""" + from types import SimpleNamespace + + current = { + "connection": {"typeName": "Connection", "attributes": { + "qualifiedName": "default/bigquery/1", "connectorName": "bigquery", + "name": "prod", "adminRoles": ["role-1"]}}, + "credential_guid": "cred-1", "extraction_method": "direct", + "include_filter": {"^p$": ["^old$"]}, "exclude_filter": {}, + "temp_table_regex": "", "enable_nested_columns": True, + "enable_bigquery_tag_sync": False, "filter_sharded_tables": True, + "hidden_datasets": False, "control_config": {}, + "control_config_strategy": "default", "atlas_auth_type": "internal", + "user-id": "u1", + } + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace( + properties={"control_config": {"type": "string"}} + ) + client.app.update.return_value = Mock(version=2) + + ( + BigqueryCrawler(client) + .load("slug-1") + .include({"p": ["a", "b"]}) + .exclude({"p": ["tmp"]}) + .exclude_regex(".*_bak$") + .import_nested_columns(False) + .import_tags(True) + .combine_sharded_tables(False) + .hidden_assets(True) + .custom_config('{"flag": 1}') + .update() + ) + + inp = client.app.update.call_args.kwargs["inputs"] + # every requested change applied + assert inp["include_filter"] == '{"^p$": ["^a$", "^b$"]}' + assert inp["exclude_filter"] == '{"^p$": ["^tmp$"]}' + assert inp["temp_table_regex"] == ".*_bak$" + assert inp["enable_nested_columns"] is False + assert inp["enable_bigquery_tag_sync"] is True + assert inp["filter_sharded_tables"] is False + assert inp["hidden_datasets"] is True + assert inp["control_config_strategy"] == "custom" + assert inp["control_config"] == '{"flag": 1}' + # rest preserved / re-injected / stripped + assert inp["connection_qualified_name"] == "default/bigquery/1" + assert inp["credential_guid"] == "cred-1" and "credential" not in inp + assert inp["atlas_auth_type"] == "internal" + assert "user-id" not in inp From 44fa9e75ef4a6965d49513389ff97b398a620422 Mon Sep 17 00:00:00 2001 From: Aryamanz29 Date: Wed, 19 Aug 2026 10:42:17 +0530 Subject: [PATCH 4/8] style: ruff format load()/update() + tests (qa-checks) --- pyatlan/model/apps/_base.py | 8 +++-- tests/unit/test_app_builders.py | 63 ++++++++++++++++++++++----------- 2 files changed, 49 insertions(+), 22 deletions(-) diff --git a/pyatlan/model/apps/_base.py b/pyatlan/model/apps/_base.py index 02071bfa8..242bf4bcd 100644 --- a/pyatlan/model/apps/_base.py +++ b/pyatlan/model/apps/_base.py @@ -303,7 +303,9 @@ def load(self, slug: str) -> "AppBuilder": self._admin_groups = list(attrs.get("adminGroups") or []) self._admin_roles = list(attrs.get("adminRoles") or []) self._credential_guid = args.get("credential_guid") or "" - self._extraction_method = args.get("extraction_method") or self._EXTRACTION_METHOD + self._extraction_method = ( + args.get("extraction_method") or self._EXTRACTION_METHOD + ) # Carry every non-structural current field so nothing is dropped on the # full replace; the structural bits are rebuilt from the state above. structural = { @@ -331,7 +333,9 @@ def update(self): # field (e.g. control_config) that the read-back returned as an object. self._metadata = self._normalize_string_inputs(dict(self._metadata)) epoch = int(time.time()) - qn = self._connection_qualified_name or f"default/{self._CONNECTOR_NAME}/{epoch}" + qn = ( + self._connection_qualified_name or f"default/{self._CONNECTOR_NAME}/{epoch}" + ) payload = self._assemble(qualified_name=qn, epoch=epoch).to_inputs() payload["connection_qualified_name"] = qn diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 4c6fcc3ab..53d7eeb4c 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -277,15 +277,22 @@ def test_load_update_preserves_reinjects_and_references_credential(): client = Mock() current = { - "connection": {"typeName": "Connection", "attributes": { - "qualifiedName": "default/bigquery/123", "connectorName": "bigquery", - "name": "prod", "adminRoles": ["role-1"]}}, + "connection": { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/bigquery/123", + "connectorName": "bigquery", + "name": "prod", + "adminRoles": ["role-1"], + }, + }, "credential_guid": "cred-1", "extraction_method": "direct", "include_filter": {"^proj$": ["^old$"]}, - "control_config": {}, # object on read-back -> must normalize to "{}" - "user-id": "u1", "workflow_id": "w1", # runtime keys -> must be dropped - "atlas_auth_type": "internal", # non-structural -> preserved + "control_config": {}, # object on read-back -> must normalize to "{}" + "user-id": "u1", + "workflow_id": "w1", # runtime keys -> must be dropped + "atlas_auth_type": "internal", # non-structural -> preserved } client.app.get.return_value.dict.return_value = { "dag": {"extract": {"inputs": {"args": current}}} @@ -306,7 +313,9 @@ def test_load_update_preserves_reinjects_and_references_credential(): assert inp["control_config"] == "{}" # normalized object -> string assert "user-id" not in inp and "workflow_id" not in inp # runtime keys dropped assert inp["atlas_auth_type"] == "internal" # preserved - assert inp["connection"]["attributes"]["adminRoles"] == ["role-1"] # connection kept + assert inp["connection"]["attributes"]["adminRoles"] == [ + "role-1" + ] # connection kept def test_update_without_load_raises(): @@ -324,7 +333,9 @@ def test_load_update_is_generic_across_builders(cls): from types import SimpleNamespace qn = f"default/{cls._CONNECTOR_NAME}/1700000000" - current = cls(Mock()).connection(qualified_name=qn).credential_guid("cred-x").preview() + current = ( + cls(Mock()).connection(qualified_name=qn).credential_guid("cred-x").preview() + ) client = Mock() client.app.get.return_value.dict.return_value = { "dag": {"extract": {"inputs": {"args": current}}} @@ -338,9 +349,9 @@ def test_load_update_is_generic_across_builders(cls): inp = kw["inputs"] assert kw["slug"] == "slug-x" assert kw["entrypoint"] == (cls._ENTRYPOINT or None) - assert inp["connection_qualified_name"] == qn # re-injected - assert inp["credential_guid"] == "cred-x" # referenced, not rotated - assert "credential" not in inp # no raw credential re-sent + assert inp["connection_qualified_name"] == qn # re-injected + assert inp["credential_guid"] == "cred-x" # referenced, not rotated + assert "credential" not in inp # no raw credential re-sent def test_update_retries_without_entrypoint_on_1003(): @@ -378,15 +389,27 @@ def test_load_update_applies_multiple_field_changes(): from types import SimpleNamespace current = { - "connection": {"typeName": "Connection", "attributes": { - "qualifiedName": "default/bigquery/1", "connectorName": "bigquery", - "name": "prod", "adminRoles": ["role-1"]}}, - "credential_guid": "cred-1", "extraction_method": "direct", - "include_filter": {"^p$": ["^old$"]}, "exclude_filter": {}, - "temp_table_regex": "", "enable_nested_columns": True, - "enable_bigquery_tag_sync": False, "filter_sharded_tables": True, - "hidden_datasets": False, "control_config": {}, - "control_config_strategy": "default", "atlas_auth_type": "internal", + "connection": { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/bigquery/1", + "connectorName": "bigquery", + "name": "prod", + "adminRoles": ["role-1"], + }, + }, + "credential_guid": "cred-1", + "extraction_method": "direct", + "include_filter": {"^p$": ["^old$"]}, + "exclude_filter": {}, + "temp_table_regex": "", + "enable_nested_columns": True, + "enable_bigquery_tag_sync": False, + "filter_sharded_tables": True, + "hidden_datasets": False, + "control_config": {}, + "control_config_strategy": "default", + "atlas_auth_type": "internal", "user-id": "u1", } client = Mock() From a2fba3e6b97ff12f0c1ccefa4b8b57698bd09062 Mon Sep 17 00:00:00 2001 From: hariharanatlan Date: Thu, 13 Aug 2026 14:16:32 +0530 Subject: [PATCH 5/8] fix(apps): send a reused credential guid on the connection, not top-level (CONNECT-843) When an app builder references an EXISTING connection by qualifiedName and reuses an already-vaulted credential guid (miners, or any builder given .credential_guid() together with .connection(qualified_name=...)), pyatlan put that guid in the top-level `credential_guid` field with no credential body. The create endpoint's credential resolver treats a non-empty top-level credential_guid as "reuse this guid", but only builds a credential body from a payload field carrying a non-empty authType. Given a guid and no body it still runs UpsertCredentialConfig(guid, {"credentialSource": "direct"}), replacing that credential's shared config record with that single key. Every workflow sharing the guid then loses authType/host/extra, so auth-type-sensitive connectors (e.g. gcp-wif) fail on their next run. Route the reused guid to connection.attributes.defaultCredentialGuid and keep credential_guid "", which is exactly the shape the UI sends for reuse flows, so the resolver takes its "no guid" path and never touches the record. Staged raw credentials, new-connection creation with an explicit guid, and the agent/SDR path are all unchanged (verified byte-for-byte on the assembled payload). Introduced by 512fb3959, which added the QN credential auto-resolution; the commit 27 minutes before it (d58bf3531) deliberately sent credential_guid="". This removes pyatlan as a trigger, not the defect itself. The weak guard is heracles handler/native_app.go:635 (and handler/workflow.go:2959) and should match the already-correct update path at native_app.go:1020, `credBodyForUpsert != nil && credentialGUID != ""`. Any other bare-guid caller still re-corrupts the record. tests/unit/test_app_builders.py::test_miner_auto_resolves_connection_credential is updated rather than left alone: it was added by 512fb3959, the same commit that introduced the defect, so its top-level-guid assertion locked in the regression instead of protecting intended behaviour. It still proves the connection is looked up and its guid reused, now asserted at the correct location, and gained an assertion rather than losing one. Co-Authored-By: Claude Opus 5 (1M context) --- pyatlan/model/apps/_base.py | 36 ++++++++-- tests/unit/test_app_builders.py | 115 +++++++++++++++++++++++++++++++- 2 files changed, 146 insertions(+), 5 deletions(-) diff --git a/pyatlan/model/apps/_base.py b/pyatlan/model/apps/_base.py index 242bf4bcd..c366474ef 100644 --- a/pyatlan/model/apps/_base.py +++ b/pyatlan/model/apps/_base.py @@ -161,7 +161,12 @@ def connection( return self # ── assembly (no network) ────────────────────────────────────────────── - def _build_connection(self, qualified_name: str) -> Dict[str, Any]: + def _build_connection( + self, + qualified_name: str, + *, + default_credential_guid: Optional[str] = None, + ) -> Dict[str, Any]: # Derive the connector from the QN (``default/{connector}/{epoch}``) so a # referenced existing connection (e.g. for miners) reports the right # connectorName, not the builder's app-id-derived fallback. @@ -175,6 +180,8 @@ def _build_connection(self, qualified_name: str) -> Dict[str, Any]: "qualifiedName": qualified_name, "connectorName": connector, } + if default_credential_guid: + attrs["defaultCredentialGuid"] = default_credential_guid if self._connection_name: attrs["name"] = self._connection_name if self._admin_users: @@ -219,9 +226,27 @@ def _assemble( resolved_guids: Optional[Dict[str, str]] = None, ) -> AppInput: resolved_guids = resolved_guids or {} + # Reusing an already-vaulted credential on an EXISTING connection (e.g. a + # miner picking up that connection's own credential): the guid rides on the + # connection entity, the way the UI sends it, and the top-level + # credential_guid stays "". A bare top-level guid with no credential body + # makes the create endpoint rewrite that credential's shared config record + # from the (absent) body, flattening it to {"credentialSource": "direct"} + # for every workflow sharing the guid (CONNECT-843). + reuse_on_existing_connection = bool( + self._extraction_method != "agent" + and self._credential_guid + and self._connection_qualified_name + and "credential_guid" not in self._raw_creds + ) kwargs: Dict[str, Any] = dict(self._HIDDEN_DEFAULTS) kwargs.update(self._metadata) - kwargs["connection"] = self._build_connection(qualified_name) + kwargs["connection"] = self._build_connection( + qualified_name, + default_credential_guid=( + self._credential_guid if reuse_on_existing_connection else None + ), + ) kwargs["extraction_method"] = self._extraction_method if self._extraction_method == "agent": kwargs["agent_json"] = self._agent_json @@ -245,9 +270,12 @@ def _assemble( else: kwargs[field] = self._raw_credential(cred, epoch=epoch, redact=True) # credential_guid is a (non-null) string in the contract: reuse an existing - # guid if given, else "" (omitting it reads as null and is rejected). + # guid if given, else "" (omitting it reads as null and is rejected). When + # the guid rides on the connection instead, this stays "" (see above). kwargs["credential_guid"] = ( - self._credential_guid if self._credential_guid is not None else "" + "" + if reuse_on_existing_connection or self._credential_guid is None + else self._credential_guid ) return self._INPUTS_CLASS(**kwargs) diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 53d7eeb4c..8060f991d 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -249,7 +249,120 @@ def test_miner_auto_resolves_connection_credential(client): SnowflakeMiner(client).connection(qualified_name="default/snowflake/123").create() assert client.asset.search.called # connection was looked up out = client.app.create.call_args.kwargs["inputs"].to_inputs() - assert out["credential_guid"] == "conn-cred-guid" # its credential reused + # CONNECT-843: the reused guid rides on the connection entity (the UI's wire + # shape), never as a bare top-level credential_guid. A top-level guid with no + # credential body makes the create endpoint rewrite that credential's shared + # config record. Do not "fix" this back to out["credential_guid"]. + attrs = out["connection"]["attributes"] + assert attrs["defaultCredentialGuid"] == "conn-cred-guid" # its credential reused + assert out["credential_guid"] == "" # and not duplicated at the top level + + +# --------------------------------------------------------------------------- # +# CONNECT-843: where a REUSED credential guid is allowed to ride +# +# Reusing an already-vaulted guid on an existing connection must send the guid +# on the connection entity (the UI's wire shape) and leave top-level +# credential_guid "". A bare top-level guid with no credential body makes the +# create endpoint rewrite that credential's shared config record down to +# {"credentialSource": "direct"}, breaking every workflow sharing the guid. +# --------------------------------------------------------------------------- # +def _resolve_to(client, guid): + """Make the connection lookup in _create() resolve to ``guid``.""" + client.asset.search.return_value = iter([Mock(default_credential_guid=guid)]) + + +@pytest.mark.parametrize( + "cls, connector", + [(apps.BigqueryMiner, "bigquery"), (apps.SnowflakeMiner, "snowflake")], + ids=["bigquery", "snowflake"], +) +def test_auto_resolved_guid_rides_on_connection_not_top_level(client, cls, connector): + # Generic across connectors: the base builder owns this, no connector + # overrides _build_connection/_assemble. + _resolve_to(client, "resolved-guid") + cls(client).connection(qualified_name=f"default/{connector}/123").create() + out = client.app.create.call_args.kwargs["inputs"].to_inputs() + attrs = out["connection"]["attributes"] + # exactly the UI's reuse shape: identity + the connection's own credential + assert attrs["defaultCredentialGuid"] == "resolved-guid" + assert attrs["qualifiedName"] == f"default/{connector}/123" + assert attrs["connectorName"] == connector + # the guid is NOT echoed at the top level, and no credential body is invented + assert out["credential_guid"] == "" + assert "credential" not in out + + +def test_explicit_guid_on_existing_connection_rides_on_connection(): + # Same routing when the caller supplies the guid itself instead of letting + # _create() resolve it. The trigger is "guid + existing connection", not + # "guid came from a lookup". + out = ( + apps.BigqueryMiner(Mock()) + .connection(qualified_name="default/bigquery/1700000000") + .credential_guid("caller-supplied-guid") + .preview() + ) + assert ( + out["connection"]["attributes"]["defaultCredentialGuid"] + == "caller-supplied-guid" + ) + assert out["credential_guid"] == "" + + +def test_explicit_guid_on_new_connection_stays_top_level(): + # Deliberately unchanged, and NOT safe. This shape is KNOWN to still trigger + # the CONNECT-843 server bug: the credential config record is keyed by the guid + # alone, so minting a new connection protects nothing -- a bare top-level guid + # with no credential body still flattens that guid's shared config record to + # {"credentialSource": "direct"} for every workflow using it. It is left as-is + # because rerouting it here would newly affect crawler-shaped apps, whose + # connection-attribute fallback is unverified (crawler forms carry an explicit + # credential-guid widget, so they may legitimately read the top-level field), + # and because the real fix for this branch is the server-side guard (see the + # heracles handoff on CONNECT-843). This test pins today's behaviour so any + # future reroute is a deliberate, evidenced change and not a drive-by. + out = ( + BigqueryCrawler(Mock()) + .connection(name="prod-bq") + .credential_guid("existing-guid") + .preview() + ) + assert out["credential_guid"] == "existing-guid" + assert "defaultCredentialGuid" not in out["connection"]["attributes"] + + +def test_staged_credential_on_existing_connection_keeps_vaulting_shape(client): + # A staged raw credential is still vaulted by the create endpoint from the + # `credential` key, even on an existing connection: nothing moves onto the + # connection and credential_guid stays "" for the server to fill in. + ( + BigqueryCrawler(client) + .workload_identity_federation(project_id="proj") + .connection(qualified_name="default/bigquery/1700000000") + .include({"proj": ["ds"]}) + .create() + ) + out = client.app.create.call_args.kwargs["inputs"].to_inputs() + assert out["credential"]["authType"] == "gcp-wif" + assert out["credential_guid"] == "" + assert "defaultCredentialGuid" not in out["connection"]["attributes"] + client.asset.search.assert_not_called() # a staged cred needs no lookup + + +def test_agent_mode_on_existing_connection_ignores_credential_guid(): + # The agent/SDR path returns before any credential routing, so neither field + # appears. Unchanged by CONNECT-843. + out = ( + apps.SnowflakeMiner(Mock()) + .agent({"name": "my-agent"}) + .connection(qualified_name="default/snowflake/123") + .credential_guid("should-be-ignored") + .preview() + ) + assert out["agent_json"] == {"name": "my-agent"} + assert "credential_guid" not in out + assert "defaultCredentialGuid" not in out["connection"]["attributes"] def test_agent_mode_uses_agent_json_not_credential(client): From fa8d5484a9151d17b04d87afbba8e796c3f0da10 Mon Sep 17 00:00:00 2001 From: Aryaman Bhushan Date: Wed, 19 Aug 2026 12:37:23 +0530 Subject: [PATCH 6/8] fix(apps): load() reads reused credential guid from the connection (CONNECT-843) After #1007 a reused credential guid rides on connection.attributes.defaultCredentialGuid with top-level credential_guid "". load() now reads the guid from either place so load().update() never drops the credential reference; update() re-routes it onto the connection via the same reuse guard. Tests assert the connection-placement shape. --- pyatlan/model/apps/_base.py | 8 +++++++- tests/unit/test_app_builders.py | 22 ++++++++++++++++++---- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/pyatlan/model/apps/_base.py b/pyatlan/model/apps/_base.py index c366474ef..4db21a040 100644 --- a/pyatlan/model/apps/_base.py +++ b/pyatlan/model/apps/_base.py @@ -330,7 +330,13 @@ def load(self, slug: str) -> "AppBuilder": self._admin_users = list(attrs.get("adminUsers") or []) self._admin_groups = list(attrs.get("adminGroups") or []) self._admin_roles = list(attrs.get("adminRoles") or []) - self._credential_guid = args.get("credential_guid") or "" + # The reused credential guid may ride on the connection entity + # (attributes.defaultCredentialGuid, the UI/CONNECT-843 shape) or, on + # older workflows, sit top-level. Read whichever is present so update() + # never drops the credential reference. + self._credential_guid = ( + args.get("credential_guid") or attrs.get("defaultCredentialGuid") or "" + ) self._extraction_method = ( args.get("extraction_method") or self._EXTRACTION_METHOD ) diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 8060f991d..723e8fc99 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -338,7 +338,13 @@ def test_staged_credential_on_existing_connection_keeps_vaulting_shape(client): # connection and credential_guid stays "" for the server to fill in. ( BigqueryCrawler(client) - .workload_identity_federation(project_id="proj") + .workload_identity_federation( + project_id="proj", + service_account_email="svc@proj.iam.gserviceaccount.com", + wif_pool_provider_id="projects/1/locations/global/workloadIdentityPools/p/providers/pr", + atlan_oauth_id="oauth-id", + atlan_oauth_secret="oauth-secret", + ) .connection(qualified_name="default/bigquery/1700000000") .include({"proj": ["ds"]}) .create() @@ -421,7 +427,10 @@ def test_load_update_preserves_reinjects_and_references_credential(): inp = kw["inputs"] assert kw["slug"] == "slug-1" and kw["entrypoint"] == "crawler" assert inp["connection_qualified_name"] == "default/bigquery/123" # re-injected - assert inp["credential_guid"] == "cred-1" and "credential" not in inp # no rotation + # CONNECT-843: reused credential rides on the connection, never top-level, and + # is never re-vaulted (no rotation). + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-1" + assert inp["credential_guid"] == "" and "credential" not in inp assert inp["include_filter"] == '{"^proj$": ["^new$"]}' # changed + anchored assert inp["control_config"] == "{}" # normalized object -> string assert "user-id" not in inp and "workflow_id" not in inp # runtime keys dropped @@ -463,7 +472,10 @@ def test_load_update_is_generic_across_builders(cls): assert kw["slug"] == "slug-x" assert kw["entrypoint"] == (cls._ENTRYPOINT or None) assert inp["connection_qualified_name"] == qn # re-injected - assert inp["credential_guid"] == "cred-x" # referenced, not rotated + # CONNECT-843: the guid rides on the connection (also proves load() reads it + # back from attributes.defaultCredentialGuid, not just top-level). + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-x" + assert inp["credential_guid"] == "" # referenced, not rotated, not top-level assert "credential" not in inp # no raw credential re-sent @@ -561,6 +573,8 @@ def test_load_update_applies_multiple_field_changes(): assert inp["control_config"] == '{"flag": 1}' # rest preserved / re-injected / stripped assert inp["connection_qualified_name"] == "default/bigquery/1" - assert inp["credential_guid"] == "cred-1" and "credential" not in inp + # CONNECT-843: reused credential on the connection, top-level empty, no rotation + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-1" + assert inp["credential_guid"] == "" and "credential" not in inp assert inp["atlas_auth_type"] == "internal" assert "user-id" not in inp From cf523f75ad418b4d0533e286864c60c74463622c Mon Sep 17 00:00:00 2001 From: mitshah-atlan Date: Mon, 17 Aug 2026 17:40:10 +0530 Subject: [PATCH 7/8] feat(apps): add a typed StandardLineage builder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Standard Lineage (cross-connection lineage) had no typed builder, so callers had to hand-build the raw `inputs` dict for `client.app.create/update`. Two details make that unreasonable to ask of a caller, and neither is expressible from a UI configmap — hence hand-written, and added to the generator's _HAND_WRITTEN set: 1. `cross_connection_qualified_names` is declared `str` in the app's input contract but means a LIST of connection qualified names. Sending a native list fails validateInputsAgainstContract server-side; it has to be json.dumps'd, and Heracles parses it back into a list for the manifest placeholder. `connections()` takes a List[str] and encodes it. 2. The defining operation is re-scoping an EXISTING workflow — adding a connection as it is onboarded — which is an update against a slug. No generated builder does updates; AppBuilder only has create()/run(). The re-scope path reads before it writes, and that is load-bearing rather than a convenience: `client.app.update` is a full replace, and the workflow's own connection entity is republished by the DAG's create-connection node on every run, so sending a rebuilt or partial connection would overwrite the real one in Atlan and strip its name and admins. `set_connections` therefore carries the persisted `connection` and `run_role` over verbatim, read from `client.app.get(slug).dag` (AppSummary tolerates unmodelled fields, so the DAG arrives as an extra). Surface: StandardLineage(client).connection(name=...).connections([...]).run() # create StandardLineage(client).add_connections(slug, [...]) # onboard StandardLineage(client).remove_connections(slug, [...]) # hand back StandardLineage(client).set_connections(slug, [...]) # replace StandardLineage(client).get_connections(slug) # read add_connections/remove_connections are idempotent and return None without publishing a version when nothing would change, so an onboarding portal can replay safely. Validation is client-side where the error is actionable: the app requires a non-empty, same-connector scope, and passing the workflow's OWN standard-lineage connection as its scope — a natural mistake, since both are "connections" — is rejected with a message that says which is wanted. Note the two distinct connectors: the workflow's own connection is minted under `standard-lineage` (_CONNECTOR_NAME), while the `connector` input names the connector of the connections in scope and is derived from them. Verified end-to-end against a live tenant (create path deliberately not exercised there — it would mint a workflow and a connection): the read, add, remove, idempotent no-op and empty-scope refusal all behave as specified, and the re-rendered DAG came back with all 14 connection attributes intact, both downstream nodes identical, and the Temporal workflow type unchanged. 22 new tests; app suite 535 passed, full unit suite 7110 passed. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: mitshah-atlan --- pyatlan/generator/generate_apps.py | 10 +- pyatlan/model/apps/__init__.py | 12 +- pyatlan/model/apps/standard_lineage.py | 287 +++++++++++++++++++++++ tests/unit/apps/test_standard_lineage.py | 223 ++++++++++++++++++ tests/unit/test_app_generated_inputs.py | 3 + 5 files changed, 528 insertions(+), 7 deletions(-) create mode 100644 pyatlan/model/apps/standard_lineage.py create mode 100644 tests/unit/apps/test_standard_lineage.py 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", } From fc37958017852ac36d75bb4494383d8bf286a60c Mon Sep 17 00:00:00 2001 From: mitshah-atlan Date: Tue, 18 Aug 2026 23:10:37 +0530 Subject: [PATCH 8/8] docs: add a module docstring to the Standard Lineage builder tests Applies the code-comments guidance to this branch's own diff only. The builder itself needed no tightening -- it already carries contract-level docstrings with no drift hazards and matches the conventions of its neighbours in pyatlan/model/apps/. The one real gap was the opposite of over-writing: the test file had no file header at all. Matches test_asset_export_flows_handwritten.py, the closest analogue in the repo (hand-written tests for a hand-written v3 builder): records that the file survives the generator's regen because it carries no AUTO-GENERATED banner, and why most cases pin the re-scope path rather than create -- client.app.update is a full replace, so what is absent from the payload matters as much as what is present. 22 tests pass; ruff format and check clean. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: mitshah-atlan --- tests/unit/apps/test_standard_lineage.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/unit/apps/test_standard_lineage.py b/tests/unit/apps/test_standard_lineage.py index 1eff2f112..c35b6d590 100644 --- a/tests/unit/apps/test_standard_lineage.py +++ b/tests/unit/apps/test_standard_lineage.py @@ -1,5 +1,16 @@ # SPDX-License-Identifier: Apache-2.0 # Copyright 2026 Atlan Pte. Ltd. +"""HAND-WRITTEN tests for the Standard Lineage builder (CONNECT-182). + +The generator's regen only clears files carrying the AUTO-GENERATED banner, so this +file survives regeneration alongside the builder it covers. + +Most cases pin the re-scope path rather than create: the builder's job is to change an +existing workflow's set of connections without disturbing the workflow's own connection +entity, and ``client.app.update`` is a full replace — so what is *absent* from the +payload matters as much as what is present. +""" + import json from unittest.mock import Mock