Skip to content
Draft
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 @@ -244,6 +244,9 @@ The same situation applies to both `client.batch_send()` and `client.sending_api
### Sending Domains API:
- Sending Domains – [`sending_domains/sending_domains.py`](examples/sending_domains/sending_domains.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
148 changes: 148 additions & 0 deletions examples/email_campaigns/email_campaigns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
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 EmailCampaignListResponse
from mailtrap.models.email_campaigns import EmailCampaignStats

API_TOKEN = "YOUR_API_TOKEN"
MAILSEND_DOMAIN_ID = "d2313359-acb4-4b87-bce6-f5774f6a1e37"

# 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(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 `mailsend_domain_id` (a UUID string).
return email_campaigns_api.create(
mt.CreateEmailCampaignParams(
name="Spring Sale",
mailsend_domain_id=MAILSEND_DOMAIN_ID,
from_display_name="Acme Marketing",
from_local_part="news",
reply_to=mt.ReplyTo(
display_name="Acme Support",
local_part="support",
domain="acme.com",
),
template_attributes=mt.CreateTemplateAttributes(
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.DeliveryOptions(emails_per_hour=1000),
contact_list_ids=[55, 56],
contact_segment_ids=[12],
template_attributes=mt.TemplateAttributes(
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,
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)

deleted = delete_email_campaign(created.id)
print(deleted)
11 changes: 11 additions & 0 deletions mailtrap/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,17 @@
from .models.contacts import ImportContactParams
from .models.contacts import UpdateContactFieldParams
from .models.contacts import UpdateContactParams
from .models.email_campaigns import CampaignTemplate
from .models.email_campaigns import CreateEmailCampaignParams
from .models.email_campaigns import CreateTemplateAttributes
from .models.email_campaigns import DeliveryOptions
from .models.email_campaigns import EmailCampaign
from .models.email_campaigns import EmailCampaignListResponse
from .models.email_campaigns import EmailCampaignStats
from .models.email_campaigns import ReplyTo
from .models.email_campaigns import ScheduleEmailCampaignParams
from .models.email_campaigns import TemplateAttributes
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)
142 changes: 142 additions & 0 deletions mailtrap/api/resources/email_campaigns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
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,
per_page: Optional[int] = None,
search: Optional[str] = None,
token: Optional[int] = None,
) -> EmailCampaignListResponse:
"""
List email campaigns for the account, newest first. ``search`` filters
by name, ``per_page`` sets the page size (max 100, default 50), and
``token`` is the page number to retrieve (default 1).
"""
params = EmailCampaignListParams(
per_page=per_page, search=search, token=token
).api_query_params
response = self._client.get(self._api_path(), params=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 ``mailsend_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. The campaign must not be in a sending state.
"""
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,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
) -> EmailCampaignStats:
"""
Get aggregated performance statistics for a single campaign. If the
campaign has never been started, all counts and rates are ``0``.
``start_date``/``end_date`` (``YYYY-MM-DD``) narrow the aggregation
window; it defaults to the whole period since the last start.
"""
params = EmailCampaignStatsParams(
start_date=start_date, end_date=end_date
).api_query_params
response = self._client.get(
f"{self._api_path(email_campaign_id)}/stats", params=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:
# The Email Campaigns endpoint is token-scoped, NOT account-scoped:
# the account is resolved from the API token server-side.
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.organizations import OrganizationsBaseApi
Expand Down Expand Up @@ -117,6 +118,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
Loading
Loading