Add support for Email Logs API

Author: mklocek

README.md

@@ -247,6 +247,9 @@ The same situation applies to both `client.batch_send()` and `client.sending_api
 ### Suppressions API:
 - Suppressions (find & delete) – [`suppressions/suppressions.py`](examples/suppressions/suppressions.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)
+
 ### General API:
 - Account Accesses management – [`general/account_accesses.py`](examples/general/account_accesses.py)
 - Accounts info – [`general/accounts.py`](examples/general/accounts.py)

examples/email_logs/email_logs.py

@@ -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_string_equal
+from mailtrap.models.email_logs import filter_string_not_empty
+from mailtrap.models.email_logs import filter_ci_equal
+
+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)}")

mailtrap/__init__.py

@@ -15,6 +15,9 @@
 from .models.contacts import ImportContactParams
 from .models.contacts import UpdateContactFieldParams
 from .models.contacts import UpdateContactParams
+from .models.email_logs import EmailLogMessage
+from .models.email_logs import EmailLogsListFilters
+from .models.email_logs import EmailLogsListResponse
 from .models.inboxes import CreateInboxParams
 from .models.inboxes import UpdateInboxParams
 from .models.mail import Address

mailtrap/api/email_logs.py

@@ -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)

mailtrap/api/resources/email_logs.py

@@ -0,0 +1,48 @@
+"""Email Logs API resource – list and get email sending logs."""
+
+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"),
+        )
+
+    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

mailtrap/client.py

@@ -6,6 +6,7 @@
 from pydantic import TypeAdapter
 
 from mailtrap.api.contacts import ContactsBaseApi
+from mailtrap.api.email_logs import EmailLogsBaseApi
 from mailtrap.api.general import GeneralApi
 from mailtrap.api.sending import SendingApi
 from mailtrap.api.sending_domains import SendingDomainsBaseApi
@@ -72,36 +73,44 @@ def testing_api(self) -> TestingApi:
 
     @property
     def email_templates_api(self) -> EmailTemplatesApi:
-        self._validate_account_id()
+        self._validate_account_id("Email Templates API")
         return EmailTemplatesApi(
             account_id=cast(str, self.account_id),
             client=HttpClient(host=GENERAL_HOST, headers=self.headers),
         )
 
     @property
     def contacts_api(self) -> ContactsBaseApi:
-        self._validate_account_id()
+        self._validate_account_id("Contacts API")
         return ContactsBaseApi(
             account_id=cast(str, self.account_id),
             client=HttpClient(host=GENERAL_HOST, headers=self.headers),
         )
 
     @property
     def suppressions_api(self) -> SuppressionsBaseApi:
-        self._validate_account_id()
+        self._validate_account_id("Suppressions API")
         return SuppressionsBaseApi(
             account_id=cast(str, self.account_id),
             client=HttpClient(host=GENERAL_HOST, headers=self.headers),
         )
 
     @property
     def sending_domains_api(self) -> SendingDomainsBaseApi:
-        self._validate_account_id()
+        self._validate_account_id("Sending Domains API")
         return SendingDomainsBaseApi(
             account_id=cast(str, self.account_id),
             client=HttpClient(host=GENERAL_HOST, headers=self.headers),
         )
 
+    @property
+    def email_logs_api(self) -> EmailLogsBaseApi:
+        self._validate_account_id("Email Logs API")
+        return EmailLogsBaseApi(
+            account_id=cast(str, self.account_id),
+            client=HttpClient(host=GENERAL_HOST, headers=self.headers),
+        )
+
     @property
     def sending_api(self) -> SendingApi:
         http_client = HttpClient(host=self._sending_api_host, headers=self.headers)
@@ -162,9 +171,9 @@ def _sending_api_host(self) -> str:
             return BULK_HOST
         return SENDING_HOST
 
-    def _validate_account_id(self) -> None:
+    def _validate_account_id(self, api_name: str = "Testing API") -> None:
         if not self.account_id:
-            raise ClientConfigurationError("`account_id` is required for Testing API")
+            raise ClientConfigurationError(f"`account_id` is required for {api_name}")
 
     def _validate_itself(self) -> None:
         if self.sandbox and not self.inbox_id: