diff --git a/pyatlan/client/aio/approval_workflow.py b/pyatlan/client/aio/approval_workflow.py new file mode 100644 index 000000000..80203f3d1 --- /dev/null +++ b/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) diff --git a/pyatlan/client/aio/client.py b/pyatlan/client/aio/client.py index 9baa3e92a..5687fd02e 100644 --- a/pyatlan/client/aio/client.py +++ b/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""" diff --git a/pyatlan/client/aio/requests.py b/pyatlan/client/aio/requests.py new file mode 100644 index 000000000..19510f140 --- /dev/null +++ b/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) diff --git a/pyatlan/client/approval_workflow.py b/pyatlan/client/approval_workflow.py new file mode 100644 index 000000000..c7894cc6f --- /dev/null +++ b/pyatlan/client/approval_workflow.py @@ -0,0 +1,135 @@ +# 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 ( + ApiCaller, + ApprovalWorkflowBulkActionRequests, + ApprovalWorkflowGetRequest, +) +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 ApprovalWorkflowClient: + """ + A client for the governance-workflow approval system (the newer Inbox, + enabled via the `Governance Workflows and Inbox` feature). For the classic + Requests module use `client.requests` instead — tenants can have both. + + Inbox tasks are `Task` assets: list them with FluentSearch on the `Task` + type, then action them here using the task GUID (or the related asset + GUID) as the group key. + """ + + def __init__(self, client: ApiCaller): + if not isinstance(client, ApiCaller): + raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters( + "client", "ApiCaller" + ) + self._client = client + + @validate_arguments + 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 = self._client._call_api(endpoint) + return ApprovalWorkflowGetRequest.process_response(raw_json) + + @validate_arguments + 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. + + :param group_key: task GUID or related asset GUID whose pending tasks + should be approved + :param sub_type: optional filter (CHANGE_MANAGEMENT, DATA_ACCESS, + PUBLICATION_MANAGEMENT, POLICY_APPROVAL) + :param comment: optional comment for the approval + :raises AtlanError: on any error during API invocation. + :returns: number of tasks queued for (async) processing + """ + endpoint, request_obj = ApprovalWorkflowBulkActionRequests.prepare_request( + group_key, "APPROVED", sub_type, comment + ) + try: + raw_json = 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 + 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. + + :param group_key: task GUID or related asset GUID whose pending tasks + should be rejected + :param sub_type: optional filter (CHANGE_MANAGEMENT, DATA_ACCESS, + PUBLICATION_MANAGEMENT, POLICY_APPROVAL) + :param comment: optional comment for the rejection + :raises AtlanError: on any error during API invocation. + :returns: number of tasks queued for (async) processing + """ + endpoint, request_obj = ApprovalWorkflowBulkActionRequests.prepare_request( + group_key, "REJECTED", sub_type, comment + ) + try: + raw_json = 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) diff --git a/pyatlan/client/atlan.py b/pyatlan/client/atlan.py index 2fd0dcc37..14fcf39f8 100644 --- a/pyatlan/client/atlan.py +++ b/pyatlan/client/atlan.py @@ -52,6 +52,8 @@ from pyatlan.client.oauth_client import OAuthClient from pyatlan.client.open_lineage import OpenLineageClient from pyatlan.client.query import QueryClient +from pyatlan.client.approval_workflow import ApprovalWorkflowClient +from pyatlan.client.requests import RequestsClient from pyatlan.client.role import RoleClient from pyatlan.client.search_log import SearchLogClient from pyatlan.client.sso import SSOClient @@ -164,6 +166,10 @@ class AtlanClient(BaseSettings): _role_client: Optional[RoleClient] = PrivateAttr(default=None) _asset_client: Optional[AssetClient] = PrivateAttr(default=None) _typedef_client: Optional[TypeDefClient] = PrivateAttr(default=None) + _approval_workflow_client: Optional[ApprovalWorkflowClient] = PrivateAttr( + default=None + ) + _requests_client: Optional[RequestsClient] = PrivateAttr(default=None) _token_client: Optional[TokenClient] = PrivateAttr(default=None) _oauth_client_client: Optional[OAuthClient] = PrivateAttr(default=None) _user_client: Optional[UserClient] = PrivateAttr(default=None) @@ -395,6 +401,18 @@ def queries(self) -> QueryClient: self._query_client = QueryClient(client=self) return self._query_client + @property + def inbox(self) -> ApprovalWorkflowClient: + if self._approval_workflow_client is None: + self._approval_workflow_client = ApprovalWorkflowClient(client=self) + return self._approval_workflow_client + + @property + def requests(self) -> RequestsClient: + if self._requests_client is None: + self._requests_client = RequestsClient(client=self) + return self._requests_client + @property def token(self) -> TokenClient: if self._token_client is None: diff --git a/pyatlan/client/common/__init__.py b/pyatlan/client/common/__init__.py index 3ba1baa14..074b609f8 100644 --- a/pyatlan/client/common/__init__.py +++ b/pyatlan/client/common/__init__.py @@ -117,6 +117,17 @@ from .query import QueryStream # Role shared logic classes +from .approval_workflow import ( + ApprovalWorkflowBulkActionRequests, + ApprovalWorkflowGetRequest, +) +from .requests import ( + RequestsAction, + RequestsCreate, + RequestsGetById, + RequestsList, + RequestsListActionable, +) from .role import RoleGet, RoleGetAll # Search log shared logic classes @@ -277,6 +288,13 @@ # Query shared logic classes "QueryStream", # Role shared logic classes + "ApprovalWorkflowBulkActionRequests", + "ApprovalWorkflowGetRequest", + "RequestsAction", + "RequestsCreate", + "RequestsGetById", + "RequestsList", + "RequestsListActionable", "RoleGet", "RoleGetAll", # Search log shared logic classes diff --git a/pyatlan/client/common/approval_workflow.py b/pyatlan/client/common/approval_workflow.py new file mode 100644 index 000000000..7c25d9af5 --- /dev/null +++ b/pyatlan/client/common/approval_workflow.py @@ -0,0 +1,54 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from __future__ import annotations + +from typing import Optional + +from pyatlan.client.constants import ( + BULK_ACTION_APPROVAL_WORKFLOW_REQUESTS, + GET_APPROVAL_WORKFLOW_REQUEST, +) +from pyatlan.model.approval_workflow import ( + ApprovalWorkflowBulkAction, + ApprovalWorkflowBulkActionResponse, + ApprovalWorkflowRequest, +) + + +class ApprovalWorkflowGetRequest: + """Shared logic for retrieving one approval-workflow request by GUID.""" + + @staticmethod + def prepare_request(guid: str): + return GET_APPROVAL_WORKFLOW_REQUEST.format_path({"request_guid": guid}) + + @staticmethod + def process_response(raw_json) -> Optional[ApprovalWorkflowRequest]: + if isinstance(raw_json, list): + raw_json = raw_json[0] if raw_json else None + return ApprovalWorkflowRequest(**raw_json) if raw_json else None + + +class ApprovalWorkflowBulkActionRequests: + """Shared logic for bulk-approving or bulk-rejecting workflow tasks.""" + + @staticmethod + def prepare_request( + group_key: str, + decision: str, + sub_type: Optional[str] = None, + comment: Optional[str] = None, + ) -> tuple: + body = ApprovalWorkflowBulkAction(group_key=group_key, decision=decision) + if sub_type is not None: + body.sub_type = getattr(sub_type, "value", sub_type) + if comment is not None: + body.comment = comment + return BULK_ACTION_APPROVAL_WORKFLOW_REQUESTS.format_path_with_params(), body + + @staticmethod + def process_response(raw_json) -> ApprovalWorkflowBulkActionResponse: + if isinstance(raw_json, dict): + return ApprovalWorkflowBulkActionResponse(**raw_json) + return ApprovalWorkflowBulkActionResponse(message=str(raw_json)) diff --git a/pyatlan/client/common/requests.py b/pyatlan/client/common/requests.py new file mode 100644 index 000000000..ea12175e4 --- /dev/null +++ b/pyatlan/client/common/requests.py @@ -0,0 +1,76 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from __future__ import annotations + +from typing import Optional + +from pyatlan.client.constants import ( + ACTION_REQUEST, + CREATE_REQUEST, + GET_ACTIONABLE_REQUESTS, + GET_REQUEST_BY_ID, + GET_REQUESTS, +) +from pyatlan.model.atlan_request import ( + AtlanRequest, + AtlanRequestAction, + AtlanRequestsCriteria, +) + + +class RequestsList: + """Shared logic for listing requests (Metadata Inbox).""" + + ENDPOINT = GET_REQUESTS + + @classmethod + def prepare_request(cls, criteria: AtlanRequestsCriteria) -> tuple: + return cls.ENDPOINT.format_path_with_params(), criteria.query_params + + +class RequestsListActionable(RequestsList): + """Shared logic for listing requests actionable by the current identity.""" + + ENDPOINT = GET_ACTIONABLE_REQUESTS + + +class RequestsGetById: + """Shared logic for retrieving a single request by its GUID.""" + + @staticmethod + def prepare_request(guid: str): + return GET_REQUEST_BY_ID.format_path({"request_id": guid}) + + @staticmethod + def process_response(raw_json) -> Optional[AtlanRequest]: + # The endpoint may return the request object directly or a + # single-element list wrapping it. + if isinstance(raw_json, list): + raw_json = raw_json[0] if raw_json else None + return AtlanRequest(**raw_json) if raw_json else None + + +class RequestsCreate: + """Shared logic for creating (raising) a request.""" + + @staticmethod + def prepare_request(request: AtlanRequest) -> tuple: + return CREATE_REQUEST.format_path_with_params(), request + + @staticmethod + def process_response(raw_json) -> Optional[AtlanRequest]: + return AtlanRequest(**raw_json) if isinstance(raw_json, dict) else None + + +class RequestsAction: + """Shared logic for approving or rejecting a request.""" + + @staticmethod + def prepare_request(guid: str, action: str, message: Optional[str] = None) -> tuple: + body = AtlanRequestAction(action=action, message=message or "") + return ACTION_REQUEST.format_path({"request_id": guid}), body + + @staticmethod + def process_response(raw_json) -> bool: + return raw_json == "success" diff --git a/pyatlan/client/constants.py b/pyatlan/client/constants.py index 11d7842d1..92b58bd06 100644 --- a/pyatlan/client/constants.py +++ b/pyatlan/client/constants.py @@ -153,6 +153,46 @@ ) # API token APIs +APPROVAL_WORKFLOW_REQUESTS_API = "approval-workflow-requests" +GET_APPROVAL_WORKFLOW_REQUEST = API( + APPROVAL_WORKFLOW_REQUESTS_API + "/{request_guid}", + HTTPMethod.GET, + HTTPStatus.OK, + endpoint=EndPoint.HERACLES, +) +BULK_ACTION_APPROVAL_WORKFLOW_REQUESTS = API( + APPROVAL_WORKFLOW_REQUESTS_API + "/actions/bulk", + HTTPMethod.PUT, + HTTPStatus.OK, + endpoint=EndPoint.HERACLES, +) + +REQUESTS_API = "requests" +GET_REQUESTS = API( + REQUESTS_API, HTTPMethod.GET, HTTPStatus.OK, endpoint=EndPoint.HERACLES +) +GET_ACTIONABLE_REQUESTS = API( + REQUESTS_API + "/actionable", + HTTPMethod.GET, + HTTPStatus.OK, + endpoint=EndPoint.HERACLES, +) +CREATE_REQUEST = API( + REQUESTS_API, HTTPMethod.POST, HTTPStatus.OK, endpoint=EndPoint.HERACLES +) +GET_REQUEST_BY_ID = API( + REQUESTS_API + "/{request_id}", + HTTPMethod.GET, + HTTPStatus.OK, + endpoint=EndPoint.HERACLES, +) +ACTION_REQUEST = API( + REQUESTS_API + "/{request_id}/action", + HTTPMethod.POST, + HTTPStatus.OK, + endpoint=EndPoint.HERACLES, +) + GET_API_TOKENS = API( TOKENS_API, HTTPMethod.GET, HTTPStatus.OK, endpoint=EndPoint.HERACLES ) diff --git a/pyatlan/client/requests.py b/pyatlan/client/requests.py new file mode 100644 index 000000000..1df56d912 --- /dev/null +++ b/pyatlan/client/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 ( + ApiCaller, + RequestsAction, + RequestsCreate, + RequestsGetById, + RequestsList, + RequestsListActionable, +) +from pyatlan.errors import ErrorCode +from pyatlan.model.atlan_request import ( + AtlanRequest, + AtlanRequestResponse, + AtlanRequestsCriteria, + build_requests_filter, +) +from pyatlan.model.enums import AtlanRequestStatus, AtlanRequestType + + +class RequestsClient: + """ + A 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: ApiCaller): + if not isinstance(client, ApiCaller): + raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters( + "client", "ApiCaller" + ) + self._client = client + + @validate_arguments + 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, + ) -> AtlanRequestResponse: + """ + List requests, optionally filtered by typed arguments. + 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 = self._client._call_api(endpoint, query_params) + return AtlanRequestResponse( + client=self._client, + endpoint=RequestsList.ENDPOINT, + criteria=criteria, + start=offset, + size=limit, + **raw_json, + ) + + @validate_arguments + 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, + ) -> AtlanRequestResponse: + """ + List requests the current identity can approve or reject. + 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 = self._client._call_api(endpoint, query_params) + return AtlanRequestResponse( + client=self._client, + endpoint=RequestsListActionable.ENDPOINT, + criteria=criteria, + start=offset, + size=limit, + **raw_json, + ) + + @validate_arguments + 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 = self._client._call_api(endpoint) + return RequestsGetById.process_response(raw_json) + + 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 = self._client._call_api(endpoint, request_obj=request_obj) + return RequestsCreate.process_response(raw_json) + + @validate_arguments + 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 = self._client._call_api(endpoint, request_obj=request_obj) + return RequestsAction.process_response(raw_json) + + @validate_arguments + 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 = self._client._call_api(endpoint, request_obj=request_obj) + return RequestsAction.process_response(raw_json) diff --git a/pyatlan/model/aio/__init__.py b/pyatlan/model/aio/__init__.py index 7e13dea06..bb7d8f3ff 100644 --- a/pyatlan/model/aio/__init__.py +++ b/pyatlan/model/aio/__init__.py @@ -13,6 +13,7 @@ from .audit import AsyncAuditSearchResults from .core import AsyncAtlanRequest, AsyncAtlanResponse from .custom_metadata import AsyncCustomMetadataDict, AsyncCustomMetadataProxy +from .atlan_request import AsyncAtlanRequestResponse from .group import AsyncGroupResponse from .keycloak_events import AsyncAdminEventResponse, AsyncKeycloakEventResponse from .lineage import AsyncLineageListResults @@ -45,6 +46,7 @@ # User response "AsyncUserResponse", # Group response + "AsyncAtlanRequestResponse", "AsyncGroupResponse", # Workflow search response "AsyncWorkflowSearchResponse", diff --git a/pyatlan/model/aio/atlan_request.py b/pyatlan/model/aio/atlan_request.py new file mode 100644 index 000000000..2761fc879 --- /dev/null +++ b/pyatlan/model/aio/atlan_request.py @@ -0,0 +1,77 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, AsyncGenerator, List, Optional + +from pydantic.v1 import Field, PrivateAttr, ValidationError, parse_obj_as + +from pyatlan.errors import ErrorCode +from pyatlan.model.atlan_request import AtlanRequest, AtlanRequestsCriteria +from pyatlan.model.core import AtlanObject +from pyatlan.utils import API + +if TYPE_CHECKING: + from pyatlan.client.aio.client import AsyncAtlanClient + + +class AsyncAtlanRequestResponse(AtlanObject): + """Async version of AtlanRequestResponse with async pagination support.""" + + _size: int = PrivateAttr() + _start: int = PrivateAttr() + _endpoint: API = PrivateAttr() + _client: AsyncAtlanClient = PrivateAttr() + _criteria: AtlanRequestsCriteria = PrivateAttr() + total_record: Optional[int] = Field( + default=None, description="Total number of requests." + ) + filter_record: Optional[int] = Field( + default=None, description="Number of requests matching the filter." + ) + records: Optional[List[AtlanRequest]] = Field( + default=None, description="Requests in this page of results." + ) + + def __init__(self, **data: Any): + super().__init__(**data) + self._endpoint = data.get("endpoint") # type: ignore[assignment] + self._client = data.get("client") # type: ignore[assignment] + self._criteria = data.get("criteria") # type: ignore[assignment] + self._start = data.get("start") or 0 + self._size = data.get("size") or 20 + + def current_page(self) -> Optional[List[AtlanRequest]]: + return self.records + + async def next_page(self, start=None, size=None) -> bool: + self._start = start or self._start + self._size + if size: + self._size = size + return await self._get_next_page() if self.records else False + + async def _get_next_page(self) -> bool: + self._criteria.offset = self._start + self._criteria.limit = self._size + raw_json = await self._client._call_api( + api=self._endpoint.format_path_with_params(), + query_params=self._criteria.query_params, + ) + if not raw_json.get("records"): + self.records = [] + return False + try: + self.records = parse_obj_as(List[AtlanRequest], raw_json.get("records")) + except ValidationError as err: + raise ErrorCode.JSON_ERROR.exception_with_parameters( + raw_json, 200, str(err) + ) from err + return True + + async def __aiter__(self) -> AsyncGenerator[AtlanRequest, None]: # type: ignore[misc] + while self.records: + for record in self.records: + yield record + if not await self.next_page(): + break diff --git a/pyatlan/model/approval_workflow.py b/pyatlan/model/approval_workflow.py new file mode 100644 index 000000000..4c47dc66f --- /dev/null +++ b/pyatlan/model/approval_workflow.py @@ -0,0 +1,116 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from pydantic.v1 import Field + +from pyatlan.model.core import AtlanObject + + +class ApprovalWorkflowObject(AtlanObject): + """Base for approval-workflow API models: this API family uses snake_case + on the wire, so the camelCase alias generator is disabled.""" + + class Config(AtlanObject.Config): + alias_generator = staticmethod( # type: ignore[assignment] + lambda field_name: field_name + ) + + +class ApprovalWorkflowRequest(ApprovalWorkflowObject): + """A governance-workflow approval request (the newer Inbox system, + enabled by the `Governance Workflows and Inbox` feature).""" + + guid: Optional[str] = Field(default=None, description="Unique identifier.") + name: Optional[str] = Field(default=None, description="Name of the request.") + qualified_name: Optional[str] = Field( + default=None, description="Qualified name of the request." + ) + description: Optional[str] = Field( + default=None, description="Description of the request." + ) + approval_workflow_guid: Optional[str] = Field( + default=None, description="GUID of the workflow this request instantiates." + ) + approval_workflow_request_type: Optional[str] = Field( + default=None, + description=( + "Type of the request, e.g. CHANGE_MANAGEMENT, DATA_ACCESS, " + "PUBLICATION_MANAGEMENT or POLICY_APPROVAL." + ), + ) + request_on_asset_guid: Optional[str] = Field( + default=None, description="GUID of the asset the request was raised on." + ) + config: Optional[Dict[str, Any]] = Field( + default=None, description="Configuration of the request." + ) + status: Optional[str] = Field(default=None, description="Status of the request.") + expires_at: Optional[str] = Field( + default=None, description="When the request expires, if it does." + ) + comment: Optional[str] = Field( + default=None, description="Comment attached to the request, if any." + ) + created_by: Optional[str] = Field( + default=None, description="User who raised the request." + ) + updated_by: Optional[str] = Field( + default=None, description="User who last updated the request." + ) + created_at: Optional[str] = Field( + default=None, description="When the request was created." + ) + updated_at: Optional[str] = Field( + default=None, description="When the request was last updated." + ) + approval_details: Optional[Any] = Field( + default=None, + description=( + "Details of the approval configuration/stages — wire shape " + "varies by platform version; kept untyped so every variant parses." + ), + ) + action_details: Optional[Any] = Field( + default=None, + description=( + "Details of actions taken on the request — wire shape varies " + "by platform version; kept untyped so every variant parses." + ), + ) + + +class ApprovalWorkflowBulkAction(ApprovalWorkflowObject): + """Body for bulk-approving or bulk-rejecting workflow tasks.""" + + group_key: str = Field( + description=( + "Group identifier: the task GUID or the related asset GUID " + "(taskRelatedAssetGuid) whose pending tasks should be actioned." + ) + ) + decision: str = Field(description="`APPROVED` or `REJECTED`.") + sub_type: Optional[str] = Field( + default=None, + description=( + "Optional task sub-type filter: CHANGE_MANAGEMENT, DATA_ACCESS, " + "PUBLICATION_MANAGEMENT or POLICY_APPROVAL." + ), + ) + comment: Optional[str] = Field( + default=None, description="Optional comment for the approval/rejection." + ) + + +class ApprovalWorkflowBulkActionResponse(ApprovalWorkflowObject): + """Response of a bulk action: tasks are queued for async processing.""" + + total_tasks: Optional[int] = Field( + default=None, description="Number of tasks queued for async processing." + ) + message: Optional[str] = Field( + default=None, description="Status message from the server." + ) diff --git a/pyatlan/model/atlan_request.py b/pyatlan/model/atlan_request.py new file mode 100644 index 000000000..07a885cb6 --- /dev/null +++ b/pyatlan/model/atlan_request.py @@ -0,0 +1,303 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from __future__ import annotations + +import json +from typing import Any, Dict, Generator, List, Optional, Union + +from pydantic.v1 import Field, PrivateAttr, ValidationError, parse_obj_as + +from pyatlan.errors import ErrorCode +from pyatlan.model.core import AtlanObject +from pyatlan.model.enums import AtlanRequestStatus, AtlanRequestType +from pyatlan.utils import API + + +def build_requests_filter( + 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, +) -> Optional[str]: + """Build the JSON filter for listing requests from typed arguments. + + Each argument is an exact match on the corresponding request field; + multiple arguments are combined with AND. ``post_filter`` is the raw + escape hatch — when given, it is used as-is and the typed arguments + must not be combined with it. + """ + typed: Dict[str, Any] = {} + if status is not None: + # status must go through the $in operator — plain equality is + # ignored by the endpoint (grammar mirrored from the Atlan UI) + typed["status"] = {"$in": [AtlanRequestStatus(status).value]} + if request_type is not None: + typed["requestType"] = {"$in": [AtlanRequestType(request_type).value]} + if destination_guid is not None: + typed["destinationGuid"] = destination_guid + if destination_qualified_name is not None: + typed["destinationQualifiedName"] = destination_qualified_name + if entity_type is not None: + typed["entityType"] = entity_type + if created_by is not None: + typed["createdBy"] = created_by + if post_filter is not None: + if typed: + raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters( + "post_filter", "raw filter cannot be combined with typed filters" + ) + return post_filter + if not typed: + return None + # Same shape the Atlan UI sends: AND of the duplicate-exclusion clause + # and the typed conditions. + return json.dumps({"$and": [{"isDuplicate": False}, typed]}) + + +class AtlanRequest(AtlanObject): + """A request (Metadata Inbox item) in Atlan, such as a suggested + attribute change, term link, or Atlan tag attachment, that can be + approved or rejected.""" + + id: Optional[str] = Field( + default=None, description="Unique identifier for the request (GUID)." + ) + version: Optional[str] = Field( + default=None, description="Version of the request in Atlan's internal store." + ) + is_active: Optional[bool] = Field( + default=None, description="Whether the request is still open (True) or not." + ) + created_at: Optional[int] = Field( + default=None, + description="Time (epoch millis) at which the request was created.", + ) + updated_at: Optional[int] = Field( + default=None, + description="Time (epoch millis) at which the request was last updated.", + ) + created_by: Optional[str] = Field( + default=None, description="User who created the request." + ) + tenant_id: Optional[str] = Field( + default=None, description="Name of the tenant (usually `default`)." + ) + source_type: Optional[str] = Field( + default=None, + description=( + "`static` for ATTRIBUTE and CUSTOM_METADATA request types, " + "`atlas` for other request types." + ), + ) + source_guid: Optional[str] = Field( + default=None, description="GUID of the source asset, if any." + ) + source_qualified_name: Optional[str] = Field( + default=None, description="Qualified name of the source asset, if any." + ) + source_attribute: Optional[str] = Field( + default=None, description="Attribute on the source asset, if any." + ) + destination_guid: Optional[str] = Field( + default=None, description="GUID of the asset the request was made against." + ) + destination_qualified_name: Optional[str] = Field( + default=None, + description="Qualified name of the asset the request was made against.", + ) + destination_attribute: Optional[str] = Field( + default=None, description="Attribute the request was made against, if any." + ) + destination_value: Optional[str] = Field( + default=None, description="Requested value for the attribute." + ) + destination_value_type: Optional[str] = Field( + default=None, description="Type of the destination attribute value." + ) + entity_type: Optional[str] = Field( + default=None, description="Type of the asset the request was made against." + ) + request_type: Optional[str] = Field( + default=None, + description=( + "Type of the request: `attribute`, `term_link`, " + "`attach_classification`, or `bm_attribute`." + ), + ) + approval_type: Optional[str] = Field( + default=None, + description="How the request must be approved: `single`, `unanimous` or `consesus`.", + ) + approved_by: Optional[Any] = Field( + default=None, + description=( + "Who approved the request. The wire shape varies by platform " + "version: a username string, a list of usernames, or a list of " + "approver-detail objects — kept untyped so every variant parses." + ), + ) + rejected_by: Optional[Any] = Field( + default=None, + description=( + "Who rejected the request. The wire shape varies by platform " + "version: a username string, a list of usernames, or a list of " + "approver-detail objects — kept untyped so every variant parses." + ), + ) + status: Optional[str] = Field( + default=None, + description="Status of the request: `active`, `approved` or `rejected`.", + ) + message: Optional[str] = Field( + default=None, description="Message to include with the request, if any." + ) + payload: Optional[Dict[str, Any]] = Field( + default=None, + description=( + "Payload for requests that carry one (Atlan tag details for " + "`attach_classification`, custom metadata values for `bm_attribute`)." + ), + ) + destination_entity: Optional[Dict[str, Any]] = Field( + default=None, + description="Limited details of the asset the request was made against.", + ) + + +class AttributeRequest(AtlanRequest): + """A request to change a single attribute value on an asset.""" + + @classmethod + def creator( + cls, + *, + destination_guid: str, + destination_qualified_name: str, + destination_attribute: str, + destination_value: str, + entity_type: str, + ) -> AttributeRequest: + """Create a request to set an attribute value on an asset. + + All wire-required fields are passed explicitly so that pyatlan's + exclude_unset serialization sends them (declared defaults are never + serialized — see BLDX-1589). + + :param destination_guid: GUID of the asset to change + :param destination_qualified_name: qualified name of the asset to change + :param destination_attribute: attribute to change (e.g. `userDescription`) + :param destination_value: value requested for the attribute + :param entity_type: type of the asset (e.g. `AtlasGlossaryTerm`) + """ + return cls( + request_type="attribute", + source_type="static", + approval_type="single", + destination_guid=destination_guid, + destination_qualified_name=destination_qualified_name, + destination_attribute=destination_attribute, + destination_value=destination_value, + entity_type=entity_type, + ) + + +class AtlanRequestAction(AtlanObject): + """Body for approving or rejecting a request.""" + + action: str = Field(description="Action to take: `approved` or `rejected`.") + message: Optional[str] = Field( + default=None, description="Optional message to include with the action." + ) + + +class AtlanRequestsCriteria(AtlanObject): + """Criteria (query parameters) for listing requests, used for paging.""" + + post_filter: Optional[str] = Field( + default=None, description="JSON filter for the list of requests." + ) + sort: Optional[str] = Field( + default=None, description="Property by which to sort the results." + ) + count: bool = Field( + default=True, description="Whether to include an overall count." + ) + offset: int = Field(default=0, description="Starting point when paging.") + limit: int = Field(default=20, description="Maximum requests per page.") + + @property + def query_params(self) -> dict: + qp: Dict[str, object] = {} + if self.post_filter: + qp["filter"] = self.post_filter + if self.sort: + qp["sort"] = self.sort + qp["count"] = self.count + qp["offset"] = self.offset + qp["limit"] = self.limit + return qp + + +class AtlanRequestResponse(AtlanObject): + """Paged response of requests. Iterate it to lazily page through ALL + matching requests (same pattern as UserResponse / GroupResponse).""" + + _size: int = PrivateAttr() + _start: int = PrivateAttr() + _endpoint: API = PrivateAttr() + _client: Any = PrivateAttr() + _criteria: AtlanRequestsCriteria = PrivateAttr() + total_record: Optional[int] = Field( + default=None, description="Total number of requests." + ) + filter_record: Optional[int] = Field( + default=None, description="Number of requests matching the filter." + ) + records: Optional[List[AtlanRequest]] = Field( + default=None, description="Requests in this page of results." + ) + + def __init__(self, **data: Any): + super().__init__(**data) + self._endpoint = data.get("endpoint") # type: ignore[assignment] + self._client = data.get("client") + self._criteria = data.get("criteria") # type: ignore[assignment] + self._start = data.get("start") or 0 + self._size = data.get("size") or 20 + + def current_page(self) -> Optional[List[AtlanRequest]]: + return self.records + + def next_page(self, start=None, size=None) -> bool: + self._start = start or self._start + self._size + if size: + self._size = size + return self._get_next_page() if self.records else False + + def _get_next_page(self) -> bool: + self._criteria.offset = self._start + self._criteria.limit = self._size + raw_json = self._client._call_api( + api=self._endpoint.format_path_with_params(), + query_params=self._criteria.query_params, + ) + if not raw_json.get("records"): + self.records = [] + return False + try: + self.records = parse_obj_as(List[AtlanRequest], raw_json.get("records")) + except ValidationError as err: + raise ErrorCode.JSON_ERROR.exception_with_parameters( + raw_json, 200, str(err) + ) from err + return True + + def __iter__(self) -> Generator[AtlanRequest, None, None]: # type: ignore[override] + while True: + yield from self.current_page() or [] + if not self.next_page(): + break diff --git a/pyatlan/model/enums.py b/pyatlan/model/enums.py index b4b41c569..f49ed83f6 100644 --- a/pyatlan/model/enums.py +++ b/pyatlan/model/enums.py @@ -1956,6 +1956,38 @@ class AtlanIcon(str, Enum): YOUTUBE_LOGO = "PhYoutubeLogo" +class ApprovalWorkflowRequestType(str, Enum): + """Type of a governance-workflow (Inbox) approval request/task.""" + + CHANGE_MANAGEMENT = "CHANGE_MANAGEMENT" + DATA_ACCESS = "DATA_ACCESS" + PUBLICATION_MANAGEMENT = "PUBLICATION_MANAGEMENT" + POLICY_APPROVAL = "POLICY_APPROVAL" + + +class AtlanRequestStatus(str, Enum): + """Status of a request in the (classic) Metadata Inbox.""" + + ACTIVE = "active" + APPROVED = "approved" + REJECTED = "rejected" + + +class AtlanRequestType(str, Enum): + """Type of a request in the (classic) Metadata Inbox.""" + + ATTRIBUTE = "attribute" + CUSTOM_METADATA = "bm_attribute" + TERM_LINK = "term_link" + ATLAN_TAG = "attach_classification" + CREATE_TYPEDEF = "create_typedef" + CREATE_GLOSSARY = "create_glossary" + CREATE_CATEGORY = "create_category" + CREATE_TERM = "create_term" + PERSONA_ACCESS = "persona_access" + PURPOSE_POLICY_ACCESS = "purpose_policy_access" + + class AtlanTagColor(str, Enum): GREEN = "Green" YELLOW = "Yellow" diff --git a/tests/integration/atlan_requests_test.py b/tests/integration/atlan_requests_test.py new file mode 100644 index 000000000..38b394f0c --- /dev/null +++ b/tests/integration/atlan_requests_test.py @@ -0,0 +1,107 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +"""Live tests for client.requests (the classic Metadata Inbox / Requests +module) — BLDX-1611. Fully self-contained: creates its own requests against +a disposable term and verifies approval applies the change on the backend. + +The governance-workflow Inbox (client.inbox) lives in inbox_test.py. +""" +import time +from typing import Generator + +import pytest +from pydantic.v1 import StrictStr + +from pyatlan.client.atlan import AtlanClient +from pyatlan.model.assets import AtlasGlossary, AtlasGlossaryTerm +from pyatlan.model.atlan_request import AttributeRequest +from pyatlan.model.enums import AtlanRequestStatus, AtlanRequestType +from tests.integration.client import TestId, delete_asset + +MODULE_NAME = TestId.make_unique("REQS") + + +@pytest.fixture(scope="module") +def glossary(client: AtlanClient) -> Generator[AtlasGlossary, None, None]: + g = AtlasGlossary.create(name=StrictStr(MODULE_NAME)) + g = client.asset.save(g).assets_created(AtlasGlossary)[0] + yield g + delete_asset(client, guid=g.guid, asset_type=AtlasGlossary) + + +@pytest.fixture(scope="module") +def term( + client: AtlanClient, glossary: AtlasGlossary +) -> Generator[AtlasGlossaryTerm, None, None]: + t = AtlasGlossaryTerm.create( + name=StrictStr(f"{MODULE_NAME}-term"), glossary_guid=glossary.guid + ) + t = client.asset.save(t).assets_created(AtlasGlossaryTerm)[0] + yield t + delete_asset(client, guid=t.guid, asset_type=AtlasGlossaryTerm) + + +def _create_request(client: AtlanClient, term: AtlasGlossaryTerm, value: str): + assert term.guid and term.qualified_name + request = AttributeRequest.creator( + destination_guid=term.guid, + destination_qualified_name=term.qualified_name, + destination_attribute="userDescription", + destination_value=value, + entity_type="AtlasGlossaryTerm", + ) + created = client.requests.create(request) + assert created and created.id + assert created.status == AtlanRequestStatus.ACTIVE.value + return created + + +def test_create_list_get_request(client: AtlanClient, term: AtlasGlossaryTerm): + created = _create_request(client, term, "requests-test-listed") + + # typed filter finds it without knowing the JSON filter grammar + response = client.requests.list( + destination_guid=term.guid, status=AtlanRequestStatus.ACTIVE + ) + found = [r for r in response.records or [] if r.id == created.id] + assert found, "created request not returned by typed-filter list()" + assert found[0].request_type == AtlanRequestType.ATTRIBUTE.value + + fetched = client.requests.get(guid=created.id) + assert fetched and fetched.id == created.id + + +@pytest.mark.order(after="test_create_list_get_request") +def test_approve_applies_the_change(client: AtlanClient, term: AtlasGlossaryTerm): + created = _create_request(client, term, "requests-test-approved") + + assert client.requests.approve(guid=created.id, message="integration approve") + + def _applied() -> bool: + asset: AtlasGlossaryTerm = client.asset.get_by_guid( + term.guid, ignore_relationships=True + ) + return asset.user_description == "requests-test-approved" + + deadline = time.time() + 30 + applied = False + while time.time() < deadline: + if _applied(): + applied = True + break + time.sleep(2) + assert applied, "approved request did not apply the attribute change" + + +@pytest.mark.order(after="test_approve_applies_the_change") +def test_reject_does_not_apply(client: AtlanClient, term: AtlasGlossaryTerm): + created = _create_request(client, term, "requests-test-rejected") + + assert client.requests.reject(guid=created.id, message="integration reject") + time.sleep(3) + asset: AtlasGlossaryTerm = client.asset.get_by_guid( + term.guid, ignore_relationships=True + ) + assert asset.user_description != "requests-test-rejected", ( + "rejected request must not apply its change" + ) diff --git a/tests/integration/inbox_test.py b/tests/integration/inbox_test.py new file mode 100644 index 000000000..aa827af55 --- /dev/null +++ b/tests/integration/inbox_test.py @@ -0,0 +1,74 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +"""Live tests for client.inbox (governance-workflow approvals) — BLDX-1611. + +Workflow tasks cannot be self-created: raising one requires an existing +governance workflow on the tenant (the `Governance Workflows and Inbox` +Labs feature, plus a configured workflow — `workflow_guids` is required by +the create endpoint). These tests are therefore gated on environment +variables naming pre-seeded pending tasks: + +- ATLAN_TEST_INBOX_TASK_GUID: guid of ONE pending task + → bulk action with a task guid actions exactly that task (group of one) +- ATLAN_TEST_INBOX_ASSET_GUID: guid of an asset with TWO OR MORE pending + tasks → bulk action with the asset guid actions the WHOLE group + +The classic Requests module (client.requests) lives in +atlan_requests_test.py and is fully self-contained. +""" +import os + +import pytest + +from pyatlan.client.atlan import AtlanClient + +TASK_GUID = os.environ.get("ATLAN_TEST_INBOX_TASK_GUID") +ASSET_GUID = os.environ.get("ATLAN_TEST_INBOX_ASSET_GUID") + + +@pytest.mark.skipif( + not TASK_GUID, + reason="needs a pending workflow task — set ATLAN_TEST_INBOX_TASK_GUID", +) +def test_task_guid_actions_single_task(client: AtlanClient): + """A task guid is a group of one: exactly that task is queued.""" + response = client.inbox.reject_all( + group_key=str(TASK_GUID), comment="integration single-task reject" + ) + assert response.message + assert response.total_tasks == 1, ( + f"task-guid group must action exactly one task, got {response.total_tasks}" + ) + + +@pytest.mark.skipif( + not ASSET_GUID, + reason=( + "needs an asset with 2+ pending workflow tasks — " + "set ATLAN_TEST_INBOX_ASSET_GUID" + ), +) +def test_asset_guid_actions_whole_group(client: AtlanClient): + """An asset guid actions EVERY pending task on that asset — the case + that distinguishes bulk from single approval.""" + response = client.inbox.approve_all( + group_key=str(ASSET_GUID), comment="integration group approve" + ) + assert response.message + assert response.total_tasks and response.total_tasks >= 2, ( + f"asset-group bulk expected 2+ tasks queued, got {response.total_tasks} — " + "seed at least two pending tasks on the asset before running" + ) + + +@pytest.mark.skipif( + not TASK_GUID, + reason="needs a pending workflow task — set ATLAN_TEST_INBOX_TASK_GUID", +) +def test_get_workflow_request(client: AtlanClient): + """A workflow request fetched by guid parses the snake_case wire.""" + request = client.inbox.get(guid=str(TASK_GUID)) + # the task guid may differ from the workflow-request guid; a None here + # is a mapping finding, not a failure — assert only on parse success + if request is not None: + assert request.guid diff --git a/tests/unit/test_approval_workflow_client.py b/tests/unit/test_approval_workflow_client.py new file mode 100644 index 000000000..631c6e894 --- /dev/null +++ b/tests/unit/test_approval_workflow_client.py @@ -0,0 +1,117 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +from json import loads +from unittest.mock import Mock + +import pytest + +from pyatlan.client.approval_workflow import ApprovalWorkflowClient +from pyatlan.client.common import ApiCaller +from pyatlan.errors import InvalidRequestError +from pyatlan.model.approval_workflow import ApprovalWorkflowRequest +from pyatlan.model.enums import ApprovalWorkflowRequestType + +WF_REQUEST_GUID = "1a2b3c4d-1111-2222-3333-444455556666" +ASSET_GUID = "9c67229e-f345-4de4-b046-c3b6cb2a5c34" + +RAW_WF_REQUEST = { + "guid": WF_REQUEST_GUID, + "name": "Access request", + "approval_workflow_request_type": "DATA_ACCESS", + "request_on_asset_guid": ASSET_GUID, + "status": "PENDING", + "created_by": "aryaman-alt", +} + + +@pytest.fixture(autouse=True) +def set_env(monkeypatch): + monkeypatch.setenv("ATLAN_BASE_URL", "https://test.atlan.com") + monkeypatch.setenv("ATLAN_API_KEY", "test-api-key") + + +@pytest.fixture() +def mock_api_caller(): + return Mock(spec=ApiCaller) + + +@pytest.fixture() +def client(mock_api_caller) -> ApprovalWorkflowClient: + return ApprovalWorkflowClient(mock_api_caller) + + +def test_init_rejects_non_api_caller(): + with pytest.raises(InvalidRequestError, match="ATLAN-PYTHON-400-048.*ApiCaller"): + ApprovalWorkflowClient("not-a-client") # type: ignore[arg-type] + + +def test_get_parses_snake_case_wire(client, mock_api_caller): + """The approval-workflow API family is snake_case on the wire — fields + must parse without camelCase aliasing.""" + mock_api_caller._call_api.return_value = RAW_WF_REQUEST + request = client.get(guid=WF_REQUEST_GUID) + + assert isinstance(request, ApprovalWorkflowRequest) + assert request.approval_workflow_request_type == "DATA_ACCESS" + assert request.request_on_asset_guid == ASSET_GUID + mock_api_caller.reset_mock() + + +@pytest.mark.parametrize( + "method, decision", + [("approve_all", "APPROVED"), ("reject_all", "REJECTED")], +) +def test_bulk_action_body_is_snake_case(client, mock_api_caller, method, decision): + mock_api_caller._call_api.return_value = { + "total_tasks": 3, + "message": "queued", + } + result = getattr(client, method)( + group_key=ASSET_GUID, sub_type=ApprovalWorkflowRequestType.DATA_ACCESS, comment="bulk" + ) + + assert result.total_tasks == 3 + assert result.message == "queued" + endpoint = mock_api_caller._call_api.call_args[0][0] + assert endpoint.path.endswith("/actions/bulk") + body = loads( + mock_api_caller._call_api.call_args.kwargs["request_obj"].json( + by_alias=True, exclude_unset=True + ) + ) + assert body == { + "group_key": ASSET_GUID, + "decision": decision, + "sub_type": "DATA_ACCESS", + "comment": "bulk", + } + mock_api_caller.reset_mock() + + +def test_recipient_scoped_1003_gets_actionable_message(client, mock_api_caller): + """The server's misleading 'No pending tasks found' (1003) is translated + into a message explaining recipient scoping (BLDX-1611).""" + from pyatlan.errors import ErrorCode + + mock_api_caller._call_api.side_effect = ( + ErrorCode.INVALID_REQUEST_PASSTHROUGH.exception_with_parameters( + "1003", "No pending tasks found for the specified group", "" + ) + ) + with pytest.raises(InvalidRequestError, match="recipient-scoped"): + client.approve_all(group_key=ASSET_GUID) + mock_api_caller.reset_mock(side_effect=True) + + +def test_bulk_action_omits_optional_fields(client, mock_api_caller): + """sub_type/comment stay off the wire when not given (exclude_unset).""" + mock_api_caller._call_api.return_value = {"total_tasks": 1, "message": "ok"} + client.approve_all(group_key=ASSET_GUID) + + body = loads( + mock_api_caller._call_api.call_args.kwargs["request_obj"].json( + by_alias=True, exclude_unset=True + ) + ) + assert body == {"group_key": ASSET_GUID, "decision": "APPROVED"} + mock_api_caller.reset_mock() diff --git a/tests/unit/test_requests_client.py b/tests/unit/test_requests_client.py new file mode 100644 index 000000000..22a0fb3af --- /dev/null +++ b/tests/unit/test_requests_client.py @@ -0,0 +1,223 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +from json import loads +from unittest.mock import Mock + +import pytest + +from pyatlan.client.common import ApiCaller +from pyatlan.client.requests import RequestsClient +from pyatlan.errors import InvalidRequestError +from pyatlan.model.atlan_request import ( + AtlanRequest, + AttributeRequest, + build_requests_filter, +) +from pyatlan.model.enums import AtlanRequestStatus, AtlanRequestType + +REQUEST_ID = "070c46dc-734b-4bed-b89f-54ae752ec589" +TERM_GUID = "9c67229e-f345-4de4-b046-c3b6cb2a5c34" + +RAW_REQUEST = { + "id": REQUEST_ID, + "version": "bold-bonus-8934", + "isActive": True, + "createdAt": 1786102423732, + "updatedAt": 1786102423732, + "createdBy": "service-account-example", + "tenantId": "default", + "sourceType": "static", + "destinationGuid": TERM_GUID, + "destinationQualifiedName": "abc@def", + "destinationAttribute": "userDescription", + "destinationValue": "requested value", + "entityType": "AtlasGlossaryTerm", + "requestType": "attribute", + "status": "active", + "approvalType": "single", +} + +RAW_LIST = {"totalRecord": 2, "filterRecord": 1, "records": [RAW_REQUEST]} + + +@pytest.fixture(autouse=True) +def set_env(monkeypatch): + monkeypatch.setenv("ATLAN_BASE_URL", "https://test.atlan.com") + monkeypatch.setenv("ATLAN_API_KEY", "test-api-key") + + +@pytest.fixture() +def mock_api_caller(): + return Mock(spec=ApiCaller) + + +@pytest.fixture() +def client(mock_api_caller) -> RequestsClient: + return RequestsClient(mock_api_caller) + + +def test_init_rejects_non_api_caller(): + with pytest.raises( + InvalidRequestError, match="ATLAN-PYTHON-400-048.*ApiCaller" + ): + RequestsClient("not-a-client") # type: ignore[arg-type] + + +def test_list_with_typed_filter(client, mock_api_caller): + mock_api_caller._call_api.return_value = RAW_LIST + response = client.list(status=AtlanRequestStatus.ACTIVE) + + assert response.total_record == 2 + assert response.filter_record == 1 + assert len(response.records) == 1 + record = response.records[0] + assert record.id == REQUEST_ID + assert record.status == "active" + assert record.destination_attribute == "userDescription" + # the typed filter reached the query params in the UI grammar + query_params = mock_api_caller._call_api.call_args[0][1] + assert loads(query_params["filter"]) == { + "$and": [{"isDuplicate": False}, {"status": {"$in": ["active"]}}] + } + mock_api_caller.reset_mock() + + +def test_actioned_records_with_list_approvers_parse(): + """rejectedBy/approvedBy come back as LISTS for multi-approver requests — + a page containing already-actioned records must parse (regression: the + str-typed fields crashed list() on any tenant with actioned requests).""" + # every shape seen or plausible in the wild must parse + for rejected_by in ( + "admin-one", + ["admin-one", "admin-two"], + [{"username": "admin-one", "timestamp": 1786100000000}], + ): + actioned = { + **RAW_REQUEST, + "status": "rejected", + "rejectedBy": rejected_by, + "approvedBy": [], + } + parsed = AtlanRequest(**actioned) + assert parsed.rejected_by == rejected_by + assert parsed.approved_by == [] + + +def test_filter_builder_combines_with_and(): + """Multiple typed filters combine with AND; enums serialize to their + wire values.""" + built = loads( + build_requests_filter( + status=AtlanRequestStatus.ACTIVE, + request_type=AtlanRequestType.ATLAN_TAG, + ) + ) + assert built == { + "$and": [ + {"isDuplicate": False}, + { + "status": {"$in": ["active"]}, + "requestType": {"$in": ["attach_classification"]}, + }, + ] + } + assert build_requests_filter() is None + + +def test_filter_builder_rejects_raw_plus_typed(): + """The raw escape hatch cannot silently swallow typed filters.""" + with pytest.raises(InvalidRequestError): + build_requests_filter( + status=AtlanRequestStatus.ACTIVE, post_filter='{"x":1}' + ) + + +def test_iteration_pages_lazily(client, mock_api_caller): + """Iterating the response fetches subsequent pages until one is empty + (UserResponse/GroupResponse pagination pattern).""" + second = {"id": "second-id", "requestType": "attribute", "status": "active"} + mock_api_caller._call_api.side_effect = [ + {"totalRecord": 2, "filterRecord": 2, "records": [RAW_REQUEST]}, + {"records": [second]}, + {"records": []}, + ] + response = client.list(status=AtlanRequestStatus.ACTIVE, limit=1) + seen = [r.id for r in response] + + assert seen == [REQUEST_ID, "second-id"] + # three calls: first page + two pagination fetches (second, then empty) + assert mock_api_caller._call_api.call_count == 3 + mock_api_caller.reset_mock(side_effect=True) + + +def test_list_actionable_uses_actionable_route(client, mock_api_caller): + mock_api_caller._call_api.return_value = RAW_LIST + response = client.list_actionable() + + assert response.total_record == 2 + endpoint = mock_api_caller._call_api.call_args[0][0] + assert "actionable" in endpoint.path + mock_api_caller.reset_mock() + + +def test_get_unwraps_single_element_list(client, mock_api_caller): + """The by-id endpoint may wrap the request in a single-element list.""" + mock_api_caller._call_api.return_value = [RAW_REQUEST] + request = client.get(guid=REQUEST_ID) + + assert isinstance(request, AtlanRequest) + assert request.id == REQUEST_ID + mock_api_caller.reset_mock() + + +def test_create_sends_all_wire_required_fields(client, mock_api_caller): + """createRequest requires requestType/approvalType/sourceType/entityType — + the creator must set them explicitly so exclude_unset serialization keeps + them (the BLDX-1589 default-stripping trap).""" + mock_api_caller._call_api.return_value = RAW_REQUEST + request = AttributeRequest.creator( + destination_guid=TERM_GUID, + destination_qualified_name="abc@def", + destination_attribute="userDescription", + destination_value="requested value", + entity_type="AtlasGlossaryTerm", + ) + created = client.create(request) + + assert created and created.id == REQUEST_ID + sent = loads( + mock_api_caller._call_api.call_args.kwargs["request_obj"].json( + by_alias=True, exclude_unset=True + ) + ) + for required in ("requestType", "approvalType", "sourceType", "entityType"): + assert required in sent, f"{required} missing from the wire payload" + assert sent["requestType"] == "attribute" + assert sent["approvalType"] == "single" + mock_api_caller.reset_mock() + + +@pytest.mark.parametrize( + "method, expected_action", + [("approve", "approved"), ("reject", "rejected")], +) +def test_action_posts_expected_body(client, mock_api_caller, method, expected_action): + mock_api_caller._call_api.return_value = "success" + result = getattr(client, method)(guid=REQUEST_ID, message="because") + + assert result is True + endpoint = mock_api_caller._call_api.call_args[0][0] + assert REQUEST_ID in endpoint.path and endpoint.path.endswith("/action") + body = loads( + mock_api_caller._call_api.call_args.kwargs["request_obj"].json( + by_alias=True, exclude_unset=True + ) + ) + assert body == {"action": expected_action, "message": "because"} + mock_api_caller.reset_mock() + + +def test_action_non_success_is_false(client, mock_api_caller): + mock_api_caller._call_api.return_value = {"unexpected": "shape"} + assert client.approve(guid=REQUEST_ID) is False + mock_api_caller.reset_mock()