feat(app): AppClient for native (v3) app workflow APIs (BLDX-1472)

.gitignore

@@ -60,3 +60,6 @@ pyatlan/documentation/*.md
 docs/_build/
 docs/generated/
 site/
+test.py
+test_apps/
+native-v3-workflow-api-guide.md

HISTORY.md

@@ -1,3 +1,21 @@
+## 9.8.0 (June 24, 2026)
+
+### New Features
+
+- **App workflow client (`client.app`)**: A new `AppClient` (sync + async) for creating, running, scheduling, and managing app workflows over the `/v1/app*` APIs. Workflows are created from an `app_id` plus a generic `inputs` dict validated server-side against the app's live input contract (`client.app.get_input_contract(...)`), so a connector never needs a hand-maintained package class. Includes `describe`, `get_input_contract`, `create`, `get`, `get_all`, `update`, `submit`, `get_run`, `cancel_run`, `add_schedule`, `remove_schedule`, and `delete`. (#959)
+- **Typed app builders (`pyatlan.model.apps`)**: A fluent, UI-equivalent builder per connector app (e.g. `BigqueryCrawler`, `SnowflakeCrawler`, `AtlanTableau`, `PostgresCrawler`, …) mirroring the "new app" wizard (Credential → Connection → Metadata). A builder vaults the credential, mints the connection qualified name, and assembles the full payload — so callers never hand-build a connection object or guess an input key, e.g. `BigqueryCrawler(client).service_account(...).connection(name=...).include({...}).run()`. Each builder and its typed `*Inputs` model are generated from the app's UI configmaps via `uv run generate-apps`; 33 connectors are covered. (#959)
+- **Resolve a workflow by name + idempotent create**: `client.app.get_all(name=...)` now filters by exact workflow name (server `?name=`), so a stateless script can resolve a slug from a name. `client.app.create(...)` is now idempotent on a duplicate name — on the server's `409` it resolves and returns the existing workflow (*create-or-reuse-by-name*) instead of raising, so re-running a migration script is safe. (#959)
+
+### Bug Fixes
+
+- **App builders — correct per-connector filter shapes**: BI/object filters for Tableau, Power BI, QuickSight, and Sigma are now serialized to the JSON-string form their input contract expects (`{id: {}}` → a JSON string), and Power BI's `odbc_dsn_config_mapping` is sent as an object — fixing `1000 inputs do not match the app input contract` errors. (#959)
+- **Miner `extraction_method`**: the Oracle, Postgres, and Teradata miners now send `extraction_method="query_history"` (per their configmaps) instead of the default `"direct"`; the Oracle miner's `start_date` is documented as an epoch timestamp. (#959)
+- **Entrypoint fallback on create**: an app that declares a named entrypoint (e.g. `crawler`) but registers its input contract only at the default slot no longer fails with `1003`; `create` retries without the entrypoint so the server resolves the default contract. (#959)
+
+### Deprecations
+
+- **`WorkflowClient` (`client.workflow`) and `model.packages.*`** are deprecated and now emit a `DeprecationWarning`. They target the legacy workflow surface, which no longer runs on newer Atlan tenants — use `client.app` instead. (#959)
+
 ## 9.7.6 (June 16, 2026)
 
 ### Bug Fixes

README.md

@@ -224,6 +224,32 @@ This will:
 - 🎨 Format code automatically
 - ⚡ Support incremental updates
 
+### App Builders Generator
+
+Generate the app builders in `pyatlan.model.apps` (one per connector
+app) from your tenant's UI configmaps. Each app gets a typed `*Inputs` model and a
+fluent, UI-equivalent builder (Credential → Connection → Metadata), plus a matching
+unit-test module under `tests/unit/apps/`.
+
+```bash
+# Needs a tenant with the apps installed
+export ATLAN_BASE_URL=https://<your-tenant-host>
+export ATLAN_API_KEY=<your-api-key>
+
+# Regenerate all app builders + their tests
+uv run generate-apps
+```
+
+This will:
+- 🔎 Discover installed apps (live discovery ∪ the manifest in `generate_apps.py`)
+- 🧩 Read each app's UI configmaps for fields, labels, enums, and the credential form
+- 🏗️ Emit a `{App}Inputs` model + `{App}` builder per app (hand-written builders are never overwritten)
+- 🧪 Generate a per-app config-assertion test module
+- 🎨 Format code automatically
+
+> To cover an app that isn't currently running on the tenant, add its `(app_id, entrypoint)`
+> to `MANIFEST` in [`pyatlan/generator/generate_apps.py`](pyatlan/generator/generate_apps.py).
+
 ## 🏗️ pyatlan_v9 Model Generation (msgspec)
 
 The `pyatlan_v9` package uses [msgspec](https://jcristharris.com/msgspec/) `Struct`-based models generated from Pkl type definitions in the [atlanhq/models](https://github.com/atlanhq/models) repo.

docs/api/client.md

@@ -38,8 +38,20 @@
 
 ## Workflow Client
 
+> **Deprecated.** Targets the legacy workflow surface, which no longer runs on
+> newer Atlan tenants. Use the **App Client** below for app workflows.
+
 ::: pyatlan.client.workflow
 
+## App Client
+
+The app-workflow client (`client.app`). Creates and manages app workflows from an
+`app_id` plus a generic `inputs` dict validated server-side against the app's live
+input contract — replacing the hand-maintained `model.packages` builders. Discover
+inputs at runtime via `client.app.get_input_contract(...)`.
+
+::: pyatlan.client.app
+
 ## Credential Client
 
 ::: pyatlan.client.credential

docs/api/model.md

@@ -32,8 +32,18 @@
 
 ::: pyatlan.model.enums
 
+## App
+
+Models for the app-workflow APIs (`client.app`).
+
+::: pyatlan.model.app
+
 ## Packages
 
+> **Deprecated.** Package builders produce legacy workflows, which no longer run
+> on newer Atlan tenants. Use `client.app` with the app's input contract instead
+> (see the **App** models above).
+
 ::: pyatlan.model.packages
 
 ## Group

pyatlan/client/aio/app.py

@@ -0,0 +1,220 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 Atlan Pte. Ltd.
+"""Async client for the App workflow APIs.
+
+Async mirror of :class:`pyatlan.client.app.AppClient` — same request building and
+response parsing (shared via :mod:`pyatlan.client.common.app`); only ``_call_api``
+is awaited. Obtain via :attr:`pyatlan.client.aio.client.AsyncAtlanClient.app`.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, Optional, Union
+
+from pydantic.v1 import validate_arguments
+
+from pyatlan.client.common import AsyncApiCaller
+from pyatlan.client.common.app import (
+    AppAddSchedule,
+    AppCancelRun,
+    AppCreate,
+    AppDelete,
+    AppGet,
+    AppGetInfo,
+    AppGetInputContract,
+    AppGetRun,
+    AppListAll,
+    AppRemoveSchedule,
+    AppSubmit,
+    AppUpdate,
+)
+from pyatlan.errors import ConflictError, ErrorCode
+from pyatlan.model.apps import AppInput
+from pyatlan.model.app import (
+    AppDeleteResponse,
+    AppInfo,
+    AppInputContract,
+    AppList,
+    AppResponse,
+    AppRunCancelResponse,
+    AppRunResponse,
+    AppSchedule,
+    AppScheduleDeleteResponse,
+    AppScheduleResponse,
+    AppSummary,
+    CreateApp,
+    UpdateApp,
+)
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class AsyncAppClient:
+    """Async version of :class:`pyatlan.client.app.AppClient`."""
+
+    def __init__(self, client: AsyncApiCaller):
+        if not isinstance(client, AsyncApiCaller):
+            raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(
+                "client", "AsyncApiCaller"
+            )
+        self._client = client
+
+    # ----------------------------- discovery ----------------------------- #
+    @validate_arguments
+    async def describe(self, app_id: str) -> AppInfo:
+        """Describe an app: native-readiness + entrypoints (contrast :meth:`get`)."""
+        raw = await self._client._call_api(AppGetInfo.prepare_request(app_id))
+        return AppGetInfo.process_response(raw)
+
+    @validate_arguments
+    async def get_input_contract(
+        self, app_id: str, entrypoint: Optional[str] = None
+    ) -> AppInputContract:
+        """Fetch the app's input contract (JSON Schema) for an entrypoint."""
+        endpoint, query_params = AppGetInputContract.prepare_request(app_id, entrypoint)
+        raw = await self._client._call_api(endpoint, query_params=query_params)
+        return AppGetInputContract.process_response(raw)
+
+    # ----------------------------- lifecycle ----------------------------- #
+    @validate_arguments(config=dict(arbitrary_types_allowed=True))
+    async def create(
+        self,
+        app_id: str,
+        name: str,
+        inputs: Union[Dict[str, Any], AppInput],
+        entrypoint: Optional[str] = None,
+        schedule: Optional[AppSchedule] = None,
+        run: Optional[bool] = None,
+    ) -> AppResponse:
+        """Create a workflow (create + version + publish + optional schedule/run).
+
+        On a duplicate ``name`` the server responds ``409``; this resolves the existing
+        workflow by name and returns it (idempotent *create-or-reuse-by-name*). A
+        non-unique name re-raises the conflict — disambiguate with :meth:`get_all`.
+        """
+        if isinstance(inputs, AppInput):
+            inputs = inputs.to_inputs()
+        # Only include optional fields when provided so exclude_unset omits them
+        # (passing None explicitly would serialize as null and reach the server).
+        request_kwargs: Dict[str, Any] = {
+            "app_id": app_id,
+            "name": name,
+            "inputs": inputs,
+        }
+        if entrypoint is not None:
+            request_kwargs["entrypoint"] = entrypoint
+        if schedule is not None:
+            request_kwargs["schedule"] = schedule
+        if run is not None:
+            request_kwargs["run"] = run
+        request = CreateApp(**request_kwargs)
+        endpoint, request_obj = AppCreate.prepare_request(request)
+        try:
+            raw = await self._client._call_api(endpoint, request_obj=request_obj)
+        except ConflictError as conflict:
+            return await self._reuse_on_conflict(name, conflict)
+        return AppCreate.process_response(raw)
+
+    async def _reuse_on_conflict(
+        self, name: str, conflict: ConflictError
+    ) -> AppResponse:
+        """Resolve a duplicate-name ``409`` to the existing workflow's slug (reuse).
+
+        A non-unique name can't be reused safely, so the conflict is re-raised.
+        """
+        existing = [w for w in (await self.get_all(name=name)).workflows if w.slug]
+        if len(existing) == 1:
+            slug = existing[0].slug
+            assert slug is not None  # guaranteed by the filter above
+            LOGGER.info("App workflow %r already exists; reusing slug %s", name, slug)
+            return AppResponse(slug=slug, version=existing[0].version)
+        raise conflict
+
+    @validate_arguments
+    async def get_all(
+        self,
+        limit: Optional[int] = None,
+        cursor: Optional[str] = None,
+        name: Optional[str] = None,
+    ) -> AppList:
+        """List published native app workflows (paginate via ``next_cursor``).
+
+        :param name: filter to workflows with this exact ``name`` (server ``?name=``);
+            use it to resolve a workflow's slug from its name.
+        """
+        endpoint, query_params = AppListAll.prepare_request(limit, cursor, name)
+        raw = await self._client._call_api(endpoint, query_params=query_params)
+        return AppListAll.process_response(raw)
+
+    @validate_arguments
+    async def get(self, slug: str) -> AppSummary:
+        """Get a single workflow by slug."""
+        raw = await self._client._call_api(AppGet.prepare_request(slug))
+        return AppGet.process_response(raw)
+
+    @validate_arguments(config=dict(arbitrary_types_allowed=True))
+    async def update(
+        self,
+        slug: str,
+        inputs: Union[Dict[str, Any], AppInput],
+        entrypoint: Optional[str] = None,
+    ) -> AppResponse:
+        """Replace a workflow's inputs and publish a new version on the same slug."""
+        if isinstance(inputs, AppInput):
+            inputs = inputs.to_inputs()
+        request_kwargs: Dict[str, Any] = {"inputs": inputs}
+        if entrypoint is not None:
+            request_kwargs["entrypoint"] = entrypoint
+        request = UpdateApp(**request_kwargs)
+        endpoint, request_obj = AppUpdate.prepare_request(slug, request)
+        raw = await self._client._call_api(endpoint, request_obj=request_obj)
+        return AppUpdate.process_response(raw)
+
+    @validate_arguments
+    async def delete(self, slug: str) -> AppDeleteResponse:
+        """Archive/delete a workflow."""
+        raw = await self._client._call_api(AppDelete.prepare_request(slug))
+        return AppDelete.process_response(raw)
+
+    # ------------------------------ running ------------------------------ #
+    @validate_arguments
+    async def submit(self, slug: str) -> AppRunResponse:
+        """Run the workflow's current published version."""
+        raw = await self._client._call_api(AppSubmit.prepare_request(slug))
+        return AppSubmit.process_response(raw)
+
+    @validate_arguments
+    async def get_run(self, run_id: str) -> AppRunResponse:
+        """Get a run's status. Poll until :attr:`AppRunResponse.is_terminal`."""
+        raw = await self._client._call_api(AppGetRun.prepare_request(run_id))
+        return AppGetRun.process_response(raw)
+
+    @validate_arguments
+    async def cancel_run(self, run_id: str) -> AppRunCancelResponse:
+        """Cancel an in-flight run."""
+        raw = await self._client._call_api(AppCancelRun.prepare_request(run_id))
+        return AppCancelRun.process_response(raw)
+
+    # ---------------------------- scheduling ----------------------------- #
+    @validate_arguments
+    async def add_schedule(
+        self, slug: str, cron: str, timezone: Optional[str] = None
+    ) -> AppScheduleResponse:
+        """Attach a cron schedule to the latest published version."""
+        # The server rejects a null timezone, so apply the documented UTC default.
+        schedule = AppSchedule(cron=cron, timezone=timezone or "UTC")
+        endpoint, request_obj = AppAddSchedule.prepare_request(slug, schedule)
+        raw = await self._client._call_api(endpoint, request_obj=request_obj)
+        return AppAddSchedule.process_response(raw)
+
+    @validate_arguments
+    async def remove_schedule(
+        self, slug: str, trigger_id: str
+    ) -> AppScheduleDeleteResponse:
+        """Remove one schedule (by its ``trigger_id``) from a workflow."""
+        raw = await self._client._call_api(
+            AppRemoveSchedule.prepare_request(slug, trigger_id)
+        )
+        return AppRemoveSchedule.process_response(raw)

pyatlan/client/aio/client.py

@@ -51,6 +51,7 @@
 from pyatlan.client.aio.task import AsyncTaskClient
 from pyatlan.client.aio.token import AsyncTokenClient
 from pyatlan.client.aio.typedef import AsyncTypeDefClient
+from pyatlan.client.aio.app import AsyncAppClient
 from pyatlan.client.aio.user import AsyncUserClient
 from pyatlan.client.aio.workflow import AsyncWorkflowClient
 from pyatlan.client.atlan import (
@@ -118,6 +119,7 @@ class AsyncAtlanClient(AtlanClient):
     _async_typedef_client: Optional[AsyncTypeDefClient] = PrivateAttr(default=None)
     _async_user_client: Optional[AsyncUserClient] = PrivateAttr(default=None)
     _async_workflow_client: Optional[AsyncWorkflowClient] = PrivateAttr(default=None)
+    _async_app_client: Optional[AsyncAppClient] = PrivateAttr(default=None)
 
     # Async cache instances
     _async_atlan_tag_cache: Optional[AsyncAtlanTagCache] = PrivateAttr(default=None)
@@ -394,6 +396,13 @@ def workflow(self) -> AsyncWorkflowClient:  # type: ignore[override]
             self._async_workflow_client = AsyncWorkflowClient(self)  # type: ignore[arg-type]
         return self._async_workflow_client
 
+    @property
+    def app(self) -> AsyncAppClient:  # type: ignore[override]
+        """Get async app workflow client with same API as sync"""
+        if self._async_app_client is None:
+            self._async_app_client = AsyncAppClient(self)  # type: ignore[arg-type]
+        return self._async_app_client
+
     @property
     def atlan_tag_cache(self) -> AsyncAtlanTagCache:  # type: ignore[override]
         """Get async Atlan tag cache with same API as sync"""
@@ -964,6 +973,7 @@ async def aclose(self):
         self._async_typedef_client = None
         self._async_user_client = None
         self._async_workflow_client = None
+        self._async_app_client = None
 
     @contextlib.asynccontextmanager  # type: ignore[arg-type]
     async def max_retries(  # type: ignore[override,misc]