feat: add contract delete SDK methods (DQ-665)

pyatlan/client/aio/client.py

@@ -16,7 +16,7 @@
 from contextlib import _AsyncGeneratorContextManager
 from http import HTTPStatus
 from types import SimpleNamespace
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
 import httpx
 from httpx_retries.retry import Retry
@@ -501,6 +501,7 @@ async def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response=False,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         """
         Async version of _call_api - mirrors sync client structure.
@@ -509,6 +510,8 @@ async def _call_api(
         params = await self._create_params(
             api, query_params, request_obj, exclude_unset
         )
+        if extra_headers:
+            params["headers"].update(extra_headers)
         if LOGGER.isEnabledFor(logging.DEBUG):
             self._api_logger(api, path)
         return await self._call_api_internal(

pyatlan/client/aio/contract.py

@@ -7,9 +7,15 @@
 from pydantic.v1 import validate_arguments
 
 from pyatlan.client.common import AsyncApiCaller, ContractInit
-from pyatlan.client.constants import CONTRACT_INIT_API
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan.model.assets import Asset
+from pyatlan.model.enums import AtlanDeleteType
+from pyatlan.model.response import AssetMutationResponse
 
 
 class AsyncContractClient:
@@ -49,3 +55,47 @@ async def generate_initial_spec(
 
         # Process response using shared logic
         return ContractInit.process_response(response)
+
+    @validate_arguments
+    async def delete(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) a data contract and all its versions (async version).
+        This deletes every version of the contract associated with the same asset
+        and cleans up the asset's contract attributes (hasContract, dataContractLatest,
+        dataContractLatestCertified).
+
+        :param guid: unique identifier (GUID) of any version of the contract to delete
+        :returns: details of the deleted contract version(s)
+        :raises AtlanError: on any API communication issue
+
+        .. warning::
+            This is an irreversible operation. All versions of the contract will be permanently removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = await self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS, query_params=query_params
+        )
+        return AssetMutationResponse(**raw_json)
+
+    @validate_arguments
+    async def delete_latest_version(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) only the latest version of a data contract (async version).
+        The previous version (if any) becomes the new latest, and the asset's
+        contract pointers are updated accordingly.
+
+        :param guid: unique identifier (GUID) of the latest contract version to delete
+        :returns: details of the deleted contract version
+        :raises AtlanError: on any API communication issue
+        :raises ApiError: if the specified GUID is not the latest version
+
+        .. warning::
+            This is an irreversible operation. Only the latest version will be removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = await self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS,
+            query_params=query_params,
+            extra_headers={CONTRACT_DELETE_SCOPE_HEADER: "single"},
+        )
+        return AssetMutationResponse(**raw_json)

pyatlan/client/atlan.py

@@ -792,9 +792,12 @@ def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response=False,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         path = self._create_path(api)
         params = self._create_params(api, query_params, request_obj, exclude_unset)
+        if extra_headers:
+            params["headers"].update(extra_headers)
         if LOGGER.isEnabledFor(logging.DEBUG):
             self._api_logger(api, path)
         return self._call_api_internal(api, path, params, text_response=text_response)

pyatlan/client/constants.py

@@ -704,3 +704,5 @@
     HTTPStatus.ACCEPTED,
     endpoint=EndPoint.CHRONOS,
 )
+
+CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"

pyatlan/client/contract.py

@@ -5,9 +5,15 @@
 from pydantic.v1 import validate_arguments
 
 from pyatlan.client.common import ApiCaller, ContractInit
-from pyatlan.client.constants import CONTRACT_INIT_API
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan.model.assets import Asset
+from pyatlan.model.enums import AtlanDeleteType
+from pyatlan.model.response import AssetMutationResponse
 
 
 class ContractClient:
@@ -44,3 +50,47 @@ def generate_initial_spec(
 
         # Process response using shared logic
         return ContractInit.process_response(response)
+
+    @validate_arguments
+    def delete(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) a data contract and all its versions.
+        This deletes every version of the contract associated with the same asset
+        and cleans up the asset's contract attributes (hasContract, dataContractLatest,
+        dataContractLatestCertified).
+
+        :param guid: unique identifier (GUID) of any version of the contract to delete
+        :returns: details of the deleted contract version(s)
+        :raises AtlanError: on any API communication issue
+
+        .. warning::
+            This is an irreversible operation. All versions of the contract will be permanently removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS, query_params=query_params
+        )
+        return AssetMutationResponse(**raw_json)
+
+    @validate_arguments
+    def delete_latest_version(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) only the latest version of a data contract.
+        The previous version (if any) becomes the new latest, and the asset's
+        contract pointers are updated accordingly.
+
+        :param guid: unique identifier (GUID) of the latest contract version to delete
+        :returns: details of the deleted contract version
+        :raises AtlanError: on any API communication issue
+        :raises ApiError: if the specified GUID is not the latest version
+
+        .. warning::
+            This is an irreversible operation. Only the latest version will be removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS,
+            query_params=query_params,
+            extra_headers={CONTRACT_DELETE_SCOPE_HEADER: "single"},
+        )
+        return AssetMutationResponse(**raw_json)

pyatlan/client/protocol.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from contextlib import _AsyncGeneratorContextManager, _GeneratorContextManager
-from typing import Any, Protocol, runtime_checkable
+from typing import Any, Dict, Optional, Protocol, runtime_checkable
 
 from httpx_retries import Retry
 
@@ -26,6 +26,7 @@ def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response: bool = False,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         pass
 
@@ -56,6 +57,7 @@ async def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response: bool = False,
+        extra_headers: Optional[Dict[str, str]] = None,
     ) -> Any:
         pass
 

pyatlan_v9/client/aio/atlan.py

@@ -418,9 +418,12 @@ async def _call_api(
         query_params=None,
         request_obj=None,
         text_response=False,
+        extra_headers=None,
     ):
         path = self._create_path(api)
         params = await self._create_params(api, query_params, request_obj)
+        if extra_headers:
+            params["headers"].update(extra_headers)
         if LOGGER.isEnabledFor(logging.DEBUG):
             self._api_logger(api, path)
         return await self._call_api_internal(

pyatlan_v9/client/aio/contract.py

@@ -5,10 +5,17 @@
 from typing import Optional
 
 from pyatlan.client.common import AsyncApiCaller
-from pyatlan.client.constants import CONTRACT_INIT_API
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
+from pyatlan_v9.client.asset import _parse_mutation_response
 from pyatlan_v9.model.assets import Asset
 from pyatlan_v9.model.contract import InitRequest
+from pyatlan_v9.model.enums import AtlanDeleteType
+from pyatlan_v9.model.response import AssetMutationResponse
 from pyatlan_v9.validate import validate_arguments
 
 
@@ -48,3 +55,47 @@ async def generate_initial_spec(
             CONTRACT_INIT_API, request_obj=request_obj
         )
         return response.get("contract")
+
+    @validate_arguments
+    async def delete(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) a data contract and all its versions (async version).
+        This deletes every version of the contract associated with the same asset
+        and cleans up the asset's contract attributes (hasContract, dataContractLatest,
+        dataContractLatestCertified).
+
+        :param guid: unique identifier (GUID) of any version of the contract to delete
+        :returns: details of the deleted contract version(s)
+        :raises AtlanError: on any API communication issue
+
+        .. warning::
+            This is an irreversible operation. All versions of the contract will be permanently removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = await self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS, query_params=query_params
+        )
+        return _parse_mutation_response(raw_json)
+
+    @validate_arguments
+    async def delete_latest_version(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) only the latest version of a data contract (async version).
+        The previous version (if any) becomes the new latest, and the asset's
+        contract pointers are updated accordingly.
+
+        :param guid: unique identifier (GUID) of the latest contract version to delete
+        :returns: details of the deleted contract version
+        :raises AtlanError: on any API communication issue
+        :raises ApiError: if the specified GUID is not the latest version
+
+        .. warning::
+            This is an irreversible operation. Only the latest version will be removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = await self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS,
+            query_params=query_params,
+            extra_headers={CONTRACT_DELETE_SCOPE_HEADER: "single"},
+        )
+        return _parse_mutation_response(raw_json)

pyatlan_v9/client/atlan.py

@@ -650,9 +650,12 @@ def _call_api(
         query_params=None,
         request_obj=None,
         text_response=False,
+        extra_headers=None,
     ):
         path = self._create_path(api)
         params = self._create_params(api, query_params, request_obj)
+        if extra_headers:
+            params["headers"].update(extra_headers)
         if LOGGER.isEnabledFor(logging.DEBUG):
             self._api_logger(api, path)
         return self._call_api_internal(api, path, params, text_response=text_response)

pyatlan_v9/client/contract.py

@@ -3,10 +3,17 @@
 from typing import Optional
 
 from pyatlan.client.common import ApiCaller
-from pyatlan.client.constants import CONTRACT_INIT_API
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
+from pyatlan_v9.client.asset import _parse_mutation_response
 from pyatlan_v9.model.assets import Asset
 from pyatlan_v9.model.contract import InitRequest
+from pyatlan_v9.model.enums import AtlanDeleteType
+from pyatlan_v9.model.response import AssetMutationResponse
 from pyatlan_v9.validate import validate_arguments
 
 
@@ -42,3 +49,47 @@ def generate_initial_spec(
         )
         response = self._client._call_api(CONTRACT_INIT_API, request_obj=request_obj)
         return response.get("contract")
+
+    @validate_arguments
+    def delete(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) a data contract and all its versions.
+        This deletes every version of the contract associated with the same asset
+        and cleans up the asset's contract attributes (hasContract, dataContractLatest,
+        dataContractLatestCertified).
+
+        :param guid: unique identifier (GUID) of any version of the contract to delete
+        :returns: details of the deleted contract version(s)
+        :raises AtlanError: on any API communication issue
+
+        .. warning::
+            This is an irreversible operation. All versions of the contract will be permanently removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS, query_params=query_params
+        )
+        return _parse_mutation_response(raw_json)
+
+    @validate_arguments
+    def delete_latest_version(self, guid: str) -> AssetMutationResponse:
+        """
+        Hard-delete (purge) only the latest version of a data contract.
+        The previous version (if any) becomes the new latest, and the asset's
+        contract pointers are updated accordingly.
+
+        :param guid: unique identifier (GUID) of the latest contract version to delete
+        :returns: details of the deleted contract version
+        :raises AtlanError: on any API communication issue
+        :raises ApiError: if the specified GUID is not the latest version
+
+        .. warning::
+            This is an irreversible operation. Only the latest version will be removed.
+        """
+        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": [guid]}
+        raw_json = self._client._call_api(
+            DELETE_ENTITIES_BY_GUIDS,
+            query_params=query_params,
+            extra_headers={CONTRACT_DELETE_SCOPE_HEADER: "single"},
+        )
+        return _parse_mutation_response(raw_json)