From 59cc2ae3158219e83c4f45fd56a5201b530bb940 Mon Sep 17 00:00:00 2001 From: brejesh4318 Date: Sat, 24 Oct 2026 23:14:31 +0530 Subject: [PATCH] Add Brevo handler Read-only handler for Brevo (formerly Sendinblue) exposing contacts, lists, campaigns (with delivery statistics) and contact_events tables via the Brevo v3 REST API. Includes rate-limit backoff, filter pushdown and tests. Co-Authored-By: Claude Sonnet 5.5 --- .../handlers/brevo_handler/README.md | 122 ++++ .../handlers/brevo_handler/__about__.py | 9 + .../handlers/brevo_handler/__init__.py | 33 + .../handlers/brevo_handler/brevo_handler.py | 83 +++ .../handlers/brevo_handler/brevo_tables.py | 642 ++++++++++++++++++ .../handlers/brevo_handler/connection_args.py | 15 + .../handlers/brevo_handler/icon.svg | 4 + .../handlers/brevo_handler/requirements.txt | 1 + .../brevo_handler/tests/test_brevo_handler.py | 185 +++++ 9 files changed, 1094 insertions(+) create mode 100644 mindsdb/integrations/handlers/brevo_handler/README.md create mode 100644 mindsdb/integrations/handlers/brevo_handler/__about__.py create mode 100644 mindsdb/integrations/handlers/brevo_handler/__init__.py create mode 100644 mindsdb/integrations/handlers/brevo_handler/brevo_handler.py create mode 100644 mindsdb/integrations/handlers/brevo_handler/brevo_tables.py create mode 100644 mindsdb/integrations/handlers/brevo_handler/connection_args.py create mode 100644 mindsdb/integrations/handlers/brevo_handler/icon.svg create mode 100644 mindsdb/integrations/handlers/brevo_handler/requirements.txt create mode 100644 mindsdb/integrations/handlers/brevo_handler/tests/test_brevo_handler.py diff --git a/mindsdb/integrations/handlers/brevo_handler/README.md b/mindsdb/integrations/handlers/brevo_handler/README.md new file mode 100644 index 0000000000..83145fca27 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/README.md @@ -0,0 +1,122 @@ +# Brevo Handler + +Brevo handler for MindsDB provides interfaces to connect to [Brevo](https://www.brevo.com/) +(formerly Sendinblue) via its Marketing API and pull email marketing data into MindsDB. + +## Brevo Handler Implementation + +This handler was implemented using the Brevo Marketing API v3 and the standard +Python `requests` library. No vendor SDK is required. + +All endpoints are served from a single host, `https://api.brevo.com/v3`, and the +API key is sent in an `api-key` request header. + +## Brevo Handler Initialization + +The Brevo handler is initialized with the following required parameter: + +* `api_key`: an API key generated in Brevo under **SMTP & API → API Keys**. + +## Implemented Features + +* `contacts` table — subscribers, with the nested `attributes` object flattened + into columns +* `lists` table — audiences and their subscriber counts +* `campaigns` table — email campaigns with send status and delivery statistics + (sent, delivered, bounces, opens, clicks, unsubscriptions, complaints), summed + across every list a campaign was sent to +* `contact_events` table — per-contact activity (opens, clicks, and so on), + built from each contact's own statistics + +Collection endpoints are paginated with `limit` and `offset`. The `count` field +in each response is the **total** number of records, not the number returned, so +it is used as the stopping condition when fetching all pages. + +## TODO + +* Support `INSERT` on `contacts` so subscribers can be added from MindsDB +* Push more filters down to the API instead of filtering locally + +All tables are read-only. Equality filters on `contacts.email`, `contacts.list_id` +and on `contact_events.email`, `campaign_id` and `event_type` are sent to Brevo +instead of being applied locally. + +## Usage + +To connect your Brevo account to MindsDB, use the below `CREATE DATABASE` statement: + +```sql +CREATE DATABASE brevo_src +WITH ENGINE = 'brevo', +PARAMETERS = { + "api_key": "xkeysib-your-key-here" +}; +``` + +Query your contacts: + +```sql +SELECT * FROM brevo_src.contacts; +``` + +Apply filters and limits: + +```sql +SELECT email, first_name, last_name +FROM brevo_src.contacts +WHERE list_id = 2 +LIMIT 10; +``` + +List your audiences: + +```sql +SELECT list_id, name, unique_subscribers +FROM brevo_src.lists; +``` + +Join Brevo contacts against customer records held in another datasource, to see +which paying customers are actually subscribed: + +```sql +SELECT c.name, c.email, c.total_spent, b.email_blacklisted +FROM postgres_src.public.customers AS c +JOIN brevo_src.contacts AS b + ON c.email = b.email +ORDER BY c.total_spent DESC; +``` + +Campaign performance: + +```sql +SELECT campaign_name, sent, delivered, open_rate +FROM brevo_src.campaigns +ORDER BY open_rate DESC; +``` + +Who opened a campaign: + +```sql +SELECT email, event_time +FROM brevo_src.contact_events +WHERE campaign_id = 1 AND event_type = 'opened' +LIMIT 50; +``` + +## Notes + +**`contact_events` can be slow.** Brevo only exposes events per contact. With an +`email = ...` filter this is one API call; without it, the handler makes one call +per contact. Filter by email or use `LIMIT` on large audiences. + +**Authorised IP addresses.** Brevo can restrict API access to an allowlist of IP +addresses. If requests fail with HTTP 401 and a message naming an +"unrecognised IP address", the key is valid but the caller's address is not +approved — add it at +[app.brevo.com/security/authorised_ips](https://app.brevo.com/security/authorised_ips). +Home connections are often assigned a new address every day or two, so this can +reappear after it was working. `check_connection` surfaces the message from +Brevo unchanged so this is distinguishable from a bad key. + +**A contact can belong to several lists.** The API returns `listIds` as an array. +Filtering by `list_id` matches a contact that belongs to that list, among others. diff --git a/mindsdb/integrations/handlers/brevo_handler/__about__.py b/mindsdb/integrations/handlers/brevo_handler/__about__.py new file mode 100644 index 0000000000..5de58597e2 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/__about__.py @@ -0,0 +1,9 @@ +__title__ = "MindsDB Brevo handler" +__package_name__ = "mindsdb_brevo_handler" +__version__ = "0.0.1" +__description__ = "MindsDB handler for Brevo" +__author__ = "Brejesh" +__github__ = "https://github.com/mindsdb/mindsdb" +__pypi__ = "https://pypi.org/project/mindsdb/" +__license__ = "MIT" +__copyright__ = "Copyright 2026 - mindsdb" diff --git a/mindsdb/integrations/handlers/brevo_handler/__init__.py b/mindsdb/integrations/handlers/brevo_handler/__init__.py new file mode 100644 index 0000000000..a10ce1985f --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/__init__.py @@ -0,0 +1,33 @@ +from mindsdb.integrations.libs.const import HANDLER_SUPPORT_LEVEL, HANDLER_TYPE + +from .__about__ import __description__ as description +from .__about__ import __version__ as version +from .connection_args import connection_args, connection_args_example + +try: + from .brevo_handler import BrevoHandler as Handler + + import_error = None +except Exception as e: + Handler = None + import_error = e + +title = "Brevo" +name = "brevo" +type = HANDLER_TYPE.DATA +icon_path = "icon.svg" +support_level = HANDLER_SUPPORT_LEVEL.COMMUNITY + +__all__ = [ + "Handler", + "connection_args", + "connection_args_example", + "description", + "icon_path", + "import_error", + "name", + "support_level", + "title", + "type", + "version", +] diff --git a/mindsdb/integrations/handlers/brevo_handler/brevo_handler.py b/mindsdb/integrations/handlers/brevo_handler/brevo_handler.py new file mode 100644 index 0000000000..fdb9f4bb8e --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/brevo_handler.py @@ -0,0 +1,83 @@ +import requests + +from mindsdb.integrations.libs.api_handler import APIHandler +from mindsdb.integrations.libs.response import HandlerStatusResponse as StatusResponse +from mindsdb.utilities import log + +from .brevo_tables import CampaignsTable, ContactEventsTable, ContactsTable, ListsTable + +logger = log.getLogger(__name__) + +BASE_URL = "https://api.brevo.com/v3" + + +class BrevoHandler(APIHandler): + """Connects MindsDB to Brevo (email marketing) through its REST API. + + Exposes four read-only tables: campaigns, contact_events, contacts and lists. + """ + + name = "brevo" + + def __init__(self, name: str = None, **kwargs): + """Create the handler and register its tables. + + No network call is made here; the API key is only checked when a query + runs or `check_connection` is called. + + Args: + name: The name of this data source in MindsDB. Defaults to None. + **kwargs: Handler options. `connection_data` should be a dict with + the `api_key` for the Brevo account. + """ + super().__init__(name) + self.connection_data = kwargs.get("connection_data", {}) or {} + self._register_table("campaigns", CampaignsTable(self)) + self._register_table("contact_events", ContactEventsTable(self)) + self._register_table("contacts", ContactsTable(self)) + self._register_table("lists", ListsTable(self)) + + def connect(self): + """Build the HTTP headers used for every Brevo request. + + Brevo has no persistent connection, so this only reads the API key and + marks the handler as connected. No network call happens here. + + Returns: + A dict with the `api-key` and `accept` headers. + + Raises: + ValueError: If `api_key` is missing from the connection data. + """ + api_key = self.connection_data.get("api_key") + if not api_key: + raise ValueError("The 'api_key' parameter is required to connect to Brevo.") + self.is_connected = True + return {"api-key": api_key, "accept": "application/json"} + + def check_connection(self) -> StatusResponse: + """Check that the API key works by calling Brevo's /account endpoint. + + Brevo's own error text is passed through on failure, because a 401 is + often caused by the account's IP allowlist rather than a bad key. + + Returns: + A StatusResponse with `success` set to True on HTTP 200. Otherwise + `success` is False and `error_message` explains why. This method + does not raise; all errors are reported through the response. + """ + response = StatusResponse(False) + try: + r = requests.get(f"{BASE_URL}/account", headers=self.connect(), timeout=30) + if r.status_code == 200: + response.success = True + else: + # Pass Brevo's own message through - a 401 here is often the IP + # allowlist, not a bad key, and the text says which. + response.error_message = f"Brevo returned {r.status_code}: {r.text[:300]}" + self.is_connected = False + except Exception as e: + logger.error(f"Error connecting to Brevo: {e}") + response.error_message = str(e) + self.is_connected = False + return response diff --git a/mindsdb/integrations/handlers/brevo_handler/brevo_tables.py b/mindsdb/integrations/handlers/brevo_handler/brevo_tables.py new file mode 100644 index 0000000000..5491cc1700 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/brevo_tables.py @@ -0,0 +1,642 @@ +import builtins +import re +import time +from concurrent.futures import ThreadPoolExecutor +from urllib.parse import quote + +import pandas as pd +import requests + +from mindsdb.integrations.libs.api_handler import APIResource +from mindsdb.integrations.utilities.sql_utils import FilterOperator +from mindsdb.utilities import log + +logger = log.getLogger(__name__) + +BASE_URL = "https://api.brevo.com/v3" +PAGE_SIZE = 50 +MAX_RETRIES = 5 +MAX_WORKERS = 7 + + +def _get_with_backoff(url, headers, params=None, timeout=30, max_retries=MAX_RETRIES): + """Send a GET request, retrying when Brevo rate-limits us. + + On an HTTP 429 the function sleeps for the number of seconds in the + Retry-After header, or 2 ** attempt seconds if the header is missing, then + tries again. Any other status code is returned straight away. + + Args: + url: The full Brevo API URL to request. + headers: HTTP headers for the request, including the api-key. + params: Optional query-string parameters. Defaults to None. + timeout: Seconds to wait for each attempt. Defaults to 30. + max_retries: How many times to retry after a 429. Defaults to + MAX_RETRIES. + + Returns: + The requests.Response from the last attempt. It can still be a 429 if + every retry was rate-limited, so the caller must check status_code. + """ + for attempt in range(max_retries + 1): + r = requests.get(url, headers=headers, params=params, timeout=timeout) + if r.status_code != 429 or attempt == max_retries: + return r + wait = float(r.headers.get("Retry-After", 2**attempt)) + logger.warning(f"Brevo: rate limited on {url}, retrying in {wait}s") + time.sleep(wait) + return r + + +def _equals(conditions, column): + """Find an equality filter on one column and mark it as handled. + + Setting `applied = True` tells MindsDB the handler already used this filter + (by sending it to Brevo), so MindsDB does not filter the rows a second time. + + Args: + conditions: The WHERE conditions MindsDB passed to the table, or None. + column: The column name to look for. + + Returns: + The value from the first `column = value` condition, or None if there + is no such condition. + """ + for c in conditions or []: + if c.column == column and c.op == FilterOperator.EQUAL: + c.applied = True + return c.value + return None + + +def _snake(name: str) -> str: + """Convert a camelCase name to snake_case. + + Args: + name: A camelCase string, for example "hardBounces". + + Returns: + The snake_case version, for example "hard_bounces". + """ + return re.sub(r"(? pd.DataFrame: + """Fetch email campaigns and their delivery statistics. + + Filters and sorting are applied by MindsDB after this method returns. + + Args: + conditions: WHERE conditions from the query. Not pushed to Brevo. + limit: Maximum number of rows to return. Defaults to None (all). + sort: Requested ordering. Not used here. + targets: Column names the query selected. Defaults to all columns. + **kwargs: Extra arguments passed by MindsDB, ignored. + + Returns: + A DataFrame with one row per campaign and the columns in + CampaignsTable.columns, restricted to `targets` when given. + + Raises: + RuntimeError: If Brevo returns a non-200 status code. + ValueError: If no api_key was given when the handler was created. + """ + headers = self.handler.connect() + campaigns = [] + offset = 0 + while True: + page = PAGE_SIZE if limit is None else min(PAGE_SIZE, limit - len(campaigns)) + if page <= 0: + break + + r = _get_with_backoff( + f"{BASE_URL}/emailCampaigns", + headers, + params={"limit": page, "offset": offset}, + ) + if r.status_code != 200: + raise RuntimeError(f"Brevo returned {r.status_code}: {r.text[:300]}") + + body = r.json() + batch = body.get("campaigns", []) + if not batch: + break + + campaigns.extend(batch) + offset += len(batch) + # `count` is the total number of campaigns, not the size of this page + if offset >= body.get("count", offset): + break + + rows = [] + for c in campaigns: + # NOTE: globalStats is always zero here; the real numbers are per-list + # in campaignStats, so sum across that array. + stats = (c.get("statistics") or {}).get("campaignStats") or [] + + sent, delivered = _total(stats, "sent"), _total(stats, "delivered") + unique_opens = _total(stats, "uniqueViews") + + rows.append( + { + "campaign_id": c.get("id"), + "campaign_name": c.get("name"), + "subject": c.get("subject"), + "status": c.get("status"), + "sent_at": c.get("sentDate"), + "sent": sent, + "delivered": delivered, + "hard_bounces": _total(stats, "hardBounces"), + "soft_bounces": _total(stats, "softBounces"), + "unique_opens": unique_opens, + "opens_total": _total(stats, "viewed"), + "unique_clicks": _total(stats, "uniqueClicks"), + "clickers": _total(stats, "clickers"), + "unsubscriptions": _total(stats, "unsubscriptions"), + "complaints": _total(stats, "complaints"), + "delivery_rate": _rate(delivered, sent), + "open_rate": _rate(unique_opens, delivered), + } + ) + + df = pd.DataFrame(rows, columns=self.columns) + if limit: + df = df.head(limit) + if targets: + keep = [c for c in self.columns if c in targets] + if keep: + df = df[keep] + return df + + def get_columns(self) -> builtins.list[str]: + """Return the column names of the campaigns table. + + Returns: + The names in CampaignsTable.columns. + """ + return self.columns + + +def _fetch_contact_events(addr, headers, campaign_id, event_type, rows, limit): + """Fetch one contact's activity and append the matching events to `rows`. + + Calls Brevo's contact endpoint with statistics enabled, then walks every + event list in the response (opens, clicks, and so on). Events are skipped + when they do not match `campaign_id` or `event_type`. If the request fails, + a warning is logged and nothing is added. + + Args: + addr: The contact's email address. + headers: HTTP headers for the request, including the api-key. + campaign_id: Keep only events from this campaign. None keeps all. + event_type: Keep only this snake_case event type, for example + "opened". None keeps all. + rows: List that matching events are appended to, as dicts with the keys + email, campaign_id, event_type and event_time. Modified in place. + limit: Stop once `rows` holds this many entries. None means no limit. + + Returns: + None. The results are written into `rows`. + """ + r = _get_with_backoff(f"{BASE_URL}/contacts/{addr}", headers, params={"statistics": "true"}) + if r.status_code != 200: + logger.warning(f"Brevo: skipping {addr}, got {r.status_code}") + return + + stats = r.json().get("statistics") or {} + for event_name, events in stats.items(): + if not isinstance(events, list): + continue + for e in events: + if campaign_id is not None and e.get("campaignId") != int(campaign_id): + continue + if event_type is not None and _snake(event_name) != event_type: + continue + rows.append( + { + "email": addr, + "campaign_id": e.get("campaignId"), + "event_type": _snake(event_name), + "event_time": e.get("eventTime"), + } + ) + if limit and len(rows) >= limit: + return + + +def _fetch_events_for_contacts(contacts, headers, campaign_id, event_type, limit): + """Fetch events for several contacts at the same time. + + Brevo only exposes events through a per-contact endpoint, so one call is + needed per contact. Each call mostly waits on the network, so running + MAX_WORKERS of them together is much faster than running them one by one. + + Every worker fills its own private list, and the lists are merged here in + the same order as `contacts`, so the output is identical to a sequential + run. Once `limit` rows are collected, calls that have not started yet are + cancelled. Calls already in flight finish and their rows are discarded. + + Args: + contacts: Contact dicts from Brevo, each with an "email" key. + headers: HTTP headers for the requests, including the api-key. + campaign_id: Keep only events from this campaign. None keeps all. + event_type: Keep only this snake_case event type, for example + "opened". None keeps all. + limit: Stop once this many rows are collected. None means no limit. + + Returns: + A list of event dicts with the keys email, campaign_id, event_type and + event_time, in the order of `contacts`. It can hold more than `limit` + rows, so the caller should trim it. + + Raises: + Exception: Anything a worker raises is re-raised here, so a failed + fetch is not silently dropped from the results. + """ + + def fetch_one(contact): + out = [] + _fetch_contact_events(contact["email"], headers, campaign_id, event_type, out, limit) + return out + + rows = [] + with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: + futures = [pool.submit(fetch_one, c) for c in contacts] + for future in futures: + rows.extend(future.result()) + if limit and len(rows) >= limit: + for pending in futures: + pending.cancel() + break + return rows + + +class ContactEventsTable(APIResource): + """Table of contact activity, one row per contact per event. + + Use it to answer questions like "who opened this campaign". Brevo has no + single endpoint for this, so rows are built from each contact's own + statistics. + """ + + columns = ["email", "campaign_id", "event_type", "event_time"] + + def list(self, conditions=None, limit=None, sort=None, targets=None, **kwargs) -> pd.DataFrame: + """Fetch contact events such as opens and clicks. + + With an `email = ...` filter this makes a single API call. Without one, + it pages through every contact and makes one call per contact, which is + slow and can hit rate limits on large audiences, so filter by email or + use `limit` where possible. + + Args: + conditions: WHERE conditions from the query. Equality filters on + email, campaign_id and event_type are applied inside the + handler; other conditions are applied by MindsDB afterwards. + limit: Maximum number of rows to return. Defaults to None (all). + sort: Requested ordering. Not used here. + targets: Column names the query selected. Defaults to all columns. + **kwargs: Extra arguments passed by MindsDB, ignored. + + Returns: + A DataFrame with the columns in ContactEventsTable.columns, + restricted to `targets` when given. + + Raises: + RuntimeError: If listing contacts returns a non-200 status code. + ValueError: If no api_key was given when the handler was created. + """ + headers = self.handler.connect() + + # Events come from a per-contact endpoint, so filtering by email turns + # this from N calls into one. Push it down whenever we can. + email = _equals(conditions, "email") + campaign_id = _equals(conditions, "campaign_id") + event_type = _equals(conditions, "event_type") + + rows = [] + + if email: + _fetch_contact_events(email, headers, campaign_id, event_type, rows, limit) + else: + offset = 0 + while True: + r = _get_with_backoff( + f"{BASE_URL}/contacts", + headers, + params={"limit": PAGE_SIZE, "offset": offset}, + ) + if r.status_code != 200: + raise RuntimeError(f"Brevo returned {r.status_code}: {r.text[:300]}") + + body = r.json() + page = body.get("contacts", []) + if not page: + break + + # Fetch this page's contacts in parallel, one page at a time, + # so we can stop as soon as LIMIT is satisfied instead of + # always fetching every page up front. + rows.extend(_fetch_events_for_contacts(page, headers, campaign_id, event_type, limit)) + + if limit and len(rows) >= limit: + break + + offset += len(page) + + # `count` is the total number of contacts, not the size of this page + if offset >= body.get("count", offset): + break + + df = pd.DataFrame(rows, columns=self.columns) + if limit: + df = df.head(limit) + if targets: + keep = [c for c in self.columns if c in targets] + if keep: + df = df[keep] + return df + + def get_columns(self) -> builtins.list[str]: + """Return the column names of the contact_events table. + + Returns: + The names in ContactEventsTable.columns. + """ + return self.columns + + +def _contact_row(c): + """Flatten one Brevo contact object into a row for the contacts table. + + Args: + c: A contact dict as returned by Brevo. + + Returns: + A dict keyed by the names in ContactsTable.columns. + """ + attrs = c.get("attributes") or {} + return { + "contact_id": c.get("id"), + "email": c.get("email"), + "first_name": attrs.get("FIRSTNAME"), + "last_name": attrs.get("LASTNAME"), + "sms": attrs.get("SMS"), + # a contact can be in several lists; SQL cells cannot hold a + # list, so join the ids into a readable string + "list_ids": ",".join(str(i) for i in (c.get("listIds") or [])), + "email_blacklisted": c.get("emailBlacklisted"), + "sms_blacklisted": c.get("smsBlacklisted"), + "created_at": c.get("createdAt"), + "modified_at": c.get("modifiedAt"), + } + + +class ContactsTable(APIResource): + """Table of subscribers, one row per contact. + + Brevo nests names and phone numbers inside an `attributes` object; the + table flattens them into ordinary columns. + """ + + columns = [ + "contact_id", + "email", + "first_name", + "last_name", + "sms", + "list_ids", + "email_blacklisted", + "sms_blacklisted", + "created_at", + "modified_at", + ] + + def list(self, conditions=None, limit=None, sort=None, targets=None, **kwargs) -> pd.DataFrame: + """Fetch contacts, paging through Brevo until all are read. + + Args: + conditions: WHERE conditions from the query. An equality filter on + email fetches that single contact, and one on list_id is + pushed to Brevo's per-list endpoint; other conditions are + applied by MindsDB afterwards. + limit: Maximum number of rows to return. Defaults to None (all). + sort: Requested ordering. Not used here. + targets: Column names the query selected. Defaults to all columns. + **kwargs: Extra arguments passed by MindsDB, ignored. + + Returns: + A DataFrame with the columns in ContactsTable.columns, restricted + to `targets` when given. list_ids is a comma-separated string + because a SQL cell cannot hold a list. + + Raises: + RuntimeError: If Brevo returns a non-200 status code. + ValueError: If no api_key was given when the handler was created. + """ + headers = self.handler.connect() + + # Brevo can fetch one contact directly by email, so an email filter is + # one request instead of paging through every contact. + email = _equals(conditions, "email") + if email: + r = _get_with_backoff(f"{BASE_URL}/contacts/{quote(str(email), safe='')}", headers) + if r.status_code == 404: + rows = [] + elif r.status_code == 200: + rows = [_contact_row(r.json())] + else: + raise RuntimeError(f"Brevo returned {r.status_code}: {r.text[:300]}") + df = pd.DataFrame(rows, columns=self.columns) + if targets: + keep = [c for c in self.columns if c in targets] + if keep: + df = df[keep] + return df + + # Brevo has a dedicated endpoint for one list, so a list_id filter is + # cheaper to push down than to apply locally afterwards. + list_id = _equals(conditions, "list_id") + path = f"/contacts/lists/{list_id}/contacts" if list_id else "/contacts" + + rows = [] + offset = 0 + while True: + page = PAGE_SIZE if limit is None else min(PAGE_SIZE, limit - len(rows)) + if page <= 0: + break + + r = _get_with_backoff( + f"{BASE_URL}{path}", + headers, + params={"limit": page, "offset": offset}, + ) + if r.status_code != 200: + raise RuntimeError(f"Brevo returned {r.status_code}: {r.text[:300]}") + + body = r.json() + contacts = body.get("contacts", []) + if not contacts: + break + + rows.extend(_contact_row(c) for c in contacts) + + offset += len(contacts) + if offset >= body.get("count", offset): + break + + df = pd.DataFrame(rows, columns=self.columns) + if targets: + keep = [c for c in self.columns if c in targets] + if keep: + df = df[keep] + return df + + def get_columns(self) -> builtins.list[str]: + """Return the column names of the contacts table. + + Returns: + The names in ContactsTable.columns. + """ + return self.columns + + +class ListsTable(APIResource): + """Table of contact lists, one row per list. + + Brevo calls them lists in the API and audiences in its web interface. + """ + + columns = [ + "list_id", + "name", + "folder_id", + "total_subscribers", + "unique_subscribers", + "total_blacklisted", + ] + + def list(self, conditions=None, limit=None, sort=None, targets=None, **kwargs) -> pd.DataFrame: + """Fetch contact lists, paging through Brevo until all are read. + + Args: + conditions: WHERE conditions from the query. Not pushed to Brevo. + limit: Maximum number of rows to return. Defaults to None (all). + sort: Requested ordering. Not used here. + targets: Column names the query selected. Defaults to all columns. + **kwargs: Extra arguments passed by MindsDB, ignored. + + Returns: + A DataFrame with the columns in ListsTable.columns, restricted to + `targets` when given. + + Raises: + RuntimeError: If Brevo returns a non-200 status code. + ValueError: If no api_key was given when the handler was created. + """ + headers = self.handler.connect() + + rows = [] + offset = 0 + while True: + page = PAGE_SIZE if limit is None else min(PAGE_SIZE, limit - len(rows)) + if page <= 0: + break + + r = _get_with_backoff( + f"{BASE_URL}/contacts/lists", + headers, + params={"limit": page, "offset": offset}, + ) + if r.status_code != 200: + raise RuntimeError(f"Brevo returned {r.status_code}: {r.text[:300]}") + + body = r.json() + lists = body.get("lists", []) + if not lists: + break + + for item in lists: + rows.append( + { + "list_id": item.get("id"), + "name": item.get("name"), + "folder_id": item.get("folderId"), + "total_subscribers": item.get("totalSubscribers"), + "unique_subscribers": item.get("uniqueSubscribers"), + "total_blacklisted": item.get("totalBlacklisted"), + } + ) + + offset += len(lists) + if offset >= body.get("count", offset): + break + + df = pd.DataFrame(rows, columns=self.columns) + if targets: + keep = [c for c in self.columns if c in targets] + if keep: + df = df[keep] + return df + + def get_columns(self) -> builtins.list[str]: + """Return the column names of the lists table. + + Returns: + The names in ListsTable.columns. + """ + return self.columns diff --git a/mindsdb/integrations/handlers/brevo_handler/connection_args.py b/mindsdb/integrations/handlers/brevo_handler/connection_args.py new file mode 100644 index 0000000000..0342b87e04 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/connection_args.py @@ -0,0 +1,15 @@ +from collections import OrderedDict + +from mindsdb.integrations.libs.const import HANDLER_CONNECTION_ARG_TYPE as ARG_TYPE + +connection_args = OrderedDict( + api_key={ + "type": ARG_TYPE.PWD, + "description": "Brevo API key", + "required": True, + "label": "API key", + "secret": True, + }, +) + +connection_args_example = OrderedDict(api_key="xkeysib-your-api-key") diff --git a/mindsdb/integrations/handlers/brevo_handler/icon.svg b/mindsdb/integrations/handlers/brevo_handler/icon.svg new file mode 100644 index 0000000000..902f3ce845 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/icon.svg @@ -0,0 +1,4 @@ + + + + diff --git a/mindsdb/integrations/handlers/brevo_handler/requirements.txt b/mindsdb/integrations/handlers/brevo_handler/requirements.txt new file mode 100644 index 0000000000..663bd1f6a2 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/requirements.txt @@ -0,0 +1 @@ +requests \ No newline at end of file diff --git a/mindsdb/integrations/handlers/brevo_handler/tests/test_brevo_handler.py b/mindsdb/integrations/handlers/brevo_handler/tests/test_brevo_handler.py new file mode 100644 index 0000000000..6a1f21f162 --- /dev/null +++ b/mindsdb/integrations/handlers/brevo_handler/tests/test_brevo_handler.py @@ -0,0 +1,185 @@ +import unittest +from unittest.mock import MagicMock, patch + +from mindsdb.integrations.handlers.brevo_handler.brevo_handler import BrevoHandler +from mindsdb.integrations.handlers.brevo_handler.brevo_tables import ( + CampaignsTable, + ContactEventsTable, + ContactsTable, + ListsTable, + _get_with_backoff, +) +from mindsdb.integrations.utilities.sql_utils import FilterOperator + + +def _response(status_code, json_body=None, text="", headers=None): + r = MagicMock() + r.status_code = status_code + r.json.return_value = json_body or {} + r.text = text + r.headers = headers or {} + return r + + +class BrevoHandlerTest(unittest.TestCase): + def setUp(self): + self.handler = BrevoHandler("test_brevo", connection_data={"api_key": "fake-key"}) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_handler.requests.get") + def test_check_connection_success(self, mock_get): + mock_get.return_value = _response(200, {"email": "me@example.com"}) + response = self.handler.check_connection() + self.assertTrue(response.success) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_handler.requests.get") + def test_check_connection_failure_passes_brevo_message(self, mock_get): + mock_get.return_value = _response(401, text="unauthorized IP") + response = self.handler.check_connection() + self.assertFalse(response.success) + self.assertIn("unauthorized IP", response.error_message) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contacts_list(self, mock_get): + mock_get.side_effect = [ + _response( + 200, + { + "contacts": [ + { + "id": 1, + "email": "a@example.com", + "attributes": {"FIRSTNAME": "A"}, + "listIds": [1, 2], + "emailBlacklisted": False, + "smsBlacklisted": False, + "createdAt": "2024-01-01", + "modifiedAt": "2024-01-02", + } + ], + "count": 1, + }, + ), + ] + table = ContactsTable(self.handler) + df = table.list() + self.assertEqual(len(df), 1) + self.assertEqual(df.iloc[0]["email"], "a@example.com") + self.assertEqual(df.iloc[0]["list_ids"], "1,2") + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contact_events_skips_failed_contact(self, mock_get): + mock_get.side_effect = [ + _response(200, {"contacts": [{"email": "a@example.com"}], "count": 1}), + _response(404, text="not found"), + ] + table = ContactEventsTable(self.handler) + df = table.list(conditions=[]) + self.assertEqual(len(df), 0) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contacts_pagination_uses_total_count(self, mock_get): + mock_get.side_effect = [ + _response(200, {"contacts": [{"email": "a@example.com"}, {"email": "b@example.com"}], "count": 3}), + _response(200, {"contacts": [{"email": "c@example.com"}], "count": 3}), + ] + df = ContactsTable(self.handler).list() + self.assertEqual(list(df["email"]), ["a@example.com", "b@example.com", "c@example.com"]) + self.assertEqual(mock_get.call_args_list[1].kwargs["params"]["offset"], 2) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contacts_email_filter_hits_single_contact(self, mock_get): + mock_get.return_value = _response(200, {"id": 1, "email": "a@example.com", "listIds": [1]}) + condition = MagicMock(column="email", op=FilterOperator.EQUAL, value="a@example.com") + df = ContactsTable(self.handler).list(conditions=[condition]) + self.assertEqual(len(df), 1) + self.assertTrue(condition.applied) + self.assertIn("/contacts/a%40example.com", mock_get.call_args.args[0]) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contacts_error_raises(self, mock_get): + mock_get.return_value = _response(500, text="boom") + with self.assertRaises(RuntimeError): + ContactsTable(self.handler).list() + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_lists_list(self, mock_get): + mock_get.return_value = _response( + 200, {"lists": [{"id": 2, "name": "News", "folderId": 1, "totalSubscribers": 5}], "count": 1} + ) + df = ListsTable(self.handler).list() + self.assertEqual(df.iloc[0]["name"], "News") + self.assertEqual(df.iloc[0]["total_subscribers"], 5) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_campaigns_sum_stats_across_lists_and_paginate(self, mock_get): + def campaign(cid): + return { + "id": cid, + "name": f"c{cid}", + "statistics": { + "campaignStats": [ + {"sent": 6, "delivered": 5, "uniqueViews": 2}, + {"sent": 4, "delivered": 3, "uniqueViews": 2}, + ] + }, + } + + mock_get.side_effect = [ + _response(200, {"campaigns": [campaign(1)], "count": 2}), + _response(200, {"campaigns": [campaign(2)], "count": 2}), + ] + df = CampaignsTable(self.handler).list() + self.assertEqual(len(df), 2) + self.assertEqual(df.iloc[0]["sent"], 10) + self.assertEqual(df.iloc[0]["delivered"], 8) + self.assertEqual(df.iloc[0]["open_rate"], 0.5) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_campaign_with_no_sends_has_null_rates(self, mock_get): + mock_get.return_value = _response(200, {"campaigns": [{"id": 1, "name": "draft"}], "count": 1}) + df = CampaignsTable(self.handler).list() + self.assertTrue(df.iloc[0]["open_rate"] is None or df.isna().iloc[0]["open_rate"]) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_contact_events_email_filter(self, mock_get): + mock_get.return_value = _response( + 200, + {"statistics": {"opened": [{"campaignId": 1, "eventTime": "t1"}, {"campaignId": 2, "eventTime": "t2"}]}}, + ) + condition = MagicMock(column="campaign_id", op=FilterOperator.EQUAL, value=1) + email = MagicMock(column="email", op=FilterOperator.EQUAL, value="a@example.com") + df = ContactEventsTable(self.handler).list(conditions=[email, condition]) + self.assertEqual(len(df), 1) + self.assertEqual(df.iloc[0]["event_type"], "opened") + + def test_missing_api_key_raises(self): + with self.assertRaises(ValueError): + BrevoHandler("x", connection_data={}).connect() + + +class BackoffTest(unittest.TestCase): + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.time.sleep") + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_retries_on_429_then_succeeds(self, mock_get, mock_sleep): + mock_get.side_effect = [ + _response(429, headers={"Retry-After": "1"}), + _response(200, {"ok": True}), + ] + r = _get_with_backoff("https://api.brevo.com/v3/contacts/a@example.com", {}) + self.assertEqual(r.status_code, 200) + self.assertEqual(mock_get.call_count, 2) + mock_sleep.assert_called_once_with(1.0) + + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.time.sleep") + @patch("mindsdb.integrations.handlers.brevo_handler.brevo_tables.requests.get") + def test_gives_up_after_max_retries(self, mock_get, mock_sleep): + mock_get.return_value = _response(429) + r = _get_with_backoff("https://api.brevo.com/v3/contacts", {}, max_retries=2) + self.assertEqual(r.status_code, 429) + self.assertEqual(mock_get.call_count, 3) + + +if __name__ == "__main__": + unittest.main() + +# python -m unittest mindsdb.integrations.handlers.brevo_handler.tests.test_brevo_handler -v