feat: send triggered emails

codeforlife/mail.py

@@ -5,23 +5,214 @@
 Dotdigital helpers.
 """
 
-import os
+import json
+import logging
 import typing as t
 from dataclasses import dataclass
 
 import requests
+from django.conf import settings
 
+from .types import JsonDict
 
-# pylint: disable-next=unused-argument
-def add_contact(email: str):
-    """Add a new contact to Dotdigital."""
-    # TODO: implement
+
+@dataclass
+class Preference:
+    """The marketing preferences for a Dotdigital contact."""
+
+    @dataclass
+    class Preference:
+        """
+        The preference values to set in the category. Only supply if
+        is_preference is false, and therefore referring to a preference
+        category.
+        """
+
+        id: int
+        is_preference: bool
+        is_opted_in: bool
+
+    id: int
+    is_preference: bool
+    preferences: t.Optional[t.List[Preference]] = None
+    is_opted_in: t.Optional[bool] = None
+
+
+# pylint: disable-next=too-many-arguments
+def add_contact(
+    email: str,
+    opt_in_type: t.Optional[
+        t.Literal["Unknown", "Single", "Double", "VerifiedDouble"]
+    ] = None,
+    email_type: t.Optional[t.Literal["PlainText, Html"]] = None,
+    data_fields: t.Optional[t.Dict[str, str]] = None,
+    consent_fields: t.Optional[t.List[t.Dict[str, str]]] = None,
+    preferences: t.Optional[t.List[Preference]] = None,
+    region: str = "r1",
+    auth: t.Optional[str] = None,
+    timeout: int = 30,
+):
+    # pylint: disable=line-too-long
+    """Add a new contact to Dotdigital.
+
+    https://developer.dotdigital.com/reference/create-contact-with-consent-and-preferences
+
+    Args:
+        email: The email address of the contact.
+        opt_in_type: The opt-in type of the contact.
+        email_type: The email type of the contact.
+        data_fields: Each contact data field is a key-value pair; the key is a string matching the data field name in Dotdigital.
+        consent_fields: The consent fields that apply to the contact.
+        preferences: The marketing preferences to be applied.
+        region: The Dotdigital region id your account belongs to e.g. r1, r2 or r3.
+        auth: The authorization header used to enable API access. If None, the value will be retrieved from the MAIL_AUTH environment variable.
+        timeout: Send timeout to avoid hanging.
+
+    Raises:
+        AssertionError: If failed to add contact.
+    """
+    # pylint: enable=line-too-long
+
+    if auth is None:
+        auth = settings.MAIL_AUTH
+
+    contact: JsonDict = {"email": email.lower()}
+    if opt_in_type is not None:
+        contact["optInType"] = opt_in_type
+    if email_type is not None:
+        contact["emailType"] = email_type
+    if data_fields is not None:
+        contact["dataFields"] = [
+            {"key": key, "value": value} for key, value in data_fields.items()
+        ]
+
+    body: JsonDict = {"contact": contact}
+    if consent_fields is not None:
+        body["consentFields"] = [
+            {
+                "fields": [
+                    {"key": key, "value": value}
+                    for key, value in fields.items()
+                ]
+            }
+            for fields in consent_fields
+        ]
+    if preferences is not None:
+        body["preferences"] = [
+            {
+                "id": preference.id,
+                "isPreference": preference.is_preference,
+                **(
+                    {}
+                    if preference.is_opted_in is None
+                    else {"isOptedIn": preference.is_opted_in}
+                ),
+                **(
+                    {}
+                    if preference.preferences is None
+                    else {
+                        "preferences": [
+                            {
+                                "id": _preference.id,
+                                "isPreference": _preference.is_preference,
+                                "isOptedIn": _preference.is_opted_in,
+                            }
+                            for _preference in preference.preferences
+                        ]
+                    }
+                ),
+            }
+            for preference in preferences
+        ]
+
+    if not settings.MAIL_ENABLED:
+        logging.info(
+            "Added contact to DotDigital:\n%s", json.dumps(body, indent=2)
+        )
+        return
+
+    response = requests.post(
+        # pylint: disable-next=line-too-long
+        url=f"https://{region}-api.dotdigital.com/v2/contacts/with-consent-and-preferences",
+        json=body,
+        headers={
+            "accept": "application/json",
+            "authorization": auth,
+        },
+        timeout=timeout,
+    )
+
+    assert response.ok, (
+        "Failed to add contact."
+        f" Reason: {response.reason}."
+        f" Text: {response.text}."
+    )
 
 
 # pylint: disable-next=unused-argument
-def remove_contact(email: str):
-    """Remove an existing contact from Dotdigital."""
-    # TODO: implement
+def remove_contact(
+    contact_identifier: str,
+    region: str = "r1",
+    auth: t.Optional[str] = None,
+    timeout: int = 30,
+):
+    # pylint: disable=line-too-long
+    """Remove an existing contact from Dotdigital.
+
+    https://developer.dotdigital.com/reference/get-contact
+    https://developer.dotdigital.com/reference/delete-contact
+
+    Args:
+        contact_identifier: Either the contact id or email address of the contact.
+        region: The Dotdigital region id your account belongs to e.g. r1, r2 or r3.
+        auth: The authorization header used to enable API access. If None, the value will be retrieved from the MAIL_AUTH environment variable.
+        timeout: Send timeout to avoid hanging.
+
+    Raises:
+        AssertionError: If failed to get contact.
+        AssertionError: If failed to delete contact.
+    """
+    # pylint: enable=line-too-long
+
+    if not settings.MAIL_ENABLED:
+        logging.info("Removed contact from DotDigital: %s", contact_identifier)
+        return
+
+    if auth is None:
+        auth = settings.MAIL_AUTH
+
+    response = requests.get(
+        # pylint: disable-next=line-too-long
+        url=f"https://{region}-api.dotdigital.com/v2/contacts/{contact_identifier}",
+        headers={
+            "accept": "application/json",
+            "authorization": auth,
+        },
+        timeout=timeout,
+    )
+
+    assert response.ok, (
+        "Failed to get contact."
+        f" Reason: {response.reason}."
+        f" Text: {response.text}."
+    )
+
+    contact_id: int = response.json()["id"]
+
+    response = requests.delete(
+        url=f"https://{region}-api.dotdigital.com/v2/contacts/{contact_id}",
+        headers={
+            "accept": "application/json",
+            "authorization": auth,
+        },
+        timeout=timeout,
+    )
+
+    assert response.ok, (
+        "Failed to delete contact."
+        f" Reason: {response.reason}."
+        f" Text: {response.text}."
+    )
 
 
 @dataclass
@@ -62,7 +253,7 @@
         metadata: The metadata for your email. It can be either a single value or a series of values in a JSON object.
         attachments: A Base64 encoded string. All attachment types are supported. Maximum file size: 15 MB.
         region: The Dotdigital region id your account belongs to e.g. r1, r2 or r3.
-        auth: The authorization header used to enable API access. If None, the value will be retrieved from the DOTDIGITAL_AUTH environment variable.
+        auth: The authorization header used to enable API access. If None, the value will be retrieved from the MAIL_AUTH environment variable.
         timeout: Send timeout to avoid hanging.
 
     Raises:
@@ -71,7 +262,7 @@
     # pylint: enable=line-too-long
 
     if auth is None:
-        auth = os.environ["DOTDIGITAL_AUTH"]
+        auth = settings.MAIL_AUTH
 
     body = {
         "campaignId": campaign_id,
@@ -103,6 +294,13 @@
             for attachment in attachments
         ]
 
+    if not settings.MAIL_ENABLED:
+        logging.info(
+            "Sent a triggered email with DotDigital:\n%s",
+            json.dumps(body, indent=2),
+        )
+        return
+
     response = requests.post(
         url=f"https://{region}-api.dotdigital.com/v2/email/triggered-campaign",
         json=body,

codeforlife/models/signals/__init__.py

@@ -7,9 +7,5 @@
 """
 
 
-from .general import (
-    UpdateFields,
-    assert_update_fields_includes,
-    update_fields_includes,
-)
+from .general import UpdateFields, update_fields_includes
 from .receiver import model_receiver

codeforlife/models/signals/general.py

@@ -19,28 +19,6 @@
         includes: The fields that should be included in the update-fields.
 
     Returns:
-        The fields missing in the update-fields. If update-fields is None, None
-        is returned.
+        A flag designating if the fields are included in the update-fields.
     """
-
-    if update_fields is None:
-        return None
-
-    return includes.difference(update_fields)
-
-
-def assert_update_fields_includes(
-    update_fields: UpdateFields, includes: t.Set[str]
-):
-    """Assert the call to .save() includes the update-fields specified.
-
-    Args:
-        update_fields: The update-fields provided in the call to .save().
-        includes: The fields that should be included in the update-fields.
-    """
-    missing_update_fields = update_fields_includes(update_fields, includes)
-    if missing_update_fields is not None:
-        assert not missing_update_fields, (
-            "Call to .save() did not include the following update-fields: "
-            f"{', '.join(missing_update_fields)}."
-        )
+    return update_fields and includes.issubset(update_fields)

codeforlife/models/signals/post_save.py

@@ -0,0 +1,91 @@
+"""
+© Ocado Group
+Created on 20/06/2024 at 11:46:02(+01:00).
+
+Helpers for module "django.db.models.signals.post_save".
+https://docs.djangoproject.com/en/3.2/ref/signals/#post-save
+"""
+
+import typing as t
+
+from . import general as _
+from .pre_save import PREVIOUS_VALUE_KEY
+
+FieldValue = t.TypeVar("FieldValue")
+
+
+def check_previous_values(
+    instance: _.AnyModel,
+    predicates: t.Dict[str, t.Callable[[t.Any], bool]],
+):
+    # pylint: disable=line-too-long
+    """Check if the previous values are as expected. If the previous value's key
+    is not on the model, this check returns False.
+
+    Args:
+        instance: The current instance.
+        predicates: A predicate for each field. The previous value is passed in as an arg and it should return True if the previous value is as expected.
+
+    Returns:
+        If all the previous values are as expected.
+    """
+    # pylint: enable=line-too-long
+
+    for field, predicate in predicates.items():
+        previous_value_key = PREVIOUS_VALUE_KEY.format(field=field)
+
+        if not hasattr(instance, previous_value_key) or not predicate(
+            getattr(instance, previous_value_key)
+        ):
+            return False
+
+    return True
+
+
+def previous_values_are_unequal(instance: _.AnyModel, fields: t.Set[str]):
+    # pylint: disable=line-too-long
+    """Check if all the previous values are not equal to the current values. If
+    the previous value's key is not on the model, this check returns False.
+
+    Args:
+        instance: The current instance.
+        fields: The fields that should not be equal.
+
+    Returns:
+        If all the previous values are not equal to the current values.
+    """
+    # pylint: enable=line-too-long
+
+    for field in fields:
+        previous_value_key = PREVIOUS_VALUE_KEY.format(field=field)
+
+        if not hasattr(instance, previous_value_key) or (
+            getattr(instance, field) == getattr(instance, previous_value_key)
+        ):
+            return False
+
+    return True
+
+
+def get_previous_value(
+    instance: _.AnyModel, field: str, cls: t.Type[FieldValue]
+):
+    # pylint: disable=line-too-long
+    """Get a previous value from the instance and assert the value is of the
+    expected type.
+
+    Args:
+        instance: The current instance.
+        field: The field to get the previous value for.
+        cls: The expected type of the value.
+
+    Returns:
+        The previous value of the field.
+    """
+    # pylint: enable=line-too-long
+
+    previous_value = getattr(instance, PREVIOUS_VALUE_KEY.format(field=field))
+
+    assert isinstance(previous_value, cls)
+
+    return previous_value