-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathclient.py
More file actions
277 lines (239 loc) · 11.6 KB
/
Copy pathclient.py
File metadata and controls
277 lines (239 loc) · 11.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
"""
Stack0 Mail Client
Complete email API with support for transactional emails, campaigns, sequences, and more
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from .audiences import AsyncAudiences, Audiences
from .campaigns import AsyncCampaigns, Campaigns
from .contacts import AsyncContacts, Contacts
from .domains import AsyncDomains, Domains
from .events import AsyncEvents, Events
from .sequences import AsyncSequences, Sequences
from .templates import AsyncTemplates, Templates
from .types import (
CancelEmailResponse,
EmailAnalyticsResponse,
GetEmailResponse,
HourlyAnalyticsResponse,
ListEmailsRequest,
ListEmailsResponse,
ListSendersRequest,
ListSendersResponse,
ResendEmailResponse,
SendBatchEmailRequest,
SendBatchEmailResponse,
SendBroadcastEmailRequest,
SendBroadcastEmailResponse,
SendEmailRequest,
SendEmailResponse,
TimeSeriesAnalyticsRequest,
TimeSeriesAnalyticsResponse,
)
if TYPE_CHECKING:
from .._client.http import AsyncHTTPClient, HTTPClient
class Mail:
"""Synchronous mail client"""
def __init__(self, http: HTTPClient) -> None:
self._http = http
self.domains = Domains(http)
self.templates = Templates(http)
self.audiences = Audiences(http)
self.contacts = Contacts(http)
self.campaigns = Campaigns(http)
self.sequences = Sequences(http)
self.events = Events(http)
def send(self, request: SendEmailRequest) -> SendEmailResponse:
"""Send a single email"""
data = self._http.post("/mail/send", request.model_dump(by_alias=True, exclude_none=True))
return SendEmailResponse.model_validate(data)
def send_batch(self, request: SendBatchEmailRequest) -> SendBatchEmailResponse:
"""Send multiple emails in a batch (up to 100)"""
data = self._http.post("/mail/send/batch", request.model_dump(by_alias=True, exclude_none=True))
return SendBatchEmailResponse.model_validate(data)
def send_broadcast(self, request: SendBroadcastEmailRequest) -> SendBroadcastEmailResponse:
"""Send a broadcast email (same content to multiple recipients, up to 1000)"""
data = self._http.post("/mail/send/broadcast", request.model_dump(by_alias=True, exclude_none=True))
return SendBroadcastEmailResponse.model_validate(data)
def get(self, id: str) -> GetEmailResponse:
"""Get email details by ID"""
data = self._http.get(f"/mail/{id}")
return GetEmailResponse.model_validate(data)
def list(self, request: ListEmailsRequest | None = None) -> ListEmailsResponse:
"""List emails with optional filters"""
params: list[str] = []
if request:
if request.project_slug:
params.append(f"projectSlug={request.project_slug}")
if request.environment:
params.append(f"environment={request.environment}")
if request.limit:
params.append(f"limit={request.limit}")
if request.offset:
params.append(f"offset={request.offset}")
if request.status:
params.append(f"status={request.status}")
if request.from_:
params.append(f"from={request.from_}")
if request.to:
params.append(f"to={request.to}")
if request.subject:
params.append(f"subject={request.subject}")
if request.tag:
params.append(f"tag={request.tag}")
if request.start_date:
date_str = request.start_date.isoformat() if isinstance(request.start_date, datetime) else request.start_date
params.append(f"startDate={date_str}")
if request.end_date:
date_str = request.end_date.isoformat() if isinstance(request.end_date, datetime) else request.end_date
params.append(f"endDate={date_str}")
if request.sort_by:
params.append(f"sortBy={request.sort_by}")
if request.sort_order:
params.append(f"sortOrder={request.sort_order}")
query = "&".join(params)
path = f"/mail{'?' + query if query else ''}"
data = self._http.get(path)
return ListEmailsResponse.model_validate(data)
def resend(self, id: str) -> ResendEmailResponse:
"""Resend a previously sent email"""
data = self._http.post(f"/mail/{id}/resend", {})
return ResendEmailResponse.model_validate(data)
def cancel(self, id: str) -> CancelEmailResponse:
"""Cancel a scheduled email"""
data = self._http.post(f"/mail/{id}/cancel", {})
return CancelEmailResponse.model_validate(data)
def get_analytics(self) -> EmailAnalyticsResponse:
"""Get overall email analytics"""
data = self._http.get("/mail/analytics")
return EmailAnalyticsResponse.model_validate(data)
def get_time_series_analytics(self, request: TimeSeriesAnalyticsRequest | None = None) -> TimeSeriesAnalyticsResponse:
"""Get time series analytics (daily breakdown)"""
params: list[str] = []
if request and request.days:
params.append(f"days={request.days}")
query = "&".join(params)
path = f"/mail/analytics/timeseries{'?' + query if query else ''}"
data = self._http.get(path)
return TimeSeriesAnalyticsResponse.model_validate(data)
def get_hourly_analytics(self) -> HourlyAnalyticsResponse:
"""Get hourly analytics"""
data = self._http.get("/mail/analytics/hourly")
return HourlyAnalyticsResponse.model_validate(data)
def list_senders(self, request: ListSendersRequest | None = None) -> ListSendersResponse:
"""List unique senders with their statistics"""
params: list[str] = []
if request:
if request.project_slug:
params.append(f"projectSlug={request.project_slug}")
if request.environment:
params.append(f"environment={request.environment}")
if request.search:
params.append(f"search={request.search}")
query = "&".join(params)
path = f"/mail/senders{'?' + query if query else ''}"
data = self._http.get(path)
return ListSendersResponse.model_validate(data)
class AsyncMail:
"""Asynchronous mail client"""
def __init__(self, http: AsyncHTTPClient) -> None:
self._http = http
self.domains = AsyncDomains(http)
self.templates = AsyncTemplates(http)
self.audiences = AsyncAudiences(http)
self.contacts = AsyncContacts(http)
self.campaigns = AsyncCampaigns(http)
self.sequences = AsyncSequences(http)
self.events = AsyncEvents(http)
async def send(self, request: SendEmailRequest) -> SendEmailResponse:
"""Send a single email"""
data = await self._http.post("/mail/send", request.model_dump(by_alias=True, exclude_none=True))
return SendEmailResponse.model_validate(data)
async def send_batch(self, request: SendBatchEmailRequest) -> SendBatchEmailResponse:
"""Send multiple emails in a batch (up to 100)"""
data = await self._http.post("/mail/send/batch", request.model_dump(by_alias=True, exclude_none=True))
return SendBatchEmailResponse.model_validate(data)
async def send_broadcast(self, request: SendBroadcastEmailRequest) -> SendBroadcastEmailResponse:
"""Send a broadcast email (same content to multiple recipients, up to 1000)"""
data = await self._http.post("/mail/send/broadcast", request.model_dump(by_alias=True, exclude_none=True))
return SendBroadcastEmailResponse.model_validate(data)
async def get(self, id: str) -> GetEmailResponse:
"""Get email details by ID"""
data = await self._http.get(f"/mail/{id}")
return GetEmailResponse.model_validate(data)
async def list(self, request: ListEmailsRequest | None = None) -> ListEmailsResponse:
"""List emails with optional filters"""
params: list[str] = []
if request:
if request.project_slug:
params.append(f"projectSlug={request.project_slug}")
if request.environment:
params.append(f"environment={request.environment}")
if request.limit:
params.append(f"limit={request.limit}")
if request.offset:
params.append(f"offset={request.offset}")
if request.status:
params.append(f"status={request.status}")
if request.from_:
params.append(f"from={request.from_}")
if request.to:
params.append(f"to={request.to}")
if request.subject:
params.append(f"subject={request.subject}")
if request.tag:
params.append(f"tag={request.tag}")
if request.start_date:
date_str = request.start_date.isoformat() if isinstance(request.start_date, datetime) else request.start_date
params.append(f"startDate={date_str}")
if request.end_date:
date_str = request.end_date.isoformat() if isinstance(request.end_date, datetime) else request.end_date
params.append(f"endDate={date_str}")
if request.sort_by:
params.append(f"sortBy={request.sort_by}")
if request.sort_order:
params.append(f"sortOrder={request.sort_order}")
query = "&".join(params)
path = f"/mail{'?' + query if query else ''}"
data = await self._http.get(path)
return ListEmailsResponse.model_validate(data)
async def resend(self, id: str) -> ResendEmailResponse:
"""Resend a previously sent email"""
data = await self._http.post(f"/mail/{id}/resend", {})
return ResendEmailResponse.model_validate(data)
async def cancel(self, id: str) -> CancelEmailResponse:
"""Cancel a scheduled email"""
data = await self._http.post(f"/mail/{id}/cancel", {})
return CancelEmailResponse.model_validate(data)
async def get_analytics(self) -> EmailAnalyticsResponse:
"""Get overall email analytics"""
data = await self._http.get("/mail/analytics")
return EmailAnalyticsResponse.model_validate(data)
async def get_time_series_analytics(self, request: TimeSeriesAnalyticsRequest | None = None) -> TimeSeriesAnalyticsResponse:
"""Get time series analytics (daily breakdown)"""
params: list[str] = []
if request and request.days:
params.append(f"days={request.days}")
query = "&".join(params)
path = f"/mail/analytics/timeseries{'?' + query if query else ''}"
data = await self._http.get(path)
return TimeSeriesAnalyticsResponse.model_validate(data)
async def get_hourly_analytics(self) -> HourlyAnalyticsResponse:
"""Get hourly analytics"""
data = await self._http.get("/mail/analytics/hourly")
return HourlyAnalyticsResponse.model_validate(data)
async def list_senders(self, request: ListSendersRequest | None = None) -> ListSendersResponse:
"""List unique senders with their statistics"""
params: list[str] = []
if request:
if request.project_slug:
params.append(f"projectSlug={request.project_slug}")
if request.environment:
params.append(f"environment={request.environment}")
if request.search:
params.append(f"search={request.search}")
query = "&".join(params)
path = f"/mail/senders{'?' + query if query else ''}"
data = await self._http.get(path)
return ListSendersResponse.model_validate(data)