Add data removal api endpoints

Author: gregsvo

examples/data_removals/create_data_removal.py

@@ -0,0 +1,23 @@
+import asyncio
+import os
+
+from dotenv import load_dotenv
+
+import postmark
+
+load_dotenv()
+account = postmark.AccountClient(os.environ["POSTMARK_ACCOUNT_TOKEN"])
+
+
+async def main():
+    result = await account.data_removals.create(
+        requested_by="admin@example.com",
+        requested_for="user@example.com",
+        notify_when_completed=True,
+    )
+
+    print(f"ID:     {result.id}")
+    print(f"Status: {result.status}")
+
+
+asyncio.run(main())

examples/data_removals/get_data_removal.py

@@ -0,0 +1,19 @@
+import asyncio
+import os
+
+from dotenv import load_dotenv
+
+import postmark
+
+load_dotenv()
+account = postmark.AccountClient(os.environ["POSTMARK_ACCOUNT_TOKEN"])
+
+
+async def main():
+    result = await account.data_removals.get(42)
+
+    print(f"ID:     {result.id}")
+    print(f"Status: {result.status}")
+
+
+asyncio.run(main())

postmark/clients/account_client.py

@@ -4,6 +4,7 @@
 
 import httpx
 
+from postmark.models.data_removals import DataRemovalManager
 from postmark.models.domains import DomainManager
 from postmark.models.servers import AccountServerManager
 from postmark.models.signatures import SenderSignatureManager
@@ -42,6 +43,7 @@ def __init__(self, account_token: str):
         self.server = AccountServerManager(self)
         self.domain = DomainManager(self)
         self.signature = SenderSignatureManager(self)
+        self.data_removals = DataRemovalManager(self)
 
     async def request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
         """

postmark/models/data_removals/__init__.py

@@ -0,0 +1,4 @@
+from .manager import DataRemovalManager
+from .schemas import DataRemoval
+
+__all__ = ["DataRemoval", "DataRemovalManager"]

postmark/models/data_removals/manager.py

@@ -0,0 +1,28 @@
+from postmark.utils.types import HTTPClient
+
+from .schemas import DataRemoval
+
+
+class DataRemovalManager:
+    def __init__(self, client: HTTPClient):
+        self.client = client
+
+    async def create(
+        self,
+        requested_by: str,
+        requested_for: str,
+        notify_when_completed: bool = False,
+    ) -> DataRemoval:
+        response = await self.client.post(
+            "/data-removals",
+            json={
+                "RequestedBy": requested_by,
+                "RequestedFor": requested_for,
+                "NotifyWhenCompleted": notify_when_completed,
+            },
+        )
+        return DataRemoval(**response.json())
+
+    async def get(self, removal_id: int) -> DataRemoval:
+        response = await self.client.get(f"/data-removals/{removal_id}")
+        return DataRemoval(**response.json())

postmark/models/data_removals/schemas.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class DataRemoval(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: int = Field(alias="ID")
+    status: str = Field(alias="Status")

tests/conftest.py

@@ -6,6 +6,7 @@
 from httpx import Response
 
 from postmark.models.bounces import BounceManager
+from postmark.models.data_removals import DataRemovalManager
 from postmark.models.domains import DomainManager
 from postmark.models.inbound import InboundManager
 from postmark.models.inbound_rules import InboundRuleManager
@@ -128,3 +129,8 @@ def webhooks(fake_client):
 @pytest.fixture
 def suppressions(fake_client):
     return SuppressionManager(fake_client), fake_client
+
+
+@pytest.fixture
+def data_removals(fake_client):
+    return DataRemovalManager(fake_client), fake_client

tests/test_data_removals.py

@@ -0,0 +1,89 @@
+"""Tests for DataRemovalManager."""
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Shared fixture data
+# ---------------------------------------------------------------------------
+
+REMOVAL = {"ID": 42, "Status": "Pending"}
+REMOVAL_DONE = {"ID": 42, "Status": "Done"}
+
+
+# ---------------------------------------------------------------------------
+# Create data removal
+# ---------------------------------------------------------------------------
+
+
+class TestCreateDataRemoval:
+    @pytest.mark.asyncio
+    async def test_create_success(self, data_removals):
+        manager, fake = data_removals
+        fake.mock_post_response(REMOVAL)
+
+        result = await manager.create(
+            requested_by="admin@example.com",
+            requested_for="user@example.com",
+        )
+
+        assert result.id == 42
+        assert result.status == "Pending"
+
+    @pytest.mark.asyncio
+    async def test_create_calls_correct_endpoint(self, data_removals):
+        manager, fake = data_removals
+        fake.mock_post_response(REMOVAL)
+
+        await manager.create(
+            requested_by="admin@example.com",
+            requested_for="user@example.com",
+        )
+
+        fake.post.assert_called_once_with(
+            "/data-removals",
+            json={
+                "RequestedBy": "admin@example.com",
+                "RequestedFor": "user@example.com",
+                "NotifyWhenCompleted": False,
+            },
+        )
+
+    @pytest.mark.asyncio
+    async def test_create_with_notify(self, data_removals):
+        manager, fake = data_removals
+        fake.mock_post_response(REMOVAL)
+
+        await manager.create(
+            requested_by="admin@example.com",
+            requested_for="user@example.com",
+            notify_when_completed=True,
+        )
+
+        body = fake.post.call_args[1]["json"]
+        assert body["NotifyWhenCompleted"] is True
+
+
+# ---------------------------------------------------------------------------
+# Get data removal
+# ---------------------------------------------------------------------------
+
+
+class TestGetDataRemoval:
+    @pytest.mark.asyncio
+    async def test_get_success(self, data_removals):
+        manager, fake = data_removals
+        fake.mock_get_response(REMOVAL_DONE)
+
+        result = await manager.get(42)
+
+        assert result.id == 42
+        assert result.status == "Done"
+
+    @pytest.mark.asyncio
+    async def test_get_calls_correct_endpoint(self, data_removals):
+        manager, fake = data_removals
+        fake.mock_get_response(REMOVAL)
+
+        await manager.get(42)
+
+        fake.get.assert_called_once_with("/data-removals/42")