feat(apps)!: typed BigQuery WIF credentials + load()/update() for app-workflow updates

[Aryamanz29]

Two related improvements to the BigQuery crawler app builder (and, for the second, all app builders). Validated live on an internal tenant with real WIF credentials — a real crawl runs to Succeeded on both create and update+rerun.

1. Type the WIF credential (breaking)

workload_identity_federation() typed only project_id; the four WIF-specific values rode through untyped **extra, so they were undiscoverable and a misspelled key was silently accepted (a broken gcp-wif credential, no error). They're now typed, required keyword-only params mapping to the exact extra keys, matching the fully-typed sibling service_account():

BigqueryCrawler(client).workload_identity_federation(
    project_id=..., service_account_email=..., wif_pool_provider_id=...,
    atlan_oauth_id=..., atlan_oauth_secret=...,
)

BREAKING (a bare project_id-only call now raises) → warrants the next major. **extra is retained for forward-compat with fields newer than this signature.

2. load() / update() — a fluent update path for app workflows

client.app.update() is a full replace with no read-back, so callers had to hand-build the entire input set. Worse, the server's CREATE path derives connection_qualified_name but UPDATE does not — so a hand-built update returns 200 while the next run fails downstream (e.g. Publish-to-Atlas). AppBuilder now offers the same shape as create, for updates:

BigqueryCrawler(client).load(slug).include({...}).update()

load() seeds the builder from the workflow's current inputs; update() full-replaces while:

• referencing the existing credential by guid on the connection entity (builds on fix(apps): send a reused credential guid on the connection, not top-level [CONNECT-843] #1007 / CONNECT-843) — never re-vaulting or rotating it,
• re-injecting connection_qualified_name (the read-back drops it; the run needs it),
• normalizing string-typed contract fields returned as objects (e.g. control_config),
• preserving every other current input and stripping runtime-only keys.

It's on the base AppBuilder, so every app builder inherits it (Snowflake, Postgres, …), not just BigQuery.

Note for the backend: ideally the server's UPDATE path should derive connection_qualified_name like CREATE does, so the SDK wouldn't need to re-inject it. Filing separately.

Builds on #1007 (CONNECT-843): reused-credential placement

load()/update() and the miner builders reuse an already-vaulted credential. #1007 fixes where that reused guid travels; this PR is stacked on it and adds the read-back half in load().

A reused guid must ride on the connection entity (attributes.defaultCredentialGuid, the UI's wire shape) with top-level credential_guid left "". A bare top-level guid with no credential body makes the CREATE/UPDATE endpoint rewrite that credential's shared config record from the (absent) body — flattening a gcp-wif credential to {"credentialSource":"direct"} and wiping its authType/extra.*. Every workflow sharing that guid then breaks: the BigQuery miner preflight resolves the credential as a service account and raises MissingCredentialsError on a WIF connection whose crawler succeeds (reproduced end-to-end — fresh crawler runs to Succeeded, then the miner fails preflight on the same credential; fixed once the guid rides on the connection).

Before (top-level — corrupts the shared credential):

"connection": { "attributes": { "qualifiedName": "default/bigquery/1700000000", "connectorName": "bigquery" } },
"credential_guid": "<existing-cred-guid>"

After (rides on the connection — stored credential untouched):

"connection": { "attributes": { "qualifiedName": "default/bigquery/1700000000", "connectorName": "bigquery", "defaultCredentialGuid": "<existing-cred-guid>" } },
"credential_guid": ""

The read-back half added here: load() reads the reused guid from attributes.defaultCredentialGuid (falling back to top-level for older workflows), so load().update() never drops the credential reference. The Update payload below reflects this fixed shape.

SDK snippets and generated payloads

Each operation — the SDK call and the exact request it produces (endpoint + body). Offline; placeholder credentials, slugs, and guids.

Create (create without running)

SDK

BigqueryCrawler(client)
    .workload_identity_federation(
        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="…",
    )
    .connection(name="prod-bigquery", admin_users=["jdoe"])
    .include({"my-project": ["analytics", "sales"]})
    .create()

Request — POST /api/service/v1/app

{
  "app_id": "bigquery-crawler",
  "entrypoint": "crawler",
  "name": "prod-bigquery",
  "run": false,
  "inputs": {
    "connection": {
      "typeName": "Connection",
      "attributes": {
        "qualifiedName": "default/bigquery/0",
        "connectorName": "bigquery",
        "name": "prod-bigquery",
        "adminRoles": [
          "<admin-role-guid>"
        ]
      }
    },
    "extraction_method": "direct",
    "credential_guid": "",
    "include_filter": "{\"^my-project$\": [\"^analytics$\", \"^sales$\"]}",
    "exclude_filter": "{}",
    "temp_table_regex": "",
    "enable_nested_columns": true,
    "enable_bigquery_tag_sync": false,
    "filter_sharded_tables": true,
    "hidden_datasets": false,
    "control_config_strategy": "default",
    "control_config": "{}",
    "preflight_check": "",
    "max_concurrent_activities": 15,
    "max_activities_per_execution": 300,
    "extract_output_chuck_size": 50000,
    "credential": {
      "authType": "gcp-wif",
      "name": "default-bigquery-0-0",
      "connectorConfigName": "atlan-connectors-bigquery",
      "connectorType": "bigquery",
      "host": "https://bigquery.googleapis.com",
      "port": 443,
      "extra": {
        "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>"
      }
    },
    "mapping_chunk_size": 1000,
    "atlas_auth_type": "internal",
    "list_datasets_per_chunk": 50
  }
}

Run (create + submit a run)

SDK

BigqueryCrawler(client)
    .workload_identity_federation(
        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="…",
    )
    .connection(name="prod-bigquery", admin_users=["jdoe"])
    .include({"my-project": ["analytics", "sales"]})
    .run()

Request — POST /api/service/v1/app

{
  "app_id": "bigquery-crawler",
  "entrypoint": "crawler",
  "name": "prod-bigquery",
  "run": true,
  "inputs": {
    "connection": {
      "typeName": "Connection",
      "attributes": {
        "qualifiedName": "default/bigquery/0",
        "connectorName": "bigquery",
        "name": "prod-bigquery",
        "adminRoles": [
          "<admin-role-guid>"
        ]
      }
    },
    "extraction_method": "direct",
    "credential_guid": "",
    "include_filter": "{\"^my-project$\": [\"^analytics$\", \"^sales$\"]}",
    "exclude_filter": "{}",
    "temp_table_regex": "",
    "enable_nested_columns": true,
    "enable_bigquery_tag_sync": false,
    "filter_sharded_tables": true,
    "hidden_datasets": false,
    "control_config_strategy": "default",
    "control_config": "{}",
    "preflight_check": "",
    "max_concurrent_activities": 15,
    "max_activities_per_execution": 300,
    "extract_output_chuck_size": 50000,
    "credential": {
      "authType": "gcp-wif",
      "name": "default-bigquery-0-0",
      "connectorConfigName": "atlan-connectors-bigquery",
      "connectorType": "bigquery",
      "host": "https://bigquery.googleapis.com",
      "port": 443,
      "extra": {
        "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>"
      }
    },
    "mapping_chunk_size": 1000,
    "atlas_auth_type": "internal",
    "list_datasets_per_chunk": 50
  }
}

Update (change multiple fields; credential referenced — no rotation)

SDK

(
    BigqueryCrawler(client).load(slug)
    .include({"my-project": ["analytics", "sales"]})   # change several fields
    .exclude({"my-project": ["tmp"]})
    .exclude_regex(".*_bak$")
    .import_nested_columns(False)
    .update()                                          # credential referenced (no rotation)
)

Request — PUT /api/service/v1/app/bigquery-crawler-abc123

{
  "entrypoint": "crawler",
  "inputs": {
    "connection": {
      "typeName": "Connection",
      "attributes": {
        "qualifiedName": "default/bigquery/1700000000",
        "connectorName": "bigquery",
        "defaultCredentialGuid": "<existing-cred-guid>",
        "name": "prod-bigquery",
        "adminRoles": [
          "<admin-role-guid>"
        ]
      }
    },
    "extraction_method": "direct",
    "credential_guid": "",
    "include_filter": "{\"^my-project$\": [\"^analytics$\", \"^sales$\"]}",
    "exclude_filter": "{\"^my-project$\": [\"^tmp$\"]}",
    "temp_table_regex": ".*_bak$",
    "enable_nested_columns": false,
    "enable_bigquery_tag_sync": false,
    "filter_sharded_tables": true,
    "hidden_datasets": false,
    "control_config_strategy": "default",
    "control_config": "{}",
    "preflight_check": "",
    "max_concurrent_activities": 15,
    "max_activities_per_execution": 300,
    "extract_output_chuck_size": 50000,
    "app_name": "bigquery-crawler",
    "atlas_auth_type": "internal",
    "list_datasets_per_chunk": 50,
    "mapping_chunk_size": 1000,
    "connection_qualified_name": "default/bigquery/1700000000"
  }
}

Rerun the current published version

SDK

client.app.submit(slug)

Request — POST /api/service/v1/app/{slug}/submit
(no request body)

Run status (poll until terminal)

SDK

client.app.get_run(run_id)

Request — GET /api/service/v1/app/runs/{run_id}
(no request body)

Cancel an in-flight run

SDK

client.app.cancel_run(run_id)

Request — POST /api/service/v1/app/runs/{run_id}/cancel
(no request body)

Add a schedule

SDK

client.app.add_schedule(slug, "0 9 * * *", "UTC")

Request — POST /api/service/v1/app/{slug}/schedule

{
  "cron": "0 9 * * *",
  "timezone": "UTC"
}

Remove a schedule

SDK

client.app.remove_schedule(slug, trigger_id)

Request — DELETE /api/service/v1/app/{slug}/schedule/{trigger_id}
(no request body)

Get a workflow

SDK

client.app.get(slug)

Request — GET /api/service/v1/app/{slug}
(no request body)

List / resolve slug by name

SDK

client.app.get_all(name="prod-bigquery")

Request — GET /api/service/v1/app?limit={n}&cursor={c}&name={name}
(no request body)

Delete a workflow

SDK

client.app.delete(slug)

Request — DELETE /api/service/v1/app/{slug}
(no request body)

Describe an app

SDK

client.app.describe("bigquery-crawler")

Request — GET /api/service/v1/apps/bigquery-crawler
(no request body)

Input contract

SDK

client.app.get_input_contract("bigquery-crawler", "crawler")

Request — GET /api/service/v1/apps/bigquery-crawler/inputs?entrypoint=crawler
(no request body)

Tests

• tests/unit/apps/test_bigquery_crawler.py — WIF typing: keys land in the credential, missing/typo field → TypeError.
• test_app_builders.py:
  • load().include().update() re-injects connection_qualified_name, references the credential on the connection (defaultCredentialGuid, top-level credential_guid ""; no rotation), normalizes control_config, drops runtime keys, preserves the rest; update() before load() raises.
  • generic across every builder — parametrized load()/update() (re-inject + credential reference on the connection, which also proves load() reads the reused guid back from defaultCredentialGuid + slug/entrypoint).
  • multiple field changes in one update (include/exclude/regex/toggles/custom_config) all apply while the rest is preserved.
  • update() 1003 fallback — retries at the default entrypoint when the named one has no registered contract.
• Full app suite green; ruff clean. No version.txt/HISTORY.md bump — versions are cut via the [release] flow.