-
Notifications
You must be signed in to change notification settings - Fork 8
Add support for Email Logs API #64
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| """Example: List email logs and get a single message by ID.""" | ||
|
|
||
| from datetime import datetime | ||
| from datetime import timedelta | ||
| from datetime import timezone | ||
|
|
||
| import mailtrap as mt | ||
| from mailtrap.models.email_logs import EmailLogsListFilters | ||
| from mailtrap.models.email_logs import filter_ci_equal | ||
| from mailtrap.models.email_logs import filter_string_equal | ||
| from mailtrap.models.email_logs import filter_string_not_empty | ||
|
|
||
| API_TOKEN = "YOUR_API_TOKEN" | ||
| ACCOUNT_ID = "YOUR_ACCOUNT_ID" | ||
|
|
||
| client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID) | ||
| email_logs_api = client.email_logs_api.email_logs | ||
|
|
||
|
|
||
| def list_email_logs(): | ||
| """List email logs (first page).""" | ||
| return email_logs_api.get_list() | ||
|
|
||
|
|
||
| def list_email_logs_with_filters(): | ||
| """List email logs from last 2 days, by category(s), with non-empty subject.""" | ||
| now = datetime.now(timezone.utc) | ||
| two_days_ago = now - timedelta(days=2) | ||
| filters = EmailLogsListFilters( | ||
| sent_after=two_days_ago.isoformat().replace("+00:00", "Z"), | ||
| sent_before=now.isoformat().replace("+00:00", "Z"), | ||
| subject=filter_string_not_empty(), | ||
| to=filter_ci_equal("recipient@example.com"), | ||
| category=filter_string_equal(["Welcome Email", "Password Reset"]), | ||
| ) | ||
| return email_logs_api.get_list(filters=filters) | ||
|
|
||
|
|
||
| def get_next_page(previous_response): | ||
| """Fetch next page using cursor from previous response.""" | ||
| if previous_response.next_page_cursor is None: | ||
| return None | ||
| return email_logs_api.get_list(search_after=previous_response.next_page_cursor) | ||
|
|
||
|
|
||
| def get_message(message_id: str): | ||
| """Get a single email log message by UUID.""" | ||
| return email_logs_api.get_by_id(message_id) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| # List first page | ||
| response = list_email_logs() | ||
| print(f"Total: {response.total_count}, messages: {len(response.messages)}") | ||
| for msg in response.messages: | ||
| print(f" {msg.message_id} | {msg.from_} -> {msg.to} | {msg.status}") | ||
|
|
||
| # Get single message | ||
| if response.messages: | ||
| detail = get_message(response.messages[0].message_id) | ||
| print(f"Detail: {detail.subject}, events: {len(detail.events)}") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| from mailtrap.api.resources.email_logs import EmailLogsApi | ||
| from mailtrap.http import HttpClient | ||
|
|
||
|
|
||
| class EmailLogsBaseApi: | ||
| def __init__(self, client: HttpClient, account_id: str) -> None: | ||
| self._account_id = account_id | ||
| self._client = client | ||
|
|
||
| @property | ||
| def email_logs(self) -> EmailLogsApi: | ||
| return EmailLogsApi(client=self._client, account_id=self._account_id) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| """Email Logs API resource – list and get email sending logs.""" | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Replace EN DASH in module docstring to satisfy Ruff RUF002. Line 1 uses 🧰 Tools🪛 Ruff (0.15.6)[warning] 1-1: Docstring contains ambiguous (RUF002) 🤖 Prompt for AI Agents |
||
|
|
||
| from typing import Optional | ||
|
|
||
| from mailtrap.http import HttpClient | ||
| from mailtrap.models.email_logs import EmailLogMessage | ||
| from mailtrap.models.email_logs import EmailLogsListFilters | ||
| from mailtrap.models.email_logs import EmailLogsListResponse | ||
|
|
||
|
|
||
| class EmailLogsApi: | ||
| def __init__(self, client: HttpClient, account_id: str) -> None: | ||
| self._account_id = account_id | ||
| self._client = client | ||
|
|
||
| def get_list( | ||
| self, | ||
| filters: Optional[EmailLogsListFilters] = None, | ||
| search_after: Optional[str] = None, | ||
| ) -> EmailLogsListResponse: | ||
| """ | ||
| List email logs (paginated). Results are ordered by sent_at descending. | ||
| Use search_after with next_page_cursor from the previous response for | ||
| the next page. | ||
| """ | ||
| params: dict[str, object] = {} | ||
| if filters is not None: | ||
| params.update(filters.to_params()) | ||
| if search_after is not None: | ||
| params["search_after"] = search_after | ||
| response = self._client.get(self._api_path(), params=params or None) | ||
| messages = [EmailLogMessage.from_api(msg) for msg in response.get("messages", [])] | ||
| return EmailLogsListResponse( | ||
| messages=messages, | ||
| total_count=response.get("total_count", 0), | ||
| next_page_cursor=response.get("next_page_cursor"), | ||
| ) | ||
|
Comment on lines
+31
to
+37
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Guard response shape before dict access/model parsing.
🛡️ Proposed fix-"""Email Logs API resource – list and get email sending logs."""
+"""Email Logs API resource - list and get email sending logs."""
-from typing import Optional
+from typing import Any
+from typing import Optional
@@
class EmailLogsApi:
@@
+ def _expect_dict_response(self, response: Any, operation: str) -> dict[str, Any]:
+ if not isinstance(response, dict):
+ raise TypeError(
+ f"Unexpected response type for {operation}: {type(response).__name__}"
+ )
+ return response
+
def get_list(
@@
- response = self._client.get(self._api_path(), params=params or None)
+ response = self._expect_dict_response(
+ self._client.get(self._api_path(), params=params or None),
+ "email logs list",
+ )
messages = [EmailLogMessage.from_api(msg) for msg in response.get("messages", [])]
@@
def get_by_id(self, sending_message_id: str) -> EmailLogMessage:
"""Get a single email log message by its UUID."""
- response = self._client.get(self._api_path(sending_message_id))
+ response = self._expect_dict_response(
+ self._client.get(self._api_path(sending_message_id)),
+ "email log by id",
+ )
return EmailLogMessage.from_api(response)Also applies to: 41-42 🤖 Prompt for AI Agents |
||
|
|
||
| def get_by_id(self, sending_message_id: str) -> EmailLogMessage: | ||
| """Get a single email log message by its UUID.""" | ||
| response = self._client.get(self._api_path(sending_message_id)) | ||
| return EmailLogMessage.from_api(response) | ||
|
|
||
| def _api_path(self, sending_message_id: Optional[str] = None) -> str: | ||
| path = f"/api/accounts/{self._account_id}/email_logs" | ||
| if sending_message_id is not None: | ||
| path = f"{path}/{sending_message_id}" | ||
| return path | ||
Uh oh!
There was an error while loading. Please reload this page.