Adds logic to manage suppressions

docs/index.rst

@@ -28,6 +28,7 @@ Contents:
    django
    tornado
    spamcheck
+   suppression
    webhooks
    testing
    reference

docs/supression.rst

@@ -0,0 +1,47 @@
+.. _supression:
+
+Suppression API
+==============
+
+You can manage Postmark's Suppression lists with a few simple calls:
+
+To view the current suppression list:
+.. code-block:: python
+
+    >>> response = postmark.suppressions.get_suppression(stream_id="test")
+    >>> suppressions = (response['Suppressions'])
+    >>> for suppression in suppressions:
+    >>>    print(suppression.email_address])
+    >>>    print("    " + suppression.suppression_reason])
+    >>>    print("    " + suppression.origin])
+    >>>    print("    " + suppression.created_at])
+    [email protected]
+        ManualSuppression
+        Recipient
+        2019-12-17T08:58:33-05:00
+    [email protected]
+        HardBounce
+        Recipient
+        2019-12-17T08:58:33-05:00
+    [email protected]
+        SpamComplaint
+        Recipient
+        2019-12-17T08:58:33-05:00
+You can search for a particular suppression with  "SuppressionReason", "Origin", "todate", "fromdate", and "EmailAddress" like:
+.. code-block:: python
+
+    >>> response = postmark.suppressions.get_suppression(stream_id="test", EmailAddress="[email protected]")
+    >>> print(response[0].email_address + " " + response[0].suppression_reason)
+    [email protected] ManualSuppression
+You can add a new suppression with:
+.. code-block:: python
+
+    >>> response = postmark.suppressions.add(stream_id="test", emails=["[email protected]"])
+    >>> print(response[0].email_address + " " + response[0].status)
+    [email protected] Suppressed
+You can delete a suppression with:
+.. code-block:: python
+
+    >>> response = postmark.suppressions.delete(stream_id="test", emails=["[email protected]"])
+    >>> print(response[0].email_address + " " + response[0].status)
+    [email protected] Deleted

src/postmarker/core.py

@@ -14,6 +14,7 @@
 from .models.server import ServerManager
 from .models.stats import StatsManager
 from .models.status import StatusManager
+from .models.suppressions import SuppressionManager
 from .models.templates import TemplateManager
 from .models.triggers import TriggersManager
 from .utils import get_args
@@ -37,6 +38,7 @@ class PostmarkClient:
         ServerManager,
         StatsManager,
         StatusManager,
+        SuppressionManager,
         TemplateManager,
         TriggersManager,
     )

src/postmarker/models/suppressions.py

@@ -0,0 +1,79 @@
+"""Supressions
+
+Information on suppression lists"""
+
+from .base import ModelManager, MessageModel
+
+
+class Suppression(MessageModel):
+    """Suppression model."""
+    email_address = None
+    suppression_reason = None
+    origin = None
+    created_at = None
+
+    def __init__(self, email_address=None, suppression_reason=None, origin=None, created_at=None, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.email_address = email_address
+        self.suppression_reason = suppression_reason
+        self.origin = origin
+        self.created_at = created_at
+
+
+class SuppressionResponse(MessageModel):
+    """Suppression request model."""
+    email_address = None
+    status = None
+    message = None
+
+    def __init__(self, email_address=None, status=None, message=None, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.email_address = email_address
+        self.status = status
+        self.message = message
+
+
+class SuppressionManager(ModelManager):
+    name = "suppressions"
+    model = Suppression
+
+    def _manage_suppression(self, endpoint, emails):
+        if type(emails) != list:
+            emails = [emails]
+        data = {'Suppressions': [{'EmailAddress': email} for email in emails]}
+        response = self.call("POST", endpoint=endpoint, data=data)
+        suppression_response_list = []
+        for suppression in response['Suppressions']:
+            suppression_response_list.append(SuppressionResponse(
+                email_address=suppression['EmailAddress'],
+                status=suppression['Status'],
+                message=suppression['Message']
+            ))
+        return suppression_response_list
+
+    def get_suppression(self, stream_id, **kwargs):
+        params = {}
+        for key, value in kwargs.items():
+            if key in ['EmailAddress', 'SuppressionReason', 'Origin', 'todate', 'fromdate']:
+                params[key] = value
+            else:
+                raise ValueError(f"Invalid parameter: {key}")
+        endpoint = f"/message-streams/{stream_id}/suppressions"
+        response = self.call("GET", endpoint=endpoint, params=params)
+        suppression_list = []
+        for suppression in response['Suppressions']:
+            suppression_list.append(Suppression(
+                email_address=suppression['EmailAddress'],
+                suppression_reason=suppression['SuppressionReason'],
+                origin=suppression['Origin'],
+                created_at=suppression['CreatedAt']
+            ))
+        return suppression_list
+
+    def add(self, stream_id, emails):
+        endpoint = f"/message-streams/{stream_id}/suppressions"
+        return self._manage_suppression(endpoint, emails)
+
+    def delete(self, stream_id, emails):
+        endpoint = f"/message-streams/{stream_id}/suppressions/delete"
+        return self._manage_suppression(endpoint, emails)