✨(marketing) implement brevo backend

Brevo is the first real backend we have to implement. The init method is
reposible to configure it.

src/lasuite/marketing/backends/brevo.py

@@ -0,0 +1,99 @@
+"""Brevo marketing automation integration."""
+
+import logging
+from urllib.parse import quote_plus
+
+import requests
+
+from lasuite.marketing.backends import ContactData
+from lasuite.marketing.exceptions import ContactCreationError
+
+from .base import BaseBackend
+
+logger = logging.getLogger(__name__)
+
+
+class BrevoBackend(BaseBackend):
+    """
+    Brevo marketing automation integration.
+
+    Handles:
+    - Contact management and segmentation
+    - Marketing campaigns and automation
+    - Email communications
+    """
+
+    def __init__(self, api_key: str, api_contact_list_ids: list[int], api_contact_attributes: dict | None = None):
+        """Configure the Brevo backend."""
+        self._api_key = api_key
+        self.api_contact_attributes = api_contact_attributes or {}
+        self.api_contact_list_ids = api_contact_list_ids
+
+    def create_or_update_contact(self, contact_data: ContactData, timeout: int = None) -> dict:
+        """
+        Create or update a Brevo contact.
+
+        Args:
+            contact_data: Contact information and attributes
+            timeout: API request timeout in seconds
+
+        Returns:
+            dict: Brevo API response
+
+        Raises:
+            ContactCreationError: If contact creation fails
+            ImproperlyConfigured: If required settings are missing
+
+        Note:
+            Contact attributes must be pre-configured in Brevo.
+            Changes to attributes can impact existing workflows.
+
+        """
+        # First try to retrieve the contact by email
+        try:
+            email = quote_plus(contact_data.email)
+            url = f"https://api.brevo.com/v3/contacts/{email}"
+            response = requests.get(
+                url, params={"identifierType": "email_id"}, headers={"api-key": self._api_key}, timeout=timeout or 10
+            )
+            response.raise_for_status()
+            contact = response.json()
+        except requests.RequestException:
+            pass
+        else:
+            # Add the list_ids from the contact in the contact_data
+            list_ids = contact.get("listIds", [])
+            contact_data.list_ids = (contact_data.list_ids or []) + list_ids
+
+        attributes = {
+            **self.api_contact_attributes,
+            **(contact_data.attributes or {}),
+        }
+
+        # Use a set to avoid duplicates
+        list_ids = set((contact_data.list_ids or []) + self.api_contact_list_ids)
+
+        payload = {
+            "email": contact_data.email,
+            "updateEnabled": contact_data.update_enabled,
+            "listIds": list(list_ids),
+            "attributes": attributes,
+        }
+
+        print(payload)
+
+        try:
+            response = requests.post(
+                "https://api.brevo.com/v3/contacts",
+                json=payload,
+                headers={"api-key": self._api_key},
+                timeout=timeout or 10,
+            )
+            response.raise_for_status()
+        except requests.RequestException as err:
+            raise ContactCreationError("Failed to create contact in Brevo") from err
+
+        if response.status_code == requests.codes.created:
+            return response.json()
+
+        return {}

tests/marketing/backends/__init__.py

@@ -0,0 +1 @@
+"""Marketing backends module."""

tests/marketing/backends/test_brevo_backend.py

@@ -0,0 +1,146 @@
+"""Test the Brevo marketing backend."""
+
+import pytest
+import responses
+from responses import matchers
+
+from lasuite.marketing.backends import ContactData
+from lasuite.marketing.backends.brevo import BrevoBackend
+from lasuite.marketing.exceptions import ContactCreationError
+
+
+@responses.activate
+def test_create_contact_success_without_existing_brevo_contact():
+    """Test successful contact creation."""
+    responses.add(
+        responses.GET,
+        "https://api.brevo.com/v3/contacts/test%40example.com?identifierType=email_id",
+        status=404,
+    )
+
+    responses.add(
+        responses.POST,
+        "https://api.brevo.com/v3/contacts",
+        headers={"api-key": "test-api-key"},
+        json={
+            "id": "test-id",
+        },
+        status=201,
+        match=[
+            matchers.json_params_matcher(
+                {
+                    "email": "test@example.com",
+                    "updateEnabled": True,
+                    "listIds": [1, 2, 3],
+                    "attributes": {"source": "test", "first_name": "Test"},
+                }
+            )
+        ],
+    )
+
+    valid_contact_data = ContactData(
+        email="test@example.com",
+        attributes={"first_name": "Test"},
+        list_ids=[1, 2],
+        update_enabled=True,
+    )
+
+    brevo_service = BrevoBackend(
+        api_key="test-api-key",
+        api_contact_list_ids=[1, 2, 3],
+        api_contact_attributes={"source": "test"},
+    )
+
+    response = brevo_service.create_or_update_contact(valid_contact_data)
+
+    assert response == {"id": "test-id"}
+
+
+@responses.activate
+def test_create_contact_success_with_existing_brevo_contact():
+    """Test successful contact creation."""
+    responses.add(
+        responses.GET,
+        "https://api.brevo.com/v3/contacts/test%40example.com?identifierType=email_id",
+        status=200,
+        json={"id": "test-id", "listIds": [4, 5]},
+    )
+
+    responses.add(
+        responses.POST,
+        "https://api.brevo.com/v3/contacts",
+        headers={"api-key": "test-api-key"},
+        json={"id": "test-id"},
+        status=201,
+        match=[
+            matchers.json_params_matcher(
+                {
+                    "email": "test@example.com",
+                    "updateEnabled": True,
+                    "listIds": [1, 2, 3, 4, 5],
+                    "attributes": {"source": "test", "first_name": "Test"},
+                }
+            )
+        ],
+    )
+
+    valid_contact_data = ContactData(
+        email="test@example.com",
+        attributes={"first_name": "Test"},
+        list_ids=[1, 2],
+        update_enabled=True,
+    )
+
+    brevo_service = BrevoBackend(
+        api_key="test-api-key",
+        api_contact_list_ids=[1, 2, 3],
+        api_contact_attributes={"source": "test"},
+    )
+
+    response = brevo_service.create_or_update_contact(valid_contact_data)
+
+    assert response == {"id": "test-id"}
+
+
+@responses.activate
+def test_create_contact_api_error():
+    """Test contact creation API error handling."""
+    responses.add(
+        responses.GET,
+        "https://api.brevo.com/v3/contacts/test%40example.com?identifierType=email_id",
+        status=404,
+    )
+
+    responses.add(
+        responses.POST,
+        "https://api.brevo.com/v3/contacts",
+        headers={"api-key": "test-api-key"},
+        json={"id": "test-id"},
+        status=400,
+        match=[
+            matchers.json_params_matcher(
+                {
+                    "email": "test@example.com",
+                    "updateEnabled": True,
+                    "listIds": [1, 2, 3],
+                    "attributes": {"source": "test", "first_name": "Test"},
+                }
+            )
+        ],
+    )
+
+    valid_contact_data = ContactData(
+        email="test@example.com",
+        attributes={"first_name": "Test"},
+        list_ids=[1, 2],
+        update_enabled=True,
+    )
+
+    brevo_service = BrevoBackend(
+        api_key="test-api-key",
+        api_contact_list_ids=[1, 2, 3],
+        api_contact_attributes={"source": "test"},
+    )
+
+    with pytest.raises(ContactCreationError, match="Failed to create contact in Brevo"):
+        brevo_service.create_or_update_contact(valid_contact_data)