Merge pull request #483 from alphagov/ris-mailchimp-list-methods

DMMailChimpClient: add get_lists_for_email(...) and permanently_remove_email_from_list(...) methods

dmutils/__init__.py

@@ -2,4 +2,4 @@
 from .flask_init import init_app, init_manager
 
 
-__version__ = '45.2.0'
+__version__ = '45.3.0'

dmutils/email/dm_mailchimp.py

@@ -3,6 +3,9 @@
 
 from json.decoder import JSONDecodeError
 from hashlib import md5
+from logging import Logger
+from typing import Callable, Iterator, Mapping, Sequence, Union
+
 from requests.exceptions import RequestException, HTTPError
 
 from mailchimp3 import MailChimp
@@ -23,22 +26,22 @@ class DMMailChimpClient(object):
 
     def __init__(
         self,
-        mailchimp_username,
-        mailchimp_api_key,
-        logger,
-        retries=0
+        mailchimp_username: str,
+        mailchimp_api_key: str,
+        logger: Logger,
+        retries: int=0,
     ):
-        self._client = MailChimp(mc_user=mailchimp_username, mc_secret=mailchimp_api_key, timeout=25)
+        self._client = MailChimp(mc_user=mailchimp_username, mc_api=mailchimp_api_key, timeout=25)
         self.logger = logger
         self.retries = retries
 
     @staticmethod
-    def get_email_hash(email_address):
+    def get_email_hash(email_address: Union[str, bytes]) -> bytes:
         """md5 hashing of lower cased emails has been chosen by mailchimp to identify email addresses"""
         formatted_email_address = str(email_address.lower()).encode('utf-8')
         return md5(formatted_email_address).hexdigest()
 
-    def timeout_retry(self, method):
+    def timeout_retry(self, method: Callable) -> Callable:
         def wrapper(*args, **kwargs):
             for i in range(1 + self.retries):
                 try:
@@ -53,7 +56,7 @@ def wrapper(*args, **kwargs):
 
         return wrapper
 
-    def create_campaign(self, campaign_data):
+    def create_campaign(self, campaign_data: Mapping) -> Union[str, bool]:
         try:
             with log_external_request(service='Mailchimp'):
                 campaign = self._client.campaigns.create(campaign_data)
@@ -70,7 +73,7 @@ def create_campaign(self, campaign_data):
             )
         return False
 
-    def set_campaign_content(self, campaign_id, content_data):
+    def set_campaign_content(self, campaign_id: str, content_data: Mapping):
         try:
             with log_external_request(service='Mailchimp'):
                 return self._client.campaigns.content.update(campaign_id, content_data)
@@ -84,7 +87,7 @@ def set_campaign_content(self, campaign_id, content_data):
             )
         return False
 
-    def send_campaign(self, campaign_id):
+    def send_campaign(self, campaign_id: str):
         try:
             with log_external_request(service='Mailchimp'):
                 self._client.campaigns.actions.send(campaign_id)
@@ -99,7 +102,7 @@ def send_campaign(self, campaign_id):
             )
         return False
 
-    def subscribe_new_email_to_list(self, list_id, email_address):
+    def subscribe_new_email_to_list(self, list_id: str, email_address: str):
         """Will subscribe email address to list if they do not already exist in that list else do nothing"""
         hashed_email = self.get_email_hash(email_address)
         try:
@@ -142,15 +145,15 @@ def subscribe_new_email_to_list(self, list_id, email_address):
             )
             return False
 
-    def subscribe_new_emails_to_list(self, list_id, email_addresses):
+    def subscribe_new_emails_to_list(self, list_id: str, email_addresses: str) -> bool:
         success = True
         for email_address in email_addresses:
             with log_external_request(service='Mailchimp'):
                 if not self.subscribe_new_email_to_list(list_id, email_address):
                     success = False
         return success
 
-    def get_email_addresses_from_list(self, list_id, pagination_size=100):
+    def get_email_addresses_from_list(self, list_id: str, pagination_size: int=100) -> Iterator[str]:
         offset = 0
         while True:
             member_data = self.timeout_retry(
@@ -161,3 +164,35 @@ def get_email_addresses_from_list(self, list_id, pagination_size=100):
             offset += pagination_size
 
             yield from [member['email_address'] for member in member_data['members']]
+
+    def get_lists_for_email(self, email_address: str) -> Sequence[Mapping]:
+        """
+            Returns a sequence of all lists the email_address has an association with (note: even if that association is
+            "unsubscribed" or "cleaned").
+        """
+        with log_external_request(service='Mailchimp'):
+            return tuple(
+                {
+                    "list_id": mailing_list["id"],
+                    "name": mailing_list["name"],
+                } for mailing_list in self._client.lists.all(get_all=True, email=email_address)["lists"]
+            )
+
+    def permanently_remove_email_from_list(self, email_address: str, list_id: str) -> bool:
+        """
+            Permanently (very permanently) erases all trace of an email address from a given list
+        """
+        hashed_email = self.get_email_hash(email_address)
+        try:
+            with log_external_request(service='Mailchimp'):
+                self._client.lists.members.delete_permanent(
+                    list_id=list_id,
+                    subscriber_hash=hashed_email,
+                )
+            return True
+        except RequestException as e:
+            self.logger.error(
+                f"Mailchimp failed to permanently remove user ({hashed_email}) from list ({list_id})",
+                extra={"error": str(e), "mailchimp_response": get_response_from_request_exception(e)},
+            )
+        return False

setup.py

@@ -31,7 +31,7 @@
          'botocore<1.11.0',
          'contextlib2>=0.4.0',
          'cryptography<2.4,>=2.3',
-         'mailchimp3<2.1.0,>=2.0.11',
+         'mailchimp3==3.0.6',
          'fleep<1.1,>=1.0.1',
          'mandrill>=1.0.57',
          'notifications-python-client<6.0.0,>=5.0.1',

tests/email/test_dm_mailchimp.py

@@ -364,3 +364,74 @@ def test_offset_increments_until_no_members(self):
                     mock.call('a_list_id', count=100, offset=600),
                     mock.call('a_list_id', count=100, offset=700),
                 ]
+
+    def test_get_lists_for_email(self):
+        dm_mailchimp_client = DMMailChimpClient('username', DUMMY_MAILCHIMP_API_KEY, logging.getLogger('mailchimp'))
+        with mock.patch.object(dm_mailchimp_client._client.lists, 'all', autospec=True) as all_lists:
+            all_lists.return_value = {
+                "lists": [
+                    {"id": "gadz00ks", "name": "Pistachios", "irrelevant": "custard"},
+                    {"id": "1886", "name": "Square the circle", "meaningless": 3.1415},
+                ],
+                "pigeon": "pasty",
+            }
+
+            with assert_external_service_log_entry(extra_modules=['mailchimp'], count=1):
+                result = dm_mailchimp_client.get_lists_for_email("[email protected]")
+
+            assert tuple(result) == (
+                {"list_id": "gadz00ks", "name": "Pistachios"},
+                {"list_id": "1886", "name": "Square the circle"},
+            )
+
+            assert all_lists.call_args_list == [
+                mock.call(get_all=True, email="[email protected]"),
+            ]
+
+    def test_permanently_remove_email_from_list_success(self):
+        dm_mailchimp_client = DMMailChimpClient('username', DUMMY_MAILCHIMP_API_KEY, logging.getLogger('mailchimp'))
+        with mock.patch.object(
+            dm_mailchimp_client._client.lists.members,
+            'delete_permanent',
+            autospec=True,
+        ) as del_perm:
+            del_perm.return_value = {"don't rely": "on me"}
+
+            with assert_external_service_log_entry(extra_modules=['mailchimp'], count=1):
+                result = dm_mailchimp_client.permanently_remove_email_from_list(
+                    "[email protected]",
+                    "modern_society",
+                )
+
+            assert result is True
+
+            assert del_perm.call_args_list == [
+                mock.call(
+                    list_id="modern_society",
+                    subscriber_hash="ee5ae5f54bdf3394d48ea4e79e6d0e39",
+                ),
+            ]
+
+    def test_permanently_remove_email_from_list_failure(self):
+        dm_mailchimp_client = DMMailChimpClient('username', DUMMY_MAILCHIMP_API_KEY, logging.getLogger('mailchimp'))
+        with mock.patch.object(
+            dm_mailchimp_client._client.lists.members,
+            'delete_permanent',
+            autospec=True,
+        ) as del_perm:
+            del_perm.side_effect = RequestException("No thoroughfare")
+
+            with assert_external_service_log_entry(successful_call=False, extra_modules=['mailchimp'], count=1):
+                result = dm_mailchimp_client.permanently_remove_email_from_list(
+                    "[email protected]",
+                    "p_kellys_budget",
+                )
+
+            assert result is False
+
+            assert del_perm.call_args_list == [
+                mock.call(
+                    list_id="p_kellys_budget",
+                    subscriber_hash="ee5ae5f54bdf3394d48ea4e79e6d0e39",
+                ),
+            ]