Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,7 @@ The same situation applies to both `client.batch_send()` and `client.sending_api

### Stats API:
- Sending stats – [`stats/stats.py`](examples/stats/stats.py)
- Email Marketing stats (campaigns & automations) – [`marketing_stats/marketing_stats.py`](examples/marketing_stats/marketing_stats.py)

### Email Logs API:
- List email logs (with filters & pagination) and get message by ID – [`email_logs/email_logs.py`](examples/email_logs/email_logs.py)
Expand Down
49 changes: 49 additions & 0 deletions examples/marketing_stats/marketing_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
import os
from datetime import datetime
from datetime import timedelta
from datetime import timezone

import mailtrap as mt
from mailtrap.models.marketing_stats import MarketingStats

API_KEY = os.environ["MAILTRAP_API_KEY"]

client = mt.MailtrapClient(token=API_KEY)
marketing_stats_api = client.marketing_stats_api.marketing_stats


def last_30_days() -> tuple[str, str]:
# Both dates are inclusive, so today plus the 29 days before it is 30 days.
end_date = datetime.now(timezone.utc).date()
start_date = end_date - timedelta(days=29)
return start_date.isoformat(), end_date.isoformat()


def get_marketing_stats() -> MarketingStats:
# Campaigns and automations together, across every sending domain and
# email service provider of the account.
start_date, end_date = last_30_days()
return marketing_stats_api.get(
params=mt.MarketingStatsFilterParams(start_date=start_date, end_date=end_date)
)


def get_marketing_stats_with_filters() -> MarketingStats:
# Campaigns only (pass ["automations"] for automations), for two sending
# domains (IDs from the Sending Domains API) and recipients at Google and
# Yahoo.
start_date, end_date = last_30_days()
return marketing_stats_api.get(
params=mt.MarketingStatsFilterParams(
start_date=start_date,
end_date=end_date,
sending_domain_ids=[1, 2],
sources=["campaigns"],
email_service_providers=["Google", "Yahoo"],
)
)


if __name__ == "__main__":
print(get_marketing_stats())
print(get_marketing_stats_with_filters())
1 change: 1 addition & 0 deletions mailtrap/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
from .models.mail import Disposition
from .models.mail import Mail
from .models.mail import MailFromTemplate
from .models.marketing_stats import MarketingStatsFilterParams
from .models.messages import UpdateEmailMessageParams
from .models.organizations import CreateSubAccountParams
from .models.paginated_templates import CreateTemplateParams
Expand Down
11 changes: 11 additions & 0 deletions mailtrap/api/marketing_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
from mailtrap.api.resources.marketing_stats import MarketingStatsApi
from mailtrap.http import HttpClient


class MarketingStatsBaseApi:
def __init__(self, client: HttpClient) -> None:
self._client = client

@property
def marketing_stats(self) -> MarketingStatsApi:
return MarketingStatsApi(client=self._client)
20 changes: 20 additions & 0 deletions mailtrap/api/resources/marketing_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
from mailtrap.http import HttpClient
from mailtrap.models.marketing_stats import MarketingStats
from mailtrap.models.marketing_stats import MarketingStatsFilterParams
from mailtrap.models.marketing_stats import MarketingStatsResponse


class MarketingStatsApi:
def __init__(self, client: HttpClient) -> None:
self._client = client

def get(self, params: MarketingStatsFilterParams) -> MarketingStats:
"""
Get aggregated Email Marketing stats, campaigns and automations
together, for a date range. Email API/SMTP traffic is not included;
``stats_api`` reports it.
"""
response = self._client.get(
"/api/marketing/stats", params=params.api_query_params
)
return MarketingStatsResponse(**response).data
7 changes: 7 additions & 0 deletions mailtrap/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from mailtrap.api.email_logs import EmailLogsBaseApi
from mailtrap.api.general import GeneralApi
from mailtrap.api.inbound import InboundBaseApi
from mailtrap.api.marketing_stats import MarketingStatsBaseApi
from mailtrap.api.organizations import OrganizationsBaseApi
from mailtrap.api.paginated_templates import TemplatesBaseApi
from mailtrap.api.resources.stats import StatsApi
Expand Down Expand Up @@ -154,6 +155,12 @@ def email_campaigns_api(self) -> EmailCampaignsBaseApi:
client=HttpClient(host=GENERAL_HOST, headers=self.headers),
)

@property
def marketing_stats_api(self) -> MarketingStatsBaseApi:
return MarketingStatsBaseApi(
client=HttpClient(host=GENERAL_HOST, headers=self.headers),
)

@property
def email_logs_api(self) -> EmailLogsBaseApi:
self._validate_account_id("Email Logs API")
Expand Down
54 changes: 54 additions & 0 deletions mailtrap/models/marketing_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
from typing import Literal
from typing import Optional

from pydantic.dataclasses import dataclass

from mailtrap.models.common import RequestParams

MarketingStatsSource = Literal["campaigns", "automations"]


@dataclass
class MarketingStats:
"""
Aggregated Email Marketing metrics (campaigns and automations). Rates are
fractions from 0 to 1: delivery and bounce rates are shares of
``sent_count``, the rest are shares of ``delivery_count``. Every count and
rate is ``0`` when nothing matched the filters.
"""

delivery_count: int
delivery_rate: float
bounce_count: int
bounce_rate: float
open_count: int
open_rate: float
click_count: int
click_rate: float
spam_count: int
spam_rate: float
unsubscription_count: int
unsubscription_rate: float
sent_count: int


@dataclass
class MarketingStatsResponse:
"""Envelope of the marketing stats response."""

data: MarketingStats


@dataclass
class MarketingStatsFilterParams(RequestParams):
"""
Query params for marketing stats. ``start_date`` and ``end_date`` are
required and inclusive (``YYYY-MM-DD``). Omitted list filters include
everything; one source narrows the results to campaigns or automations.
"""

start_date: str
end_date: str
sending_domain_ids: Optional[list[int]] = None
sources: Optional[list[MarketingStatsSource]] = None
email_service_providers: Optional[list[str]] = None
Empty file.
168 changes: 168 additions & 0 deletions tests/unit/api/marketing_stats/test_marketing_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,168 @@
from typing import Any
from urllib.parse import parse_qs
from urllib.parse import urlparse

import pytest
import responses
from pydantic import ValidationError

from mailtrap.api.resources.marketing_stats import MarketingStatsApi
from mailtrap.config import GENERAL_HOST
from mailtrap.exceptions import APIError
from mailtrap.http import HttpClient
from mailtrap.models.marketing_stats import MarketingStats
from mailtrap.models.marketing_stats import MarketingStatsFilterParams
from tests import conftest

MARKETING_STATS_URL = f"https://{GENERAL_HOST}/api/marketing/stats"


@pytest.fixture
def client() -> MarketingStatsApi:
return MarketingStatsApi(client=HttpClient(GENERAL_HOST))


SAMPLE_STATS: dict[str, Any] = {
"delivery_count": 1200,
"delivery_rate": 0.96,
"bounce_count": 50,
"bounce_rate": 0.04,
"open_count": 600,
"open_rate": 0.5,
"click_count": 240,
"click_rate": 0.2,
"spam_count": 3,
"spam_rate": 0.0025,
"unsubscription_count": 12,
"unsubscription_rate": 0.01,
"sent_count": 1250,
}


def _default_params() -> MarketingStatsFilterParams:
return MarketingStatsFilterParams(start_date="2026-05-01", end_date="2026-05-31")


class TestMarketingStatsApi:

@pytest.mark.parametrize(
"status_code,response_json,expected_error_message",
[
(
400,
{"errors": "Invalid end_date: '2026-13-01'"},
"Invalid end_date: '2026-13-01'",
),
(
conftest.UNAUTHORIZED_STATUS_CODE,
conftest.UNAUTHORIZED_RESPONSE,
conftest.UNAUTHORIZED_ERROR_MESSAGE,
),
(
conftest.FORBIDDEN_STATUS_CODE,
conftest.FORBIDDEN_RESPONSE,
conftest.FORBIDDEN_ERROR_MESSAGE,
),
(
conftest.RATE_LIMIT_ERROR_STATUS_CODE,
conftest.RATE_LIMIT_ERROR_RESPONSE,
conftest.RATE_LIMIT_ERROR_MESSAGE,
),
],
)
@responses.activate
def test_get_should_raise_api_errors(
self,
client: MarketingStatsApi,
status_code: int,
response_json: dict,
expected_error_message: str,
) -> None:
responses.get(
MARKETING_STATS_URL,
status=status_code,
json=response_json,
)

with pytest.raises(APIError) as exc_info:
client.get(_default_params())

assert expected_error_message in str(exc_info.value)

@pytest.mark.parametrize(
"stats_dict",
[SAMPLE_STATS, dict.fromkeys(SAMPLE_STATS, 0)],
ids=["with_activity", "nothing_matched"],
)
@responses.activate
def test_get_should_unwrap_data_envelope(
self, client: MarketingStatsApi, stats_dict: dict[str, Any]
) -> None:
responses.get(
MARKETING_STATS_URL,
json={"data": stats_dict},
status=200,
)

stats = client.get(_default_params())

assert stats == MarketingStats(**stats_dict)

@pytest.mark.parametrize(
"params,expected_query",
[
(
_default_params(),
{"start_date": ["2026-05-01"], "end_date": ["2026-05-31"]},
),
(
MarketingStatsFilterParams(
start_date="2026-05-01",
end_date="2026-05-31",
sending_domain_ids=[3938, 3939],
sources=["campaigns"],
email_service_providers=["Google", "Google Workspace"],
),
{
"start_date": ["2026-05-01"],
"end_date": ["2026-05-31"],
"sending_domain_ids[]": ["3938", "3939"],
"sources[]": ["campaigns"],
"email_service_providers[]": ["Google", "Google Workspace"],
},
),
],
ids=["dates_only", "with_filters_as_array_params"],
)
@responses.activate
def test_get_should_send_query_params(
self,
client: MarketingStatsApi,
params: MarketingStatsFilterParams,
expected_query: dict[str, list[str]],
) -> None:
responses.get(
MARKETING_STATS_URL,
json={"data": SAMPLE_STATS},
status=200,
)

client.get(params)

query = parse_qs(urlparse(responses.calls[0].request.url).query)
assert query == expected_query


class TestMarketingStatsFilterParams:

def test_should_reject_unknown_source(self) -> None:
with pytest.raises(ValidationError):
MarketingStatsFilterParams(
start_date="2026-05-01",
end_date="2026-05-31",
sources=["newsletters"], # type: ignore[list-item]
)

def test_should_require_both_dates(self) -> None:
with pytest.raises(ValidationError):
MarketingStatsFilterParams(start_date="2026-05-01") # type: ignore[call-arg]
5 changes: 5 additions & 0 deletions tests/unit/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,11 @@ def test_email_campaigns_api_does_not_require_account_id(self) -> None:

assert client.email_campaigns_api.email_campaigns is not None

def test_marketing_stats_api_does_not_require_account_id(self) -> None:
client = self.get_client()

assert client.marketing_stats_api.marketing_stats is not None

@pytest.mark.parametrize(
"arguments, expected_url",
[
Expand Down
2 changes: 1 addition & 1 deletion tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ commands = pre-commit run --all-files
[testenv:mypy]
deps =
-r requirements.test.txt
mypy==1.15.0
mypy==1.19.1
types-requests
commands = mypy ./mailtrap

Expand Down
Loading