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
3 changes: 3 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,9 @@ The same situation applies to both `client.batch_send()` and `client.sending_api
- Messages (list/get/delete + reply/reply_all/forward) – [`inbound/messages.py`](examples/inbound/messages.py)
- Threads (list/get/delete) – [`inbound/threads.py`](examples/inbound/threads.py)

### Email Campaigns API:
- Email Campaigns (list, create, get, update, delete, lifecycle actions, stats) – [`email_campaigns/email_campaigns.py`](examples/email_campaigns/email_campaigns.py)

### Webhooks API:
- Webhooks management – [`webhooks/webhooks.py`](examples/webhooks/webhooks.py)
- Verifying webhook signatures – [`webhooks/verify_signature.py`](examples/webhooks/verify_signature.py)
Expand Down
159 changes: 159 additions & 0 deletions examples/email_campaigns/email_campaigns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
from datetime import datetime
from datetime import timedelta
from datetime import timezone

import mailtrap as mt
from mailtrap.models.common import DeletedObject
from mailtrap.models.email_campaigns import EmailCampaign
from mailtrap.models.email_campaigns import EmailCampaignListParams
from mailtrap.models.email_campaigns import EmailCampaignListResponse
from mailtrap.models.email_campaigns import EmailCampaignStats
from mailtrap.models.email_campaigns import EmailCampaignStatsParams

API_TOKEN = "YOUR_API_TOKEN"
DOMAIN_ID = 4321

# The Email Campaigns API is token-scoped — no `account_id` is needed.
client = mt.MailtrapClient(token=API_TOKEN)
email_campaigns_api = client.email_campaigns_api.email_campaigns


def list_email_campaigns() -> EmailCampaignListResponse:
# `search` filters by name; `token` is the page number (page-token
# pagination); `per_page` caps at 100 (default 50).
return email_campaigns_api.get_list(
EmailCampaignListParams(per_page=50, search="Spring", token=1)
)


def get_email_campaign(email_campaign_id: int) -> EmailCampaign:
return email_campaigns_api.get_by_id(email_campaign_id=email_campaign_id)


def create_email_campaign() -> EmailCampaign:
# A campaign is created in the `draft` state and must reference a verified
# sending domain via `domain_id` (as returned by the Sending Domains
# endpoints).
return email_campaigns_api.create(
mt.CreateEmailCampaignParams(
name="Spring Sale",
domain_id=DOMAIN_ID,
from_display_name="Acme Marketing",
from_local_part="news",
reply_to=mt.EmailCampaignReplyTo(
display_name="Acme Support",
local_part="support",
domain="acme.com",
),
template_attributes=mt.CreateEmailCampaignTemplateAttributes(
subject="Spring is here — 30% off"
),
)
)


def update_email_campaign(email_campaign_id: int) -> EmailCampaign:
# Only supplied fields are changed. The campaign's template is edited in
# place — pass only the `template_attributes` sub-fields you want changed.
return email_campaigns_api.update(
email_campaign_id=email_campaign_id,
campaign_params=mt.UpdateEmailCampaignParams(
name="Spring Sale (updated)",
delivery_mode="gradual",
delivery_options=mt.EmailCampaignDeliveryOptions(emails_per_hour=1000),
contact_list_ids=[55, 56],
contact_segment_ids=[12],
template_attributes=mt.EmailCampaignTemplateAttributes(
subject="Spring is here — 30% off everything",
body_html=(
"<html><body>"
"<h1>Hi {{first_name}}!</h1>"
'<p><a href="__unsubscribe_url__">Unsubscribe</a></p>'
"</body></html>"
),
merge_tags=["first_name"],
),
),
)


def schedule_email_campaign(email_campaign_id: int) -> EmailCampaign:
# The campaign must be a `draft`; the time must be in the future (at most
# 1 month ahead) and comes back in `current_state_metadata.scheduled_at`.
send_at = datetime.now(timezone.utc) + timedelta(days=1)
return email_campaigns_api.schedule(
email_campaign_id=email_campaign_id,
schedule_params=mt.ScheduleEmailCampaignParams(
datetime=send_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
)


def cancel_email_campaign(email_campaign_id: int) -> EmailCampaign:
# Cancels a `scheduled` campaign, returning it to `draft`.
return email_campaigns_api.cancel(email_campaign_id=email_campaign_id)


def start_email_campaign(email_campaign_id: int) -> EmailCampaign:
# Starts sending a `draft` campaign immediately.
return email_campaigns_api.start(email_campaign_id=email_campaign_id)


def terminate_email_campaign(email_campaign_id: int) -> EmailCampaign:
# Aborts a campaign that is currently sending.
return email_campaigns_api.terminate(email_campaign_id=email_campaign_id)


def reset_email_campaign(email_campaign_id: int) -> EmailCampaign:
# Resets a `scheduled` campaign back to `draft`.
return email_campaigns_api.reset(email_campaign_id=email_campaign_id)


def get_email_campaign_stats(email_campaign_id: int) -> EmailCampaignStats:
today = datetime.now(timezone.utc).date()
return email_campaigns_api.get_stats(
email_campaign_id=email_campaign_id,
params=EmailCampaignStatsParams(
start_date=(today - timedelta(days=30)).isoformat(),
end_date=today.isoformat(),
),
)


def delete_email_campaign(email_campaign_id: int) -> DeletedObject:
# The API responds with 204 No Content.
return email_campaigns_api.delete(email_campaign_id=email_campaign_id)


if __name__ == "__main__":
listed = list_email_campaigns()
print(listed.data)
print(listed.pagination)

created = create_email_campaign()
print(created)

fetched = get_email_campaign(created.id)
print(fetched)

updated = update_email_campaign(created.id)
print(updated)

scheduled = schedule_email_campaign(created.id)
print(scheduled.current_state_metadata)

cancelled = cancel_email_campaign(created.id)
print(cancelled.current_state)

started = start_email_campaign(created.id)
print(started.current_state)

stats = get_email_campaign_stats(created.id)
print(stats)

# Only a campaign in the `draft` state can be deleted, and a campaign that
# has been started can never return to `draft` — so delete a fresh draft
# rather than the one started above.
throwaway = create_email_campaign()
deleted = delete_email_campaign(throwaway.id)
print(deleted)
13 changes: 13 additions & 0 deletions mailtrap/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,19 @@
from .models.contacts import ImportContactParams
from .models.contacts import UpdateContactFieldParams
from .models.contacts import UpdateContactParams
from .models.email_campaigns import CreateEmailCampaignParams
from .models.email_campaigns import CreateEmailCampaignTemplateAttributes
from .models.email_campaigns import EmailCampaign
from .models.email_campaigns import EmailCampaignDeliveryOptions
from .models.email_campaigns import EmailCampaignListParams
from .models.email_campaigns import EmailCampaignListResponse
from .models.email_campaigns import EmailCampaignReplyTo
from .models.email_campaigns import EmailCampaignStats
from .models.email_campaigns import EmailCampaignStatsParams
from .models.email_campaigns import EmailCampaignTemplate
from .models.email_campaigns import EmailCampaignTemplateAttributes
from .models.email_campaigns import ScheduleEmailCampaignParams
from .models.email_campaigns import UpdateEmailCampaignParams
from .models.email_logs import EmailLogMessage
from .models.email_logs import EmailLogsListFilters
from .models.email_logs import EmailLogsListResponse
Expand Down
11 changes: 11 additions & 0 deletions mailtrap/api/email_campaigns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
from mailtrap.api.resources.email_campaigns import EmailCampaignsApi
from mailtrap.http import HttpClient


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

@property
def email_campaigns(self) -> EmailCampaignsApi:
return EmailCampaignsApi(client=self._client)
135 changes: 135 additions & 0 deletions mailtrap/api/resources/email_campaigns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
from typing import Optional

from mailtrap.http import HttpClient
from mailtrap.models.common import DeletedObject
from mailtrap.models.email_campaigns import CreateEmailCampaignParams
from mailtrap.models.email_campaigns import EmailCampaign
from mailtrap.models.email_campaigns import EmailCampaignListParams
from mailtrap.models.email_campaigns import EmailCampaignListResponse
from mailtrap.models.email_campaigns import EmailCampaignResponse
from mailtrap.models.email_campaigns import EmailCampaignStats
from mailtrap.models.email_campaigns import EmailCampaignStatsParams
from mailtrap.models.email_campaigns import EmailCampaignStatsResponse
from mailtrap.models.email_campaigns import ScheduleEmailCampaignParams
from mailtrap.models.email_campaigns import UpdateEmailCampaignParams


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

def get_list(
self, params: Optional[EmailCampaignListParams] = None
) -> EmailCampaignListResponse:
"""
List email campaigns for the account, newest first. ``params`` filters
by name and paginates the result; omit it for the first page with API
defaults.
"""
query_params = params.api_query_params if params else None
response = self._client.get(self._api_path(), params=query_params or None)
return EmailCampaignListResponse(**response)

def get_by_id(self, email_campaign_id: int) -> EmailCampaign:
"""
Get a single email campaign by id.
"""
response = self._client.get(self._api_path(email_campaign_id))
return EmailCampaignResponse(**response).data

def create(self, campaign_params: CreateEmailCampaignParams) -> EmailCampaign:
"""
Create a new email campaign in the ``draft`` state. The campaign must
reference an existing sending domain via ``domain_id`` and
include a template ``subject`` within ``template_attributes``.
"""
response = self._client.post(self._api_path(), json=campaign_params.api_data)
return EmailCampaignResponse(**response).data

def update(
self, email_campaign_id: int, campaign_params: UpdateEmailCampaignParams
) -> EmailCampaign:
"""
Update an existing ``draft`` email campaign. Only the fields supplied
in ``campaign_params`` are sent to the API.
"""
response = self._client.patch(
self._api_path(email_campaign_id),
json=campaign_params.api_data,
)
return EmailCampaignResponse(**response).data

def delete(self, email_campaign_id: int) -> DeletedObject:
"""
Delete an email campaign. Only a campaign in the ``draft`` state can be
deleted.
"""
self._client.delete(self._api_path(email_campaign_id))
return DeletedObject(email_campaign_id)

def start(self, email_campaign_id: int) -> EmailCampaign:
"""
Start sending a ``draft`` campaign immediately.
"""
return self._action(email_campaign_id, "start")

def schedule(
self, email_campaign_id: int, schedule_params: ScheduleEmailCampaignParams
) -> EmailCampaign:
"""
Schedule a ``draft`` campaign to start sending at a future time. The
time is reported back in ``current_state_metadata.scheduled_at``.
"""
response = self._client.post(
f"{self._api_path(email_campaign_id)}/schedule",
json=schedule_params.api_data,
)
return EmailCampaignResponse(**response).data

def cancel(self, email_campaign_id: int) -> EmailCampaign:
"""
Cancel a ``scheduled`` campaign, returning it to the ``draft`` state.
"""
return self._action(email_campaign_id, "cancel")

def terminate(self, email_campaign_id: int) -> EmailCampaign:
"""
Terminate a campaign that is currently sending (``started``,
``queued``, or ``paused``), aborting the in-flight send.
"""
return self._action(email_campaign_id, "terminate")

def reset(self, email_campaign_id: int) -> EmailCampaign:
"""
Reset a ``scheduled`` campaign back to the ``draft`` state.
"""
return self._action(email_campaign_id, "reset")

def get_stats(
self,
email_campaign_id: int,
params: Optional[EmailCampaignStatsParams] = None,
) -> EmailCampaignStats:
"""
Get aggregated performance statistics for a single campaign. If the
campaign has never been started, all counts and rates are ``0``.
``params`` narrows the aggregation window; omit it to cover the whole
period since the campaign was last started.
"""
query_params = params.api_query_params if params else None
response = self._client.get(
f"{self._api_path(email_campaign_id)}/stats", params=query_params or None
)
return EmailCampaignStatsResponse(**response).data

def _action(self, email_campaign_id: int, action: str) -> EmailCampaign:
response = self._client.post(f"{self._api_path(email_campaign_id)}/{action}")
return EmailCampaignResponse(**response).data

def _api_path(self, email_campaign_id: Optional[int] = None) -> str:
# Token-scoped: the account is resolved from the API token server-side,
# so the path takes no account id.
path = "/api/email_campaigns"
if email_campaign_id is not None:
return f"{path}/{email_campaign_id}"
return path
9 changes: 9 additions & 0 deletions mailtrap/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from pydantic import TypeAdapter

from mailtrap.api.contacts import ContactsBaseApi
from mailtrap.api.email_campaigns import EmailCampaignsBaseApi
from mailtrap.api.email_logs import EmailLogsBaseApi
from mailtrap.api.general import GeneralApi
from mailtrap.api.inbound import InboundBaseApi
Expand Down Expand Up @@ -124,6 +125,14 @@ def sending_domains_api(self) -> SendingDomainsBaseApi:
client=HttpClient(host=GENERAL_HOST, headers=self.headers),
)

@property
def email_campaigns_api(self) -> EmailCampaignsBaseApi:
# Token-scoped (`/api/email_campaigns`) — the account is resolved
# server-side from the token, so no `account_id` is required.
return EmailCampaignsBaseApi(
client=HttpClient(host=GENERAL_HOST, headers=self.headers),
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

@property
def email_logs_api(self) -> EmailLogsBaseApi:
self._validate_account_id("Email Logs API")
Expand Down
14 changes: 14 additions & 0 deletions mailtrap/models/common.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from typing import Any
from typing import Optional
from typing import TypeVar
from typing import Union
from typing import cast
Expand Down Expand Up @@ -32,3 +33,16 @@ def api_query_params(self: T) -> dict[str, Any]:
@dataclass
class DeletedObject:
id: Union[int, str]


@dataclass
class Pagination:
"""Page-token pagination metadata returned with a paginated list response."""

token: Optional[int] = None
prev_token: Optional[int] = None
next_token: Optional[int] = None
first_url: Optional[str] = None
prev_url: Optional[str] = None
current_url: Optional[str] = None
next_url: Optional[str] = None
Loading
Loading