fix(apps): send the full existing connection on a miner run (AICHAT-1798)

pyatlan/errors.py

@@ -706,6 +706,13 @@ class ErrorCode(Enum):
         "See https://docs.atlan.com/product/capabilities/build-apps/sdks/python/apps/manage-apps",
         InvalidRequestError,
     )
+    CONNECTION_READ_FOR_APP_FAILED = (
+        400,
+        "ATLAN-PYTHON-400-081",
+        "Could not read connection '{0}', so the run was not started.",
+        "Verify the qualifiedName and that this credential can read the connection.",
+        InvalidRequestError,
+    )
     AUTHENTICATION_PASSTHROUGH = (
         401,
         "ATLAN-PYTHON-401-000",

pyatlan/model/apps/_base.py

@@ -23,12 +23,45 @@
 
 from pydantic.v1 import BaseModel, Extra
 
+from pyatlan.errors import ErrorCode
 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"})
 
+# The connection attributes an app run puts on its connection object — the stored
+# config the UI forwards on a miner run (identity, credential/policy strategy,
+# admins, query settings). Deliberately NOT the computed analytics/popularity
+# fields (popularityScore, viewScore, sourceRead*, assetMc*, …) that a full
+# entity read carries: echoing those from an extract step is noise at best and
+# can clobber real popularity at worst. Fetched by name so only these come back.
+_CONNECTION_WIRE_ATTRS = (
+    "name",
+    "connectorName",
+    "defaultCredentialGuid",
+    "category",
+    "adminUsers",
+    "adminGroups",
+    "adminRoles",
+    "allowQuery",
+    "allowQueryPreview",
+    "queryTimeout",
+    "rowLimit",
+    "credentialStrategy",
+    "previewCredentialStrategy",
+    "policyStrategy",
+    "policyStrategyForSamplePreview",
+    "objectStorageUploadThreshold",
+    "hasPopularityInsights",
+    "connectionDbtEnvironments",
+    "connectionIsDQEnabled",
+    "isPartial",
+    "isSampleDataPreviewEnabled",
+    "vectorEmbeddingsEnabled",
+    "sourceLogo",
+)
+
 
 class AppInput(BaseModel):
     """A typed, configmap-derived ``inputs`` payload for an app workflow."""
@@ -114,9 +147,11 @@ def __init__(self, client: Any):
         self._admin_roles: List[str] = []
         self._metadata: Dict[str, Any] = {}
         self._update_slug: Optional[str] = None
-        # Full persisted connection captured by load(); re-sent verbatim by
-        # update() so the full-replace drops no connection attributes. Cleared by
-        # an explicit connection() call, which then wins.
+        # A connection object to send verbatim (typeName + attributes): the full
+        # persisted connection captured by load() (update path), or the connection's
+        # stored config read back by _create() when a fresh run references an
+        # existing connection by QN — so its name reaches the payload and the run
+        # cannot rename it (AICHAT-1798). Cleared by an explicit connection() call.
         self._loaded_connection: Optional[Any] = None
 
     # ── Step 1 · Credential ────────────────────────────────────────────────
@@ -459,43 +494,98 @@ def _vault_credential(self, cred: Credential) -> str:
         )
         return CredentialResponse(**raw).id or ""
 
-    def _resolve_connection_credential(self, qualified_name: str) -> Optional[str]:
-        """Look up an existing connection's ``defaultCredentialGuid`` so a caller
-        referencing a connection by QN (e.g. miners) reuses that connection's
-        credential without having to know its guid. Best-effort: returns None if
-        the connection can't be read."""
-        try:
-            from pyatlan.model.assets import Connection
-            from pyatlan.model.fluent_search import FluentSearch
-
-            request = (
-                FluentSearch()
-                .where(Connection.TYPE_NAME.eq("Connection"))
-                .where(Connection.QUALIFIED_NAME.eq(qualified_name))
-                .include_on_results(Connection.DEFAULT_CREDENTIAL_GUID)
-                .page_size(1)
-            ).to_request()
-            for asset in self._client.asset.search(request):
-                return asset.default_credential_guid
-        except Exception:  # noqa: BLE001
-            return None
-        return None
+    def _load_existing_connection(self, qualified_name: str) -> Dict[str, Any]:
+        """Read an existing connection's stored config and return it as the
+        ``{typeName, attributes}`` wire object, so a caller referencing a connection
+        by QN (e.g. miners) sends the connection the way the UI does — not a stub
+        built from the QN alone.
+
+        This is what keeps the connection's name intact: popularity/publish derives
+        the connection name from this payload, so a stub missing ``name`` makes the
+        exporter fall back to the qualifiedName's numeric tail and rename the
+        connection to that number (AICHAT-1798). Only the config attributes the UI
+        forwards are fetched (``_CONNECTION_WIRE_ATTRS``) — never the computed
+        analytics/popularity fields, which must not be echoed from an extract step.
+
+        Raises whatever the read raised (e.g. ``NotFoundError``) rather than
+        returning a stub: a run that cannot read the connection must not rename it.
+        The caller decides whether an explicit name makes a failure recoverable.
+
+        The connection's own ``defaultCredentialGuid`` rides on the read-back, so
+        the credential is reused with no extra lookup. Explicit ``connection(name=,
+        admin_*=)`` / ``credential_guid()`` still win — they are layered back on."""
+        from pyatlan.model.assets import Connection
+
+        connection = self._client.asset.get_by_qualified_name(
+            qualified_name=qualified_name,
+            asset_type=Connection,
+            min_ext_info=True,
+            ignore_relationships=True,
+            attributes=list(_CONNECTION_WIRE_ATTRS),
+        )
+        # Serialize via .json() (not .dict()): pydantic's encoders turn set-typed
+        # attributes into lists and enums/datetimes into their wire form, so the
+        # connection is JSON-safe the same way the stored DAG (load path) is.
+        # .dict() would leave raw sets that fail to serialize.
+        attrs = (
+            json.loads(
+                connection.json(by_alias=True, exclude_none=True, exclude_unset=True)
+            ).get("attributes")
+            or {}
+        )
+        # Keep only the config attributes the UI forwards — defensive in case the
+        # read returns more than was asked for — so no computed analytics/popularity
+        # field can ride along on a full-replace.
+        attrs = {k: v for k, v in attrs.items() if k in _CONNECTION_WIRE_ATTRS}
+        # Identity the run was given always wins over the read-back.
+        attrs["qualifiedName"] = qualified_name
+        parts = qualified_name.split("/")
+        if len(parts) >= 3 and parts[0] == "default":
+            attrs.setdefault("connectorName", parts[1])
+        # Explicit builder values override the stored connection.
+        if self._connection_name is not None:
+            attrs["name"] = self._connection_name
+        if self._admin_users:
+            attrs["adminUsers"] = self._admin_users
+        if self._admin_groups:
+            attrs["adminGroups"] = self._admin_groups
+        if self._admin_roles:
+            attrs["adminRoles"] = self._admin_roles
+        if self._credential_guid:
+            attrs["defaultCredentialGuid"] = self._credential_guid
+        return {"typeName": "Connection", "attributes": attrs}
 
     def _create(self, *, name: Optional[str], run: bool, schedule: Optional[Any]):
         epoch = int(time.time())
         qn = (
             self._connection_qualified_name or f"default/{self._CONNECTOR_NAME}/{epoch}"
         )
-        # Referencing an existing connection without a credential (e.g. miners):
-        # reuse that connection's own credential (its defaultCredentialGuid), looked
-        # up by QN — so the caller only needs to supply the connection.
+        # Referencing an existing connection by QN (e.g. miners): send the whole
+        # connection the UI/rerun way — a full read-back — so a full-replace
+        # downstream drops no attributes and cannot rename it (AICHAT-1798). The
+        # read-back carries the connection's own defaultCredentialGuid, so the
+        # credential is reused too. Falls back to a QN-only stub if the read fails.
+        # Skipped when a raw credential is being vaulted onto the connection (that
+        # path builds its own) or when load()/an explicit connection already set one.
         if (
             self._extraction_method != "agent"
             and not self._raw_creds
-            and self._credential_guid is None
+            and self._loaded_connection is None
             and self._connection_qualified_name
         ):
-            self._credential_guid = self._resolve_connection_credential(qn)
+            try:
+                self._loaded_connection = self._load_existing_connection(qn)
+            except Exception as exc:  # noqa: BLE001
+                # Reading the connection is how its name reaches the run. If it
+                # can't be read and no name was given, a QN-only stub would let
+                # popularity/publish rename the connection to the qualifiedName's
+                # numeric tail (AICHAT-1798) — fail loudly instead of silently
+                # renaming. An explicit connection(name=...) is already safe, so
+                # fall back to the stub there.
+                if self._connection_name is None:
+                    raise ErrorCode.CONNECTION_READ_FOR_APP_FAILED.exception_with_parameters(
+                        qn
+                    ) from exc
         # Named credential fields (e.g. dbt's api_credential_guid) aren't vaulted
         # from the payload — vault them now and place the issued guid in the field.
         resolved_guids: Dict[str, str] = {}

tests/unit/test_app_builders.py

@@ -13,6 +13,7 @@
 import pytest
 
 import pyatlan.model.apps as apps
+from pyatlan.errors import InvalidRequestError
 from pyatlan.model.apps import AppBuilder, BigqueryCrawler, SnowflakeMiner
 
 # Every concrete builder (the hand-written flagship + all generated ones).
@@ -234,28 +235,78 @@ def test_miner_references_existing_connection_by_qn_only():
 
 
 def test_connector_name_derived_from_qn(client):
-    # Even when the builder's connector fallback differs, the QN wins.
+    # Even when the read-back omits connectorName, the QN supplies it.
+    client.asset.get_by_qualified_name.return_value = _connection(
+        "default/snowflake/123", name="conn"
+    )
     SnowflakeMiner(client).connection(qualified_name="default/snowflake/123").create()
     out = client.app.create.call_args.kwargs["inputs"].to_inputs()
     assert out["connection"]["attributes"]["connectorName"] == "snowflake"
 
 
-def test_miner_auto_resolves_connection_credential(client):
-    # Referencing an existing connection by QN (no credential) → the builder looks
-    # up the connection and reuses its defaultCredentialGuid on create().
-    client.asset.search.return_value = iter(
-        [Mock(default_credential_guid="conn-cred-guid")]
+def _connection(qn, **attrs):
+    """A real Connection asset, so _load_existing_connection can serialize it the
+    way it serializes a live read-back."""
+    from pyatlan.model.assets import Connection
+
+    conn = Connection()
+    conn.qualified_name = qn
+    for key, value in attrs.items():
+        setattr(conn, key, value)
+    return conn
+
+
+_NOISE_ATTRS = (
+    "popularityScore",
+    "viewScore",
+    "sourceReadTopUserList",
+    "assetMcIncidentNames",
+    "starredBy",
+)
+
+
+def test_miner_sends_the_existing_connection_config(client):
+    # Referencing an existing connection by QN (no credential): the builder reads
+    # the connection's stored config back and sends it — name, credential, config —
+    # the way the UI does, so the run cannot rename it to the qualifiedName tail.
+    # The read-back here also carries computed analytics (popularityScore/...), to
+    # prove those are dropped, not forwarded.
+    read_back = _connection(
+        "default/snowflake/123",
+        name="sales-snowflake",
+        default_credential_guid="conn-cred-guid",
+        category="warehouse",
+        row_limit=10000,
+        admin_users=["u1"],
+        popularity_score=1.5,
+        view_score=2.0,
     )
+    client.asset.get_by_qualified_name.return_value = read_back
     SnowflakeMiner(client).connection(qualified_name="default/snowflake/123").create()
-    assert client.asset.search.called  # connection was looked up
+
+    # the read-back asks for ONLY the UI's config attributes — never the computed
+    # analytics/popularity fields — so they cannot ride along on a full-replace.
+    from pyatlan.model.apps._base import _CONNECTION_WIRE_ATTRS
+
+    requested = client.asset.get_by_qualified_name.call_args.kwargs["attributes"]
+    assert requested == list(_CONNECTION_WIRE_ATTRS)
+    assert all(noise not in requested for noise in _NOISE_ATTRS)
+
     out = client.app.create.call_args.kwargs["inputs"].to_inputs()
-    # 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
+    # AICHAT-1798: the connection's name + config ride on the payload, so
+    # popularity/publish cannot rename it to the qualifiedName's numeric tail.
+    assert attrs["name"] == "sales-snowflake"
+    assert attrs["category"] == "warehouse"
+    assert attrs["rowLimit"] == 10000
+    assert attrs["adminUsers"] == ["u1"]
+    assert attrs["connectorName"] == "snowflake"
+    # CONNECT-843: the connection's own credential rides on the entity (the UI's
+    # wire shape), never as a bare top-level credential_guid.
+    assert attrs["defaultCredentialGuid"] == "conn-cred-guid"
     assert out["credential_guid"] == ""  # and not duplicated at the top level
+    # and NONE of the computed analytics/popularity fields leak through.
+    assert all(noise not in attrs for noise in _NOISE_ATTRS)
 
 
 # --------------------------------------------------------------------------- #
@@ -267,9 +318,12 @@ def test_miner_auto_resolves_connection_credential(client):
 # 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)])
+def _resolve_to(client, guid, name="looked-up-conn"):
+    """Make the connection read-back in _create() return a full connection carrying
+    ``guid`` and ``name``."""
+    client.asset.get_by_qualified_name.return_value = _connection(
+        "default/x/123", name=name, default_credential_guid=guid
+    )
 
 
 @pytest.mark.parametrize(
@@ -284,15 +338,51 @@ def test_auto_resolved_guid_rides_on_connection_not_top_level(client, cls, conne
     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
+    # exactly the UI's reuse shape: identity + the connection's own credential + name
     assert attrs["defaultCredentialGuid"] == "resolved-guid"
     assert attrs["qualifiedName"] == f"default/{connector}/123"
     assert attrs["connectorName"] == connector
+    assert attrs["name"] == "looked-up-conn"  # AICHAT-1798: name preserved
     # 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_connection_name_wins_over_looked_up_name(client):
+    # A caller-supplied name is never clobbered by the connection's stored name.
+    _resolve_to(client, "g", name="stored-in-atlas")
+    SnowflakeMiner(client).connection(
+        qualified_name="default/snowflake/123", name="chosen-by-caller"
+    ).create()
+    out = client.app.create.call_args.kwargs["inputs"].to_inputs()
+    assert out["connection"]["attributes"]["name"] == "chosen-by-caller"
+
+
+def test_read_back_failure_raises_rather_than_risking_a_rename(client):
+    # If the connection can't be read and no name was given, sending a QN-only stub
+    # would rename it to the qualifiedName tail — so fail loudly instead of running.
+    client.asset.get_by_qualified_name.side_effect = Exception("not found")
+    with pytest.raises(InvalidRequestError, match="ATLAN-PYTHON-400-081"):
+        SnowflakeMiner(client).connection(
+            qualified_name="default/snowflake/123"
+        ).create()
+    client.app.create.assert_not_called()  # nothing was submitted
+
+
+def test_read_back_failure_with_explicit_name_falls_back(client):
+    # An explicit name is already safe from the rename, so a read failure is
+    # recoverable: the run still goes out, carrying the caller's name.
+    client.asset.get_by_qualified_name.side_effect = Exception("not found")
+    SnowflakeMiner(client).connection(
+        qualified_name="default/snowflake/123", name="chosen"
+    ).create()
+    attrs = client.app.create.call_args.kwargs["inputs"].to_inputs()["connection"][
+        "attributes"
+    ]
+    assert attrs["name"] == "chosen"
+    assert attrs["qualifiedName"] == "default/snowflake/123"
+
+
 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
@@ -353,7 +443,7 @@ def test_staged_credential_on_existing_connection_keeps_vaulting_shape(client):
     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
+    client.asset.get_by_qualified_name.assert_not_called()  # staged cred: no read-back
 
 
 def test_agent_mode_on_existing_connection_ignores_credential_guid():