feat: client.requests + client.inbox — approve/reject both request systems programmatically (BLDX-1611)

pyatlan/client/aio/approval_workflow.py

@@ -0,0 +1,111 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 Atlan Pte. Ltd.
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic.v1 import validate_arguments
+
+from pyatlan.client.common import (
+    ApprovalWorkflowBulkActionRequests,
+    ApprovalWorkflowGetRequest,
+    AsyncApiCaller,
+)
+from pyatlan.errors import ErrorCode, InvalidRequestError
+from pyatlan.model.enums import ApprovalWorkflowRequestType
+from pyatlan.model.approval_workflow import (
+    ApprovalWorkflowBulkActionResponse,
+    ApprovalWorkflowRequest,
+)
+
+
+def _raise_if_recipient_scoped(err: InvalidRequestError, group_key: str):
+    """Translate the server's misleading 1003 into an actionable message.
+
+    Bulk actions are RECIPIENT-scoped: the server reports "No pending tasks
+    found for the specified group" even when the group visibly has pending
+    tasks — whenever none of them are addressed to the calling identity.
+    """
+    if "No pending tasks found" not in str(err):
+        return
+    raise ErrorCode.INVALID_REQUEST_PASSTHROUGH.exception_with_parameters(
+        "1003",
+        (
+            f"no actionable pending tasks in group '{group_key}' for the "
+            "calling identity. Two common causes: (1) every task in the "
+            "group is already actioned (approved/rejected/withdrawn) — "
+            "check task_execution_action via a Task search; (2) the pending "
+            "tasks are addressed to a different user — bulk approvals are "
+            "recipient-scoped, and an admin role does not override this. "
+            "To automate approvals, the token's identity must be the "
+            "workflow's approver (the workflow builder currently supports "
+            "only human users and groups as approvers, so automation may "
+            "require a user token)."
+        ),
+        "",
+    ) from err
+
+
+class AsyncApprovalWorkflowClient:
+    """
+    Async client for the governance-workflow approval system (the newer
+    Inbox). For the classic Requests module use `client.requests` instead —
+    tenants can have both.
+    """
+
+    def __init__(self, client: AsyncApiCaller):
+        if not isinstance(client, AsyncApiCaller):
+            raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(
+                "client", "AsyncApiCaller"
+            )
+        self._client = client
+
+    @validate_arguments
+    async def get(self, guid: str) -> Optional[ApprovalWorkflowRequest]:
+        """
+        Retrieve one approval-workflow request by its GUID.
+
+        :param guid: unique identifier of the workflow request
+        :raises AtlanError: on any error during API invocation.
+        :returns: the workflow request, or None if it does not exist
+        """
+        endpoint = ApprovalWorkflowGetRequest.prepare_request(guid)
+        raw_json = await self._client._call_api(endpoint)
+        return ApprovalWorkflowGetRequest.process_response(raw_json)
+
+    @validate_arguments
+    async def approve_all(
+        self,
+        group_key: str,
+        sub_type: Optional[ApprovalWorkflowRequestType] = None,
+        comment: Optional[str] = None,
+    ) -> ApprovalWorkflowBulkActionResponse:
+        """Bulk-approve all pending workflow tasks in a group."""
+        endpoint, request_obj = ApprovalWorkflowBulkActionRequests.prepare_request(
+            group_key, "APPROVED", sub_type, comment
+        )
+        try:
+            raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        except InvalidRequestError as err:
+            _raise_if_recipient_scoped(err, group_key)
+            raise
+        return ApprovalWorkflowBulkActionRequests.process_response(raw_json)
+
+    @validate_arguments
+    async def reject_all(
+        self,
+        group_key: str,
+        sub_type: Optional[ApprovalWorkflowRequestType] = None,
+        comment: Optional[str] = None,
+    ) -> ApprovalWorkflowBulkActionResponse:
+        """Bulk-reject all pending workflow tasks in a group."""
+        endpoint, request_obj = ApprovalWorkflowBulkActionRequests.prepare_request(
+            group_key, "REJECTED", sub_type, comment
+        )
+        try:
+            raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        except InvalidRequestError as err:
+            _raise_if_recipient_scoped(err, group_key)
+            raise
+        return ApprovalWorkflowBulkActionRequests.process_response(raw_json)

pyatlan/client/aio/client.py

@@ -49,6 +49,8 @@
 from pyatlan.client.aio.search_log import AsyncSearchLogClient
 from pyatlan.client.aio.sso import AsyncSSOClient
 from pyatlan.client.aio.task import AsyncTaskClient
+from pyatlan.client.aio.approval_workflow import AsyncApprovalWorkflowClient
+from pyatlan.client.aio.requests import AsyncRequestsClient
 from pyatlan.client.aio.token import AsyncTokenClient
 from pyatlan.client.aio.typedef import AsyncTypeDefClient
 from pyatlan.client.aio.app import AsyncAppClient
@@ -114,6 +116,10 @@ class AsyncAtlanClient(AtlanClient):
     _async_search_log_client: Optional[AsyncSearchLogClient] = PrivateAttr(default=None)
     _async_sso_client: Optional[AsyncSSOClient] = PrivateAttr(default=None)
     _async_task_client: Optional[AsyncTaskClient] = PrivateAttr(default=None)
+    _async_approval_workflow_client: Optional[AsyncApprovalWorkflowClient] = (
+        PrivateAttr(default=None)
+    )
+    _async_requests_client: Optional[AsyncRequestsClient] = PrivateAttr(default=None)
     _async_token_client: Optional[AsyncTokenClient] = PrivateAttr(default=None)
     _async_oauth_client_client: Optional[AsyncOAuthClient] = PrivateAttr(default=None)
     _async_typedef_client: Optional[AsyncTypeDefClient] = PrivateAttr(default=None)
@@ -361,6 +367,20 @@ def tasks(self) -> AsyncTaskClient:  # type: ignore[override]
             self._async_task_client = AsyncTaskClient(client=self)  # type: ignore[arg-type]
         return self._async_task_client
 
+    @property
+    def inbox(self) -> AsyncApprovalWorkflowClient:  # type: ignore[override]
+        """Async approval-workflow client (governance Inbox)"""
+        if self._async_approval_workflow_client is None:
+            self._async_approval_workflow_client = AsyncApprovalWorkflowClient(self)  # type: ignore[arg-type]
+        return self._async_approval_workflow_client
+
+    @property
+    def requests(self) -> AsyncRequestsClient:  # type: ignore[override]
+        """Async requests client for Metadata Inbox operations"""
+        if self._async_requests_client is None:
+            self._async_requests_client = AsyncRequestsClient(self)  # type: ignore[arg-type]
+        return self._async_requests_client
+
     @property
     def token(self) -> AsyncTokenClient:  # type: ignore[override]
         """Get async token client with same API as sync"""

pyatlan/client/aio/requests.py

@@ -0,0 +1,218 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 Atlan Pte. Ltd.
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic.v1 import validate_arguments
+
+from pyatlan.client.common import (
+    AsyncApiCaller,
+    RequestsAction,
+    RequestsCreate,
+    RequestsGetById,
+    RequestsList,
+    RequestsListActionable,
+)
+from pyatlan.errors import ErrorCode
+from pyatlan.model.aio.atlan_request import AsyncAtlanRequestResponse
+from pyatlan.model.atlan_request import (
+    AtlanRequest,
+    AtlanRequestsCriteria,
+    build_requests_filter,
+)
+from pyatlan.model.enums import AtlanRequestStatus, AtlanRequestType
+
+
+class AsyncRequestsClient:
+    """
+    Async client for operating on Atlan requests (the Metadata Inbox):
+    listing, retrieving, creating, approving and rejecting them.
+
+    Note: requests are only visible to the identity behind the API token —
+    an API key's service account must be an admin (or the designated
+    approver) to see and action requests raised for human users.
+    """
+
+    def __init__(self, client: AsyncApiCaller):
+        if not isinstance(client, AsyncApiCaller):
+            raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(
+                "client", "AsyncApiCaller"
+            )
+        self._client = client
+
+    @validate_arguments
+    async def list(
+        self,
+        status: Optional[AtlanRequestStatus] = None,
+        request_type: Optional[AtlanRequestType] = None,
+        destination_guid: Optional[str] = None,
+        destination_qualified_name: Optional[str] = None,
+        entity_type: Optional[str] = None,
+        created_by: Optional[str] = None,
+        post_filter: Optional[str] = None,
+        sort: Optional[str] = None,
+        count: bool = True,
+        offset: int = 0,
+        limit: int = 20,
+    ) -> AsyncAtlanRequestResponse:
+        """
+        List requests, optionally filtered by typed arguments.
+        Async-iterate the response to lazily page through ALL matches.
+
+        :param status: only requests with this status (AtlanRequestStatus, e.g. ACTIVE)
+        :param request_type: only this type (AtlanRequestType, e.g. ATTRIBUTE, ATLAN_TAG)
+        :param destination_guid: only requests against this asset GUID
+        :param destination_qualified_name: only requests against this qualified name
+        :param entity_type: only requests against this asset type
+        :param created_by: only requests raised by this user
+        :param post_filter: raw JSON filter (escape hatch — cannot be combined with the typed filters above)
+        :param sort: property by which to sort the results, e.g. `-createdAt`
+        :param count: whether to include the total number of records
+        :param offset: starting point for results, for paging
+        :param limit: maximum number of results per page
+        :raises AtlanError: on any error during API invocation.
+        :returns: a lazily-pageable response of requests
+        """
+        criteria = AtlanRequestsCriteria(
+            post_filter=build_requests_filter(
+                status=status,
+                request_type=request_type,
+                destination_guid=destination_guid,
+                destination_qualified_name=destination_qualified_name,
+                entity_type=entity_type,
+                created_by=created_by,
+                post_filter=post_filter,
+            ),
+            sort=sort,
+            count=count,
+            offset=offset,
+            limit=limit,
+        )
+        endpoint, query_params = RequestsList.prepare_request(criteria)
+        raw_json = await self._client._call_api(endpoint, query_params)
+        return AsyncAtlanRequestResponse(
+            client=self._client,
+            endpoint=RequestsList.ENDPOINT,
+            criteria=criteria,
+            start=offset,
+            size=limit,
+            **raw_json,
+        )
+
+    @validate_arguments
+    async def list_actionable(
+        self,
+        status: Optional[AtlanRequestStatus] = None,
+        request_type: Optional[AtlanRequestType] = None,
+        destination_guid: Optional[str] = None,
+        destination_qualified_name: Optional[str] = None,
+        entity_type: Optional[str] = None,
+        created_by: Optional[str] = None,
+        post_filter: Optional[str] = None,
+        sort: Optional[str] = None,
+        count: bool = True,
+        offset: int = 0,
+        limit: int = 20,
+    ) -> AsyncAtlanRequestResponse:
+        """
+        List requests the current identity can approve or reject.
+        Async-iterate the response to lazily page through ALL matches.
+
+        :param status: only requests with this status (AtlanRequestStatus, e.g. ACTIVE)
+        :param request_type: only this type (AtlanRequestType, e.g. ATTRIBUTE, ATLAN_TAG)
+        :param destination_guid: only requests against this asset GUID
+        :param destination_qualified_name: only requests against this qualified name
+        :param entity_type: only requests against this asset type
+        :param created_by: only requests raised by this user
+        :param post_filter: raw JSON filter (escape hatch — cannot be combined with the typed filters above)
+        :param sort: property by which to sort the results, e.g. `-createdAt`
+        :param count: whether to include the total number of records
+        :param offset: starting point for results, for paging
+        :param limit: maximum number of results per page
+        :raises AtlanError: on any error during API invocation.
+        :returns: a lazily-pageable response of requests
+        """
+        criteria = AtlanRequestsCriteria(
+            post_filter=build_requests_filter(
+                status=status,
+                request_type=request_type,
+                destination_guid=destination_guid,
+                destination_qualified_name=destination_qualified_name,
+                entity_type=entity_type,
+                created_by=created_by,
+                post_filter=post_filter,
+            ),
+            sort=sort,
+            count=count,
+            offset=offset,
+            limit=limit,
+        )
+        endpoint, query_params = RequestsListActionable.prepare_request(criteria)
+        raw_json = await self._client._call_api(endpoint, query_params)
+        return AsyncAtlanRequestResponse(
+            client=self._client,
+            endpoint=RequestsListActionable.ENDPOINT,
+            criteria=criteria,
+            start=offset,
+            size=limit,
+            **raw_json,
+        )
+
+    @validate_arguments
+    async def get(self, guid: str) -> Optional[AtlanRequest]:
+        """
+        Retrieve a single request by its GUID.
+
+        :param guid: unique identifier of the request
+        :raises AtlanError: on any error during API invocation.
+        :returns: the request, or None if it does not exist
+        """
+        endpoint = RequestsGetById.prepare_request(guid)
+        raw_json = await self._client._call_api(endpoint)
+        return RequestsGetById.process_response(raw_json)
+
+    async def create(self, request: AtlanRequest) -> Optional[AtlanRequest]:
+        """
+        Create (raise) a new request.
+
+        :param request: the request to create, e.g. via AttributeRequest.creator()
+        :raises AtlanError: on any error during API invocation.
+        :returns: the created request, including its server-assigned id
+        """
+        endpoint, request_obj = RequestsCreate.prepare_request(request)
+        raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        return RequestsCreate.process_response(raw_json)
+
+    @validate_arguments
+    async def approve(self, guid: str, message: Optional[str] = None) -> bool:
+        """
+        Approve a request. Approval applies the requested change.
+
+        :param guid: unique identifier of the request to approve
+        :param message: optional message to include with the approval
+        :raises AtlanError: on any error during API invocation.
+        :returns: True if the request was approved
+        """
+        endpoint, request_obj = RequestsAction.prepare_request(
+            guid, "approved", message
+        )
+        raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        return RequestsAction.process_response(raw_json)
+
+    @validate_arguments
+    async def reject(self, guid: str, message: Optional[str] = None) -> bool:
+        """
+        Reject a request.
+
+        :param guid: unique identifier of the request to reject
+        :param message: optional message to include with the rejection
+        :raises AtlanError: on any error during API invocation.
+        :returns: True if the request was rejected
+        """
+        endpoint, request_obj = RequestsAction.prepare_request(
+            guid, "rejected", message
+        )
+        raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        return RequestsAction.process_response(raw_json)