From 6b5ce554bf79d361aba4495d39c07d2cd2aa642d Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 11:00:21 +0200 Subject: [PATCH 1/6] Add inbound email models Pydantic models for the Inbound Email API: folders, inboxes, messages (list/details + attachments), threads (summary/detail/messages), send result, and create/update params. Reply and forward params share a private base and reuse Address/Attachment from the mail models. --- mailtrap/__init__.py | 6 + mailtrap/models/inbound.py | 237 +++++++++++++++++++++++++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 mailtrap/models/inbound.py diff --git a/mailtrap/__init__.py b/mailtrap/__init__.py index ca9c7d6..bf32a5b 100644 --- a/mailtrap/__init__.py +++ b/mailtrap/__init__.py @@ -20,6 +20,12 @@ from .models.email_logs import EmailLogMessage from .models.email_logs import EmailLogsListFilters from .models.email_logs import EmailLogsListResponse +from .models.inbound import CreateInboundFolderParams +from .models.inbound import CreateInboundInboxParams +from .models.inbound import ForwardInboundMessageParams +from .models.inbound import ReplyInboundMessageParams +from .models.inbound import UpdateInboundFolderParams +from .models.inbound import UpdateInboundInboxParams from .models.inboxes import CreateInboxParams from .models.inboxes import UpdateInboxParams from .models.mail import Address diff --git a/mailtrap/models/inbound.py b/mailtrap/models/inbound.py new file mode 100644 index 0000000..3d68a38 --- /dev/null +++ b/mailtrap/models/inbound.py @@ -0,0 +1,237 @@ +"""Models for the Inbound Email API (folders, inboxes, messages, threads).""" + +from typing import Any +from typing import Literal +from typing import Optional + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic.dataclasses import dataclass + +from mailtrap.models.common import RequestParams +from mailtrap.models.mail.address import Address +from mailtrap.models.mail.attachment import Attachment + +ContentDisposition = Literal["attachment", "inline"] + +# --- Attachments --- + + +@dataclass +class InboundAttachment: + """ + Attachment metadata on a received message. download_url and + download_url_expires_at are only populated on get-by-id and thread responses. + """ + + attachment_id: str + size: Optional[int] = None + filename: Optional[str] = None + content_type: Optional[str] = None + content_disposition: Optional[ContentDisposition] = None + content_id: Optional[str] = None + download_url: Optional[str] = None + download_url_expires_at: Optional[str] = None + + +# --- Folders --- + + +@dataclass +class InboundFolder: + id: int + name: str + + +# --- Inboxes --- + + +@dataclass +class InboundInbox: + id: int + name: str + address: str + domain_id: int + + +# --- Messages --- + + +class InboundMessage(BaseModel): + """A received message (list / summary shape).""" + + model_config = ConfigDict(populate_by_name=True) + + id: str + inbox_id: int + from_: Optional[str] = Field(default=None, alias="from") + to: list[str] = Field(default_factory=list) + cc: list[str] = Field(default_factory=list) + bcc: list[str] = Field(default_factory=list) + reply_to: Optional[str] = None + subject: Optional[str] = None + rfc_message_id: Optional[str] = None + in_reply_to: Optional[str] = None + references: list[str] = Field(default_factory=list) + headers: Optional[dict[str, str]] = None + size: Optional[int] = None + html_size: Optional[int] = None + text_size: Optional[int] = None + received_at: str + thread_id: Optional[str] = None + attachments: list[InboundAttachment] = Field(default_factory=list) + + +class InboundMessageDetails(InboundMessage): + """A received message with body and attachment download URLs (get-by-id).""" + + raw_message_url: Optional[str] = None + raw_message_expires_at: Optional[str] = None + html_body: Optional[str] = None + text_body: Optional[str] = None + + +@dataclass +class InboundMessagesListResponse: + """Paginated response from list messages (cursor via last_id).""" + + data: list[InboundMessage] = Field(default_factory=list) + total_count: int = 0 + last_id: Optional[str] = None + + +# --- Threads --- + + +class InboundThreadSummary(BaseModel): + """Thread overview (list shape and the head of a thread).""" + + id: str + subject: Optional[str] = None + message_count: int + size: int + first_message_at: str + last_received_at: Optional[str] = None + last_sent_at: Optional[str] = None + last_activity_at: str + last_message_id: Optional[str] = None + senders: list[str] = Field(default_factory=list) + recipients: list[str] = Field(default_factory=list) + attachments: list[InboundAttachment] = Field(default_factory=list) + + +class InboundThreadMessage(BaseModel): + """ + A message inside a thread. Only visibility_status and direction are + guaranteed; placeholder entries omit the rest. + """ + + model_config = ConfigDict(populate_by_name=True) + + visibility_status: Literal["available", "placeholder"] + direction: Literal["inbound", "outbound"] + id: Optional[str] = None + message_group_id: Optional[str] = None + subject: Optional[str] = None + rfc_message_id: Optional[str] = None + in_reply_to: Optional[str] = None + references: Optional[list[str]] = None + from_: Optional[str] = Field(default=None, alias="from") + to: Optional[list[str]] = None + cc: Optional[list[str]] = None + bcc: Optional[list[str]] = None + reply_to: Optional[str] = None + created_at: Optional[str] = None + email_size: Optional[int] = None + text_body: Optional[str] = None + html_body: Optional[str] = None + attachments: Optional[list[InboundAttachment]] = None + delivery_status: Optional[str] = None + delivered_at: Optional[str] = None + bounced_at: Optional[str] = None + + +class InboundThread(InboundThreadSummary): + """A thread with its messages embedded (oldest first).""" + + messages: list[InboundThreadMessage] = Field(default_factory=list) + + +@dataclass +class InboundThreadsListResponse: + """Paginated response from list threads (cursor via last_id).""" + + data: list[InboundThreadSummary] = Field(default_factory=list) + total_count: int = 0 + last_id: Optional[str] = None + + +@dataclass +class InboundSendResult: + """Result of a reply, reply-all, or forward (sends a real email).""" + + message_ids: list[str] = Field(default_factory=list) + + +# --- Request params --- + + +@dataclass +class CreateInboundFolderParams(RequestParams): + name: str + + +@dataclass +class UpdateInboundFolderParams(RequestParams): + name: str + + +@dataclass +class CreateInboundInboxParams(RequestParams): + """Omit domain_id for a Mailtrap-hosted inbox; pass it for a custom-domain inbox.""" + + name: str + domain_id: Optional[int] = None + + +@dataclass +class UpdateInboundInboxParams(RequestParams): + name: str + + +@dataclass +class _InboundMessageParams(RequestParams): + """ + Shared body for replying to and forwarding an inbound message. `sender` + (serialized as `from`) is rejected for Mailtrap-hosted inboxes and required + for custom-domain inboxes. + """ + + sender: Optional[Address] = Field(default=None, serialization_alias="from") + to: Optional[list[Address]] = None + cc: Optional[list[Address]] = None + bcc: Optional[list[Address]] = None + reply_to: Optional[Address] = None + text: Optional[str] = None + html: Optional[str] = None + category: Optional[str] = None + attachments: Optional[list[Attachment]] = None + headers: Optional[dict[str, str]] = None + custom_variables: Optional[dict[str, Any]] = None + + +@dataclass +class ReplyInboundMessageParams(_InboundMessageParams): + """Body for replying to (or reply-all-ing) an inbound message.""" + + +@dataclass +class ForwardInboundMessageParams(_InboundMessageParams): + """Forward an inbound message; requires at least one recipient in `to`.""" + + to: list[Address] = Field(default_factory=list) + + def __post_init__(self) -> None: + if not self.to: + raise ValueError("`to` must contain at least one recipient for forward") From 1277b3023cf05fa6b99aafa3fbbcba4bae35ece6 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 11:13:40 +0200 Subject: [PATCH 2/6] Add inbound email API classes and client wiring Token-scoped resource classes for inbound folders, inboxes, messages (list/get/delete + reply/reply_all/forward), and threads under /api/inbound, grouped by InboundBaseApi and exposed as client.inbound_api. Mirrors the general_api wiring: no account_id. --- mailtrap/api/inbound.py | 26 ++++++++ mailtrap/api/resources/inbound_folders.py | 43 +++++++++++++ mailtrap/api/resources/inbound_inboxes.py | 47 ++++++++++++++ mailtrap/api/resources/inbound_messages.py | 71 ++++++++++++++++++++++ mailtrap/api/resources/inbound_threads.py | 41 +++++++++++++ mailtrap/client.py | 7 +++ 6 files changed, 235 insertions(+) create mode 100644 mailtrap/api/inbound.py create mode 100644 mailtrap/api/resources/inbound_folders.py create mode 100644 mailtrap/api/resources/inbound_inboxes.py create mode 100644 mailtrap/api/resources/inbound_messages.py create mode 100644 mailtrap/api/resources/inbound_threads.py diff --git a/mailtrap/api/inbound.py b/mailtrap/api/inbound.py new file mode 100644 index 0000000..2250e40 --- /dev/null +++ b/mailtrap/api/inbound.py @@ -0,0 +1,26 @@ +from mailtrap.api.resources.inbound_folders import InboundFoldersApi +from mailtrap.api.resources.inbound_inboxes import InboundInboxesApi +from mailtrap.api.resources.inbound_messages import InboundMessagesApi +from mailtrap.api.resources.inbound_threads import InboundThreadsApi +from mailtrap.http import HttpClient + + +class InboundBaseApi: + def __init__(self, client: HttpClient) -> None: + self._client = client + + @property + def folders(self) -> InboundFoldersApi: + return InboundFoldersApi(client=self._client) + + @property + def inboxes(self) -> InboundInboxesApi: + return InboundInboxesApi(client=self._client) + + @property + def messages(self) -> InboundMessagesApi: + return InboundMessagesApi(client=self._client) + + @property + def threads(self) -> InboundThreadsApi: + return InboundThreadsApi(client=self._client) diff --git a/mailtrap/api/resources/inbound_folders.py b/mailtrap/api/resources/inbound_folders.py new file mode 100644 index 0000000..3478640 --- /dev/null +++ b/mailtrap/api/resources/inbound_folders.py @@ -0,0 +1,43 @@ +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import CreateInboundFolderParams +from mailtrap.models.inbound import InboundFolder +from mailtrap.models.inbound import UpdateInboundFolderParams + + +class InboundFoldersApi: + def __init__(self, client: HttpClient) -> None: + self._client = client + + def get_list(self) -> list[InboundFolder]: + """Get all inbound folders in the account.""" + response = self._client.get(self._api_path()) + return [InboundFolder(**folder) for folder in response] + + def get_by_id(self, folder_id: int) -> InboundFolder: + """Get an inbound folder by ID.""" + response = self._client.get(self._api_path(folder_id)) + return InboundFolder(**response) + + def create(self, params: CreateInboundFolderParams) -> InboundFolder: + """Create a new inbound folder.""" + response = self._client.post(self._api_path(), json=params.api_data) + return InboundFolder(**response) + + def update(self, folder_id: int, params: UpdateInboundFolderParams) -> InboundFolder: + """Rename an inbound folder.""" + response = self._client.patch(self._api_path(folder_id), json=params.api_data) + return InboundFolder(**response) + + def delete(self, folder_id: int) -> DeletedObject: + """Delete an inbound folder along with all of its inboxes.""" + self._client.delete(self._api_path(folder_id)) + return DeletedObject(folder_id) + + def _api_path(self, folder_id: Optional[int] = None) -> str: + path = "/api/inbound/folders" + if folder_id: + return f"{path}/{folder_id}" + return path diff --git a/mailtrap/api/resources/inbound_inboxes.py b/mailtrap/api/resources/inbound_inboxes.py new file mode 100644 index 0000000..85c289a --- /dev/null +++ b/mailtrap/api/resources/inbound_inboxes.py @@ -0,0 +1,47 @@ +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import CreateInboundInboxParams +from mailtrap.models.inbound import InboundInbox +from mailtrap.models.inbound import UpdateInboundInboxParams + + +class InboundInboxesApi: + def __init__(self, client: HttpClient) -> None: + self._client = client + + def get_list(self, folder_id: int) -> list[InboundInbox]: + """Get all inboxes in an inbound folder.""" + response = self._client.get(self._api_path(folder_id)) + return [InboundInbox(**inbox) for inbox in response] + + def get_by_id(self, folder_id: int, inbox_id: int) -> InboundInbox: + """Get an inbound inbox by ID.""" + response = self._client.get(self._api_path(folder_id, inbox_id)) + return InboundInbox(**response) + + def create(self, folder_id: int, params: CreateInboundInboxParams) -> InboundInbox: + """Create a new inbound inbox in a folder.""" + response = self._client.post(self._api_path(folder_id), json=params.api_data) + return InboundInbox(**response) + + def update( + self, folder_id: int, inbox_id: int, params: UpdateInboundInboxParams + ) -> InboundInbox: + """Rename an inbound inbox.""" + response = self._client.patch( + self._api_path(folder_id, inbox_id), json=params.api_data + ) + return InboundInbox(**response) + + def delete(self, folder_id: int, inbox_id: int) -> DeletedObject: + """Delete an inbound inbox.""" + self._client.delete(self._api_path(folder_id, inbox_id)) + return DeletedObject(inbox_id) + + def _api_path(self, folder_id: int, inbox_id: Optional[int] = None) -> str: + path = f"/api/inbound/folders/{folder_id}/inboxes" + if inbox_id: + return f"{path}/{inbox_id}" + return path diff --git a/mailtrap/api/resources/inbound_messages.py b/mailtrap/api/resources/inbound_messages.py new file mode 100644 index 0000000..78241c2 --- /dev/null +++ b/mailtrap/api/resources/inbound_messages.py @@ -0,0 +1,71 @@ +from typing import Any +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import ForwardInboundMessageParams +from mailtrap.models.inbound import InboundMessageDetails +from mailtrap.models.inbound import InboundMessagesListResponse +from mailtrap.models.inbound import InboundSendResult +from mailtrap.models.inbound import ReplyInboundMessageParams + + +class InboundMessagesApi: + def __init__(self, client: HttpClient) -> None: + self._client = client + + def get_list( + self, inbox_id: int, last_id: Optional[str] = None + ) -> InboundMessagesListResponse: + """ + List received messages in an inbox. Pass last_id from a previous + response to fetch the next page. + """ + params: dict[str, Any] = {} + if last_id: + params["last_id"] = last_id + response = self._client.get(self._api_path(inbox_id), params=params or None) + return InboundMessagesListResponse(**response) + + def get_by_id(self, inbox_id: int, message_id: str) -> InboundMessageDetails: + """Get a single message with its body and attachment download URLs.""" + response = self._client.get(self._api_path(inbox_id, message_id)) + return InboundMessageDetails(**response) + + def delete(self, inbox_id: int, message_id: str) -> DeletedObject: + """Delete a message.""" + self._client.delete(self._api_path(inbox_id, message_id)) + return DeletedObject(message_id) + + def reply( + self, inbox_id: int, message_id: str, params: ReplyInboundMessageParams + ) -> InboundSendResult: + """Reply to a message (to the original sender). Sends a real email.""" + response = self._client.post( + f"{self._api_path(inbox_id, message_id)}/reply", json=params.api_data + ) + return InboundSendResult(**response) + + def reply_all( + self, inbox_id: int, message_id: str, params: ReplyInboundMessageParams + ) -> InboundSendResult: + """Reply to a message and copy its other recipients. Sends a real email.""" + response = self._client.post( + f"{self._api_path(inbox_id, message_id)}/reply_all", json=params.api_data + ) + return InboundSendResult(**response) + + def forward( + self, inbox_id: int, message_id: str, params: ForwardInboundMessageParams + ) -> InboundSendResult: + """Forward a message to new recipients. Sends a real email.""" + response = self._client.post( + f"{self._api_path(inbox_id, message_id)}/forward", json=params.api_data + ) + return InboundSendResult(**response) + + def _api_path(self, inbox_id: int, message_id: Optional[str] = None) -> str: + path = f"/api/inbound/inboxes/{inbox_id}/messages" + if message_id: + return f"{path}/{message_id}" + return path diff --git a/mailtrap/api/resources/inbound_threads.py b/mailtrap/api/resources/inbound_threads.py new file mode 100644 index 0000000..6e1daf5 --- /dev/null +++ b/mailtrap/api/resources/inbound_threads.py @@ -0,0 +1,41 @@ +from typing import Any +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundThread +from mailtrap.models.inbound import InboundThreadsListResponse + + +class InboundThreadsApi: + def __init__(self, client: HttpClient) -> None: + self._client = client + + def get_list( + self, inbox_id: int, last_id: Optional[str] = None + ) -> InboundThreadsListResponse: + """ + List conversation threads in an inbox. Pass last_id from a previous + response to fetch the next page. + """ + params: dict[str, Any] = {} + if last_id: + params["last_id"] = last_id + response = self._client.get(self._api_path(inbox_id), params=params or None) + return InboundThreadsListResponse(**response) + + def get_by_id(self, inbox_id: int, thread_id: str) -> InboundThread: + """Get a single thread with its messages embedded (oldest first).""" + response = self._client.get(self._api_path(inbox_id, thread_id)) + return InboundThread(**response) + + def delete(self, inbox_id: int, thread_id: str) -> DeletedObject: + """Delete a thread.""" + self._client.delete(self._api_path(inbox_id, thread_id)) + return DeletedObject(thread_id) + + def _api_path(self, inbox_id: int, thread_id: Optional[str] = None) -> str: + path = f"/api/inbound/inboxes/{inbox_id}/threads" + if thread_id: + return f"{path}/{thread_id}" + return path diff --git a/mailtrap/client.py b/mailtrap/client.py index 0369776..dcb42e1 100644 --- a/mailtrap/client.py +++ b/mailtrap/client.py @@ -9,6 +9,7 @@ from mailtrap.api.contacts import ContactsBaseApi from mailtrap.api.email_logs import EmailLogsBaseApi from mailtrap.api.general import GeneralApi +from mailtrap.api.inbound import InboundBaseApi from mailtrap.api.organizations import OrganizationsBaseApi from mailtrap.api.resources.stats import StatsApi from mailtrap.api.sending import SendingApi @@ -76,6 +77,12 @@ def general_api(self) -> GeneralApi: client=HttpClient(host=GENERAL_HOST, headers=self.headers), ) + @property + def inbound_api(self) -> InboundBaseApi: + return InboundBaseApi( + client=HttpClient(host=GENERAL_HOST, headers=self.headers), + ) + @property def testing_api(self) -> TestingApi: self._validate_account_id() From 0c3ef6e40ce18934fa28934f92e302086d708133 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 11:19:00 +0200 Subject: [PATCH 3/6] Add inbound email tests responses-mocked API tests for inbound folders, inboxes, messages, and threads (happy paths, error tables, token-scoped URLs, flat request bodies, last_id pagination) plus model tests for params serialization, the forward recipient guard, and response parsing. --- tests/unit/api/inbound/__init__.py | 0 .../unit/api/inbound/test_inbound_folders.py | 109 ++++++++++ .../unit/api/inbound/test_inbound_inboxes.py | 125 ++++++++++++ .../unit/api/inbound/test_inbound_messages.py | 187 ++++++++++++++++++ .../unit/api/inbound/test_inbound_threads.py | 133 +++++++++++++ tests/unit/models/test_inbound.py | 75 +++++++ 6 files changed, 629 insertions(+) create mode 100644 tests/unit/api/inbound/__init__.py create mode 100644 tests/unit/api/inbound/test_inbound_folders.py create mode 100644 tests/unit/api/inbound/test_inbound_inboxes.py create mode 100644 tests/unit/api/inbound/test_inbound_messages.py create mode 100644 tests/unit/api/inbound/test_inbound_threads.py create mode 100644 tests/unit/models/test_inbound.py diff --git a/tests/unit/api/inbound/__init__.py b/tests/unit/api/inbound/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/api/inbound/test_inbound_folders.py b/tests/unit/api/inbound/test_inbound_folders.py new file mode 100644 index 0000000..4a4b919 --- /dev/null +++ b/tests/unit/api/inbound/test_inbound_folders.py @@ -0,0 +1,109 @@ +import pytest +import responses + +from mailtrap.api.resources.inbound_folders import InboundFoldersApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import CreateInboundFolderParams +from mailtrap.models.inbound import InboundFolder +from mailtrap.models.inbound import UpdateInboundFolderParams +from tests import conftest + +FOLDER_ID = 42 +BASE_URL = f"https://{GENERAL_HOST}/api/inbound/folders" + +ERROR_CASES = [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.NOT_FOUND_STATUS_CODE, + conftest.NOT_FOUND_RESPONSE, + conftest.NOT_FOUND_ERROR_MESSAGE, + ), +] + + +@pytest.fixture +def folders_api() -> InboundFoldersApi: + return InboundFoldersApi(client=HttpClient(GENERAL_HOST)) + + +class TestInboundFoldersApi: + + @responses.activate + def test_get_list_should_return_folders(self, folders_api: InboundFoldersApi) -> None: + responses.get(BASE_URL, json=[{"id": FOLDER_ID, "name": "Support"}], status=200) + + folders = folders_api.get_list() + + assert all(isinstance(f, InboundFolder) for f in folders) + assert folders[0].id == FOLDER_ID + assert folders[0].name == "Support" + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", ERROR_CASES + ) + @responses.activate + def test_get_list_should_raise_api_errors( + self, + folders_api: InboundFoldersApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get(BASE_URL, status=status_code, json=response_json) + + with pytest.raises(APIError) as exc_info: + folders_api.get_list() + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_get_by_id_should_return_folder(self, folders_api: InboundFoldersApi) -> None: + responses.get( + f"{BASE_URL}/{FOLDER_ID}", + json={"id": FOLDER_ID, "name": "Support"}, + status=200, + ) + + folder = folders_api.get_by_id(FOLDER_ID) + + assert isinstance(folder, InboundFolder) + assert folder.id == FOLDER_ID + + @responses.activate + def test_create_should_return_folder(self, folders_api: InboundFoldersApi) -> None: + responses.post(BASE_URL, json={"id": FOLDER_ID, "name": "Support"}, status=200) + + folder = folders_api.create(CreateInboundFolderParams(name="Support")) + + assert folder.id == FOLDER_ID + assert responses.calls[-1].request.body == b'{"name": "Support"}' + + @responses.activate + def test_update_should_return_folder(self, folders_api: InboundFoldersApi) -> None: + responses.patch( + f"{BASE_URL}/{FOLDER_ID}", + json={"id": FOLDER_ID, "name": "Renamed"}, + status=200, + ) + + folder = folders_api.update(FOLDER_ID, UpdateInboundFolderParams(name="Renamed")) + + assert folder.name == "Renamed" + + @responses.activate + def test_delete_should_return_deleted_object( + self, folders_api: InboundFoldersApi + ) -> None: + responses.delete(f"{BASE_URL}/{FOLDER_ID}", status=204) + + deleted = folders_api.delete(FOLDER_ID) + + assert isinstance(deleted, DeletedObject) + assert deleted.id == FOLDER_ID diff --git a/tests/unit/api/inbound/test_inbound_inboxes.py b/tests/unit/api/inbound/test_inbound_inboxes.py new file mode 100644 index 0000000..9cadf6e --- /dev/null +++ b/tests/unit/api/inbound/test_inbound_inboxes.py @@ -0,0 +1,125 @@ +import pytest +import responses + +from mailtrap.api.resources.inbound_inboxes import InboundInboxesApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import CreateInboundInboxParams +from mailtrap.models.inbound import InboundInbox +from mailtrap.models.inbound import UpdateInboundInboxParams +from tests import conftest + +FOLDER_ID = 42 +INBOX_ID = 9 +BASE_URL = f"https://{GENERAL_HOST}/api/inbound/folders/{FOLDER_ID}/inboxes" + + +def _inbox_dict() -> dict: + return { + "id": INBOX_ID, + "name": "Tickets", + "address": "tickets@inbound.example.com", + "domain_id": 3, + } + + +@pytest.fixture +def inboxes_api() -> InboundInboxesApi: + return InboundInboxesApi(client=HttpClient(GENERAL_HOST)) + + +class TestInboundInboxesApi: + + @responses.activate + def test_get_list_should_return_inboxes(self, inboxes_api: InboundInboxesApi) -> None: + responses.get(BASE_URL, json=[_inbox_dict()], status=200) + + inboxes = inboxes_api.get_list(FOLDER_ID) + + assert all(isinstance(i, InboundInbox) for i in inboxes) + assert inboxes[0].id == INBOX_ID + assert inboxes[0].address == "tickets@inbound.example.com" + + @responses.activate + def test_get_by_id_should_return_inbox(self, inboxes_api: InboundInboxesApi) -> None: + responses.get(f"{BASE_URL}/{INBOX_ID}", json=_inbox_dict(), status=200) + + inbox = inboxes_api.get_by_id(FOLDER_ID, INBOX_ID) + + assert isinstance(inbox, InboundInbox) + assert inbox.domain_id == 3 + + @responses.activate + def test_create_should_send_flat_body_and_return_inbox( + self, inboxes_api: InboundInboxesApi + ) -> None: + responses.post(BASE_URL, json=_inbox_dict(), status=200) + + params = CreateInboundInboxParams(name="Tickets", domain_id=3) + inbox = inboxes_api.create(FOLDER_ID, params) + + assert inbox.id == INBOX_ID + assert responses.calls[-1].request.body == b'{"name": "Tickets", "domain_id": 3}' + + @responses.activate + def test_create_should_omit_domain_id_when_not_set( + self, inboxes_api: InboundInboxesApi + ) -> None: + responses.post(BASE_URL, json=_inbox_dict(), status=200) + + inboxes_api.create(FOLDER_ID, CreateInboundInboxParams(name="Tickets")) + + assert responses.calls[-1].request.body == b'{"name": "Tickets"}' + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", + [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.VALIDATION_ERRORS_STATUS_CODE, + {"errors": {"name": ["can't be blank"]}}, + "name: can't be blank", + ), + ], + ) + @responses.activate + def test_create_should_raise_api_errors( + self, + inboxes_api: InboundInboxesApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.post(BASE_URL, status=status_code, json=response_json) + + with pytest.raises(APIError) as exc_info: + inboxes_api.create(FOLDER_ID, CreateInboundInboxParams(name="")) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_update_should_return_inbox(self, inboxes_api: InboundInboxesApi) -> None: + responses.patch(f"{BASE_URL}/{INBOX_ID}", json=_inbox_dict(), status=200) + + inbox = inboxes_api.update( + FOLDER_ID, INBOX_ID, UpdateInboundInboxParams(name="Tickets") + ) + + assert inbox.id == INBOX_ID + + @responses.activate + def test_delete_should_return_deleted_object( + self, inboxes_api: InboundInboxesApi + ) -> None: + responses.delete(f"{BASE_URL}/{INBOX_ID}", status=204) + + deleted = inboxes_api.delete(FOLDER_ID, INBOX_ID) + + assert isinstance(deleted, DeletedObject) + assert deleted.id == INBOX_ID diff --git a/tests/unit/api/inbound/test_inbound_messages.py b/tests/unit/api/inbound/test_inbound_messages.py new file mode 100644 index 0000000..da14e9e --- /dev/null +++ b/tests/unit/api/inbound/test_inbound_messages.py @@ -0,0 +1,187 @@ +import pytest +import responses + +from mailtrap.api.resources.inbound_messages import InboundMessagesApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import ForwardInboundMessageParams +from mailtrap.models.inbound import InboundMessageDetails +from mailtrap.models.inbound import InboundMessagesListResponse +from mailtrap.models.inbound import InboundSendResult +from mailtrap.models.inbound import ReplyInboundMessageParams +from mailtrap.models.mail.address import Address +from tests import conftest + +INBOX_ID = 9 +MESSAGE_ID = "1871574677877796928" +BASE_URL = f"https://{GENERAL_HOST}/api/inbound/inboxes/{INBOX_ID}/messages" + + +@pytest.fixture +def messages_api() -> InboundMessagesApi: + return InboundMessagesApi(client=HttpClient(GENERAL_HOST)) + + +class TestInboundMessagesApi: + + @responses.activate + def test_get_list_should_return_page(self, messages_api: InboundMessagesApi) -> None: + responses.get( + BASE_URL, + json={ + "data": [ + { + "id": MESSAGE_ID, + "inbox_id": INBOX_ID, + "from": "customer@example.com", + "subject": "Question", + "received_at": "2026-01-15T10:30:00Z", + } + ], + "total_count": 1, + "last_id": MESSAGE_ID, + }, + status=200, + ) + + page = messages_api.get_list(INBOX_ID) + + assert isinstance(page, InboundMessagesListResponse) + assert page.total_count == 1 + assert page.last_id == MESSAGE_ID + assert page.data[0].from_ == "customer@example.com" + + @responses.activate + def test_get_list_should_pass_last_id_cursor( + self, messages_api: InboundMessagesApi + ) -> None: + responses.get( + BASE_URL, json={"data": [], "total_count": 0, "last_id": None}, status=200 + ) + + messages_api.get_list(INBOX_ID, last_id="cursor-1") + + assert "last_id=cursor-1" in responses.calls[-1].request.url + + @responses.activate + def test_get_by_id_should_return_details( + self, messages_api: InboundMessagesApi + ) -> None: + responses.get( + f"{BASE_URL}/{MESSAGE_ID}", + json={ + "id": MESSAGE_ID, + "inbox_id": INBOX_ID, + "from": "customer@example.com", + "received_at": "2026-01-15T10:30:00Z", + "html_body": "

Hi

", + "attachments": [{"attachment_id": "a1", "download_url": "https://x/a1"}], + }, + status=200, + ) + + message = messages_api.get_by_id(INBOX_ID, MESSAGE_ID) + + assert isinstance(message, InboundMessageDetails) + assert message.html_body == "

Hi

" + assert message.attachments[0].download_url == "https://x/a1" + + @responses.activate + def test_delete_should_return_deleted_object( + self, messages_api: InboundMessagesApi + ) -> None: + responses.delete(f"{BASE_URL}/{MESSAGE_ID}", status=204) + + deleted = messages_api.delete(INBOX_ID, MESSAGE_ID) + + assert isinstance(deleted, DeletedObject) + assert deleted.id == MESSAGE_ID + + @responses.activate + def test_reply_should_send_flat_body_and_return_result( + self, messages_api: InboundMessagesApi + ) -> None: + responses.post( + f"{BASE_URL}/{MESSAGE_ID}/reply", + json={"message_ids": ["s1"]}, + status=200, + ) + + result = messages_api.reply( + INBOX_ID, MESSAGE_ID, ReplyInboundMessageParams(text="Thanks!") + ) + + assert isinstance(result, InboundSendResult) + assert result.message_ids == ["s1"] + assert responses.calls[-1].request.body == b'{"text": "Thanks!"}' + + @responses.activate + def test_reply_all_should_return_result( + self, messages_api: InboundMessagesApi + ) -> None: + responses.post( + f"{BASE_URL}/{MESSAGE_ID}/reply_all", + json={"message_ids": ["s1", "s2"]}, + status=200, + ) + + result = messages_api.reply_all( + INBOX_ID, MESSAGE_ID, ReplyInboundMessageParams(text="All") + ) + + assert result.message_ids == ["s1", "s2"] + + @responses.activate + def test_forward_should_serialize_recipients_and_return_result( + self, messages_api: InboundMessagesApi + ) -> None: + responses.post( + f"{BASE_URL}/{MESSAGE_ID}/forward", + json={"message_ids": ["s1"]}, + status=200, + ) + + params = ForwardInboundMessageParams(to=[Address(email="colleague@example.com")]) + result = messages_api.forward(INBOX_ID, MESSAGE_ID, params) + + assert result.message_ids == ["s1"] + assert ( + responses.calls[-1].request.body + == b'{"to": [{"email": "colleague@example.com"}]}' + ) + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", + [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.NOT_FOUND_STATUS_CODE, + conftest.NOT_FOUND_RESPONSE, + conftest.NOT_FOUND_ERROR_MESSAGE, + ), + ], + ) + @responses.activate + def test_reply_should_raise_api_errors( + self, + messages_api: InboundMessagesApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.post( + f"{BASE_URL}/{MESSAGE_ID}/reply", status=status_code, json=response_json + ) + + with pytest.raises(APIError) as exc_info: + messages_api.reply( + INBOX_ID, MESSAGE_ID, ReplyInboundMessageParams(text="Thanks!") + ) + + assert expected_error_message in str(exc_info.value) diff --git a/tests/unit/api/inbound/test_inbound_threads.py b/tests/unit/api/inbound/test_inbound_threads.py new file mode 100644 index 0000000..89ebd57 --- /dev/null +++ b/tests/unit/api/inbound/test_inbound_threads.py @@ -0,0 +1,133 @@ +import pytest +import responses + +from mailtrap.api.resources.inbound_threads import InboundThreadsApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundThread +from mailtrap.models.inbound import InboundThreadsListResponse +from tests import conftest + +INBOX_ID = 9 +THREAD_ID = "1871574677878845504" +BASE_URL = f"https://{GENERAL_HOST}/api/inbound/inboxes/{INBOX_ID}/threads" + + +@pytest.fixture +def threads_api() -> InboundThreadsApi: + return InboundThreadsApi(client=HttpClient(GENERAL_HOST)) + + +class TestInboundThreadsApi: + + @responses.activate + def test_get_list_should_return_page(self, threads_api: InboundThreadsApi) -> None: + responses.get( + BASE_URL, + json={ + "data": [ + { + "id": THREAD_ID, + "subject": "Billing", + "message_count": 2, + "size": 100, + "first_message_at": "2026-01-15T10:00:00Z", + "last_activity_at": "2026-01-15T11:00:00Z", + } + ], + "total_count": 1, + "last_id": THREAD_ID, + }, + status=200, + ) + + page = threads_api.get_list(INBOX_ID) + + assert isinstance(page, InboundThreadsListResponse) + assert page.total_count == 1 + assert page.data[0].message_count == 2 + + @responses.activate + def test_get_list_should_pass_last_id_cursor( + self, threads_api: InboundThreadsApi + ) -> None: + responses.get( + BASE_URL, json={"data": [], "total_count": 0, "last_id": None}, status=200 + ) + + threads_api.get_list(INBOX_ID, last_id="cursor-1") + + assert "last_id=cursor-1" in responses.calls[-1].request.url + + @responses.activate + def test_get_by_id_should_return_thread_with_messages( + self, threads_api: InboundThreadsApi + ) -> None: + responses.get( + f"{BASE_URL}/{THREAD_ID}", + json={ + "id": THREAD_ID, + "subject": "Billing", + "message_count": 1, + "size": 50, + "first_message_at": "2026-01-15T10:00:00Z", + "last_activity_at": "2026-01-15T10:00:00Z", + "messages": [ + { + "visibility_status": "available", + "direction": "inbound", + "from": "customer@example.com", + } + ], + }, + status=200, + ) + + thread = threads_api.get_by_id(INBOX_ID, THREAD_ID) + + assert isinstance(thread, InboundThread) + assert thread.messages[0].direction == "inbound" + assert thread.messages[0].from_ == "customer@example.com" + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", + [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.NOT_FOUND_STATUS_CODE, + conftest.NOT_FOUND_RESPONSE, + conftest.NOT_FOUND_ERROR_MESSAGE, + ), + ], + ) + @responses.activate + def test_get_by_id_should_raise_api_errors( + self, + threads_api: InboundThreadsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get(f"{BASE_URL}/{THREAD_ID}", status=status_code, json=response_json) + + with pytest.raises(APIError) as exc_info: + threads_api.get_by_id(INBOX_ID, THREAD_ID) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_delete_should_return_deleted_object( + self, threads_api: InboundThreadsApi + ) -> None: + responses.delete(f"{BASE_URL}/{THREAD_ID}", status=204) + + deleted = threads_api.delete(INBOX_ID, THREAD_ID) + + assert isinstance(deleted, DeletedObject) + assert deleted.id == THREAD_ID diff --git a/tests/unit/models/test_inbound.py b/tests/unit/models/test_inbound.py new file mode 100644 index 0000000..074ecb5 --- /dev/null +++ b/tests/unit/models/test_inbound.py @@ -0,0 +1,75 @@ +import pytest + +from mailtrap.models.inbound import CreateInboundFolderParams +from mailtrap.models.inbound import CreateInboundInboxParams +from mailtrap.models.inbound import ForwardInboundMessageParams +from mailtrap.models.inbound import InboundMessage +from mailtrap.models.inbound import InboundThread +from mailtrap.models.inbound import ReplyInboundMessageParams +from mailtrap.models.mail.address import Address + + +class TestInboundParams: + def test_create_folder_api_data(self) -> None: + assert CreateInboundFolderParams(name="Support").api_data == {"name": "Support"} + + def test_create_inbox_includes_domain_id(self) -> None: + params = CreateInboundInboxParams(name="Tickets", domain_id=3) + assert params.api_data == {"name": "Tickets", "domain_id": 3} + + def test_create_inbox_omits_domain_id_when_none(self) -> None: + assert CreateInboundInboxParams(name="Tickets").api_data == {"name": "Tickets"} + + def test_reply_serializes_sender_as_from_and_excludes_none(self) -> None: + params = ReplyInboundMessageParams( + sender=Address(email="me@example.com"), text="Thanks!" + ) + assert params.api_data == { + "from": {"email": "me@example.com"}, + "text": "Thanks!", + } + + def test_forward_serializes_recipients(self) -> None: + params = ForwardInboundMessageParams( + to=[Address(email="colleague@example.com")], text="FYI" + ) + assert params.api_data == { + "to": [{"email": "colleague@example.com"}], + "text": "FYI", + } + + def test_forward_requires_at_least_one_recipient(self) -> None: + with pytest.raises(ValueError): + ForwardInboundMessageParams(to=[]) + + +class TestInboundResponseModels: + def test_message_parses_from_alias_and_attachments(self) -> None: + message = InboundMessage( + **{ + "id": "m1", + "inbox_id": 9, + "from": "customer@example.com", + "received_at": "2026-01-15T10:30:00Z", + "attachments": [{"attachment_id": "a1", "download_url": "https://x/a1"}], + } + ) + assert message.from_ == "customer@example.com" + assert message.attachments[0].download_url == "https://x/a1" + assert message.to == [] + + def test_thread_parses_nested_messages(self) -> None: + thread = InboundThread( + **{ + "id": "t1", + "message_count": 1, + "size": 10, + "first_message_at": "2026-01-15T10:00:00Z", + "last_activity_at": "2026-01-15T10:00:00Z", + "messages": [ + {"visibility_status": "placeholder", "direction": "outbound"} + ], + } + ) + assert thread.messages[0].visibility_status == "placeholder" + assert thread.messages[0].from_ is None From 013a32b296d443c497ac986c13b93ce7f0a6f15b Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 11:30:27 +0200 Subject: [PATCH 4/6] Add inbound-related fields to email-log and sending-domain models EmailLogMessage gains rfc_message_id, in_reply_to, references, and thread_id; SendingDomain gains inbound_enabled and inbound_verified. These fields ship with the Inbound Email API. --- mailtrap/models/email_logs.py | 6 ++++ mailtrap/models/sending_domains.py | 2 ++ tests/unit/models/test_email_log_models.py | 29 ++++++++++++++++ tests/unit/models/test_sending_domains.py | 39 ++++++++++++++++++++++ 4 files changed, 76 insertions(+) diff --git a/mailtrap/models/email_logs.py b/mailtrap/models/email_logs.py index dc1cb00..6ffaba4 100644 --- a/mailtrap/models/email_logs.py +++ b/mailtrap/models/email_logs.py @@ -134,6 +134,10 @@ class EmailLogMessage(BaseModel): message_id: str status: Literal["delivered", "not_delivered", "enqueued", "opted_out"] subject: Optional[str] = None + rfc_message_id: Optional[str] = None + in_reply_to: Optional[str] = None + references: list[str] = Field(default_factory=list) + thread_id: Optional[str] = None from_: str = Field(alias="from") to: str sent_at: str @@ -161,6 +165,8 @@ def from_api(cls, data: dict[str, Any]) -> "EmailLogMessage": payload["custom_variables"] = {} if payload.get("template_variables") is None: payload["template_variables"] = {} + if payload.get("references") is None: + payload["references"] = [] return cls(**payload) diff --git a/mailtrap/models/sending_domains.py b/mailtrap/models/sending_domains.py index a470c0e..338c2fc 100644 --- a/mailtrap/models/sending_domains.py +++ b/mailtrap/models/sending_domains.py @@ -40,6 +40,8 @@ class SendingDomain: alert_recipient_email: Optional[str] = None dns_verified_at: Optional[str] = None dns_records: list[DnsRecord] = Field(default_factory=list) + inbound_enabled: Optional[bool] = None + inbound_verified: Optional[bool] = None @dataclass diff --git a/tests/unit/models/test_email_log_models.py b/tests/unit/models/test_email_log_models.py index 6f4b9e5..1df4ee6 100644 --- a/tests/unit/models/test_email_log_models.py +++ b/tests/unit/models/test_email_log_models.py @@ -41,6 +41,35 @@ def test_parses_list_item_with_from_key(self) -> None: assert msg.status == "delivered" assert msg.raw_message_url is None assert msg.events == [] + # threading fields default when absent + assert msg.rfc_message_id is None + assert msg.in_reply_to is None + assert msg.references == [] + assert msg.thread_id is None + + def test_parses_threading_fields_when_present(self) -> None: + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_variables": {}, + "opens_count": 0, + "clicks_count": 0, + "rfc_message_id": "", + "in_reply_to": "", + "references": ["", ""], + "thread_id": "thread-1", + } + msg = EmailLogMessage.from_api(data) + assert msg.rfc_message_id == "" + assert msg.in_reply_to == "" + assert msg.references == ["", ""] + assert msg.thread_id == "thread-1" def test_from_api_handles_from_and_events(self) -> None: data: dict[str, Any] = { diff --git a/tests/unit/models/test_sending_domains.py b/tests/unit/models/test_sending_domains.py index 43d7160..6e699c1 100644 --- a/tests/unit/models/test_sending_domains.py +++ b/tests/unit/models/test_sending_domains.py @@ -1,7 +1,46 @@ from mailtrap.models.sending_domains import CreateSendingDomainParams +from mailtrap.models.sending_domains import SendingDomain from mailtrap.models.sending_domains import SendSetupInstructionsParams +def _sending_domain_dict() -> dict: + return { + "id": 1, + "domain_name": "example.com", + "demo": False, + "compliance_status": "verified", + "dns_verified": True, + "open_tracking_enabled": True, + "click_tracking_enabled": True, + "auto_unsubscribe_link_enabled": False, + "custom_domain_tracking_enabled": False, + "health_alerts_enabled": True, + "critical_alerts_enabled": True, + "permissions": { + "can_read": True, + "can_update": True, + "can_destroy": True, + }, + } + + +class TestSendingDomain: + def test_parses_inbound_flags_when_present(self) -> None: + data = { + **_sending_domain_dict(), + "inbound_enabled": True, + "inbound_verified": False, + } + domain = SendingDomain(**data) + assert domain.inbound_enabled is True + assert domain.inbound_verified is False + + def test_inbound_flags_default_when_absent(self) -> None: + domain = SendingDomain(**_sending_domain_dict()) + assert domain.inbound_enabled is None + assert domain.inbound_verified is None + + class TestCreateSendingDomainParams: def test_api_data_should_return_dict_with_all_props(self) -> None: entity = CreateSendingDomainParams(domain_name="test.co") From d3b6ed63eb34c08cdeec0e2ca1ed56302eb74102 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 12:01:04 +0200 Subject: [PATCH 5/6] Add inbound email examples and README section Runnable example scripts for inbound folders, inboxes, messages (list/get/delete + reply/reply_all/forward), and threads, plus an Inbound Email API entry in the README examples list. --- README.md | 6 ++++ examples/inbound/folders.py | 38 ++++++++++++++++++++++++++ examples/inbound/inboxes.py | 40 +++++++++++++++++++++++++++ examples/inbound/messages.py | 53 ++++++++++++++++++++++++++++++++++++ examples/inbound/threads.py | 33 ++++++++++++++++++++++ 5 files changed, 170 insertions(+) create mode 100644 examples/inbound/folders.py create mode 100644 examples/inbound/inboxes.py create mode 100644 examples/inbound/messages.py create mode 100644 examples/inbound/threads.py diff --git a/README.md b/README.md index db927e6..913fd03 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,12 @@ 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) +### Inbound Email API: +- Folders management – [`inbound/folders.py`](examples/inbound/folders.py) +- Inboxes management – [`inbound/inboxes.py`](examples/inbound/inboxes.py) +- 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) + ### Webhooks API: - Webhooks management – [`webhooks/webhooks.py`](examples/webhooks/webhooks.py) - Verifying webhook signatures – [`webhooks/verify_signature.py`](examples/webhooks/verify_signature.py) diff --git a/examples/inbound/folders.py b/examples/inbound/folders.py new file mode 100644 index 0000000..9416b36 --- /dev/null +++ b/examples/inbound/folders.py @@ -0,0 +1,38 @@ +import mailtrap as mt +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundFolder + +API_TOKEN = "YOUR_API_TOKEN" + +client = mt.MailtrapClient(token=API_TOKEN) +folders_api = client.inbound_api.folders + + +def list_folders() -> list[InboundFolder]: + return folders_api.get_list() + + +def get_folder(folder_id: int) -> InboundFolder: + return folders_api.get_by_id(folder_id) + + +def create_folder(name: str) -> InboundFolder: + return folders_api.create(mt.CreateInboundFolderParams(name=name)) + + +def update_folder(folder_id: int, name: str) -> InboundFolder: + return folders_api.update(folder_id, mt.UpdateInboundFolderParams(name=name)) + + +def delete_folder(folder_id: int) -> DeletedObject: + return folders_api.delete(folder_id) + + +if __name__ == "__main__": + folder = create_folder("Support") + print(folder) + + print(list_folders()) + print(get_folder(folder.id)) + print(update_folder(folder.id, "Support (renamed)")) + print(delete_folder(folder.id)) diff --git a/examples/inbound/inboxes.py b/examples/inbound/inboxes.py new file mode 100644 index 0000000..f83517b --- /dev/null +++ b/examples/inbound/inboxes.py @@ -0,0 +1,40 @@ +import mailtrap as mt +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundInbox + +API_TOKEN = "YOUR_API_TOKEN" +FOLDER_ID = 42 + +client = mt.MailtrapClient(token=API_TOKEN) +inboxes_api = client.inbound_api.inboxes + + +def list_inboxes(folder_id: int) -> list[InboundInbox]: + return inboxes_api.get_list(folder_id) + + +def get_inbox(folder_id: int, inbox_id: int) -> InboundInbox: + return inboxes_api.get_by_id(folder_id, inbox_id) + + +def create_inbox(folder_id: int, name: str) -> InboundInbox: + # Omit domain_id for a Mailtrap-hosted inbox; pass it for a custom-domain inbox. + return inboxes_api.create(folder_id, mt.CreateInboundInboxParams(name=name)) + + +def update_inbox(folder_id: int, inbox_id: int, name: str) -> InboundInbox: + return inboxes_api.update(folder_id, inbox_id, mt.UpdateInboundInboxParams(name=name)) + + +def delete_inbox(folder_id: int, inbox_id: int) -> DeletedObject: + return inboxes_api.delete(folder_id, inbox_id) + + +if __name__ == "__main__": + inbox = create_inbox(FOLDER_ID, "Tickets") + print(inbox) + + print(list_inboxes(FOLDER_ID)) + print(get_inbox(FOLDER_ID, inbox.id)) + print(update_inbox(FOLDER_ID, inbox.id, "Tickets (renamed)")) + print(delete_inbox(FOLDER_ID, inbox.id)) diff --git a/examples/inbound/messages.py b/examples/inbound/messages.py new file mode 100644 index 0000000..c1f040f --- /dev/null +++ b/examples/inbound/messages.py @@ -0,0 +1,53 @@ +import mailtrap as mt +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundMessageDetails +from mailtrap.models.inbound import InboundMessagesListResponse +from mailtrap.models.inbound import InboundSendResult +from mailtrap.models.mail.address import Address + +API_TOKEN = "YOUR_API_TOKEN" +INBOX_ID = 9 + +client = mt.MailtrapClient(token=API_TOKEN) +messages_api = client.inbound_api.messages + + +def list_messages(inbox_id: int) -> InboundMessagesListResponse: + # Pass last_id from the previous page to paginate. + return messages_api.get_list(inbox_id) + + +def get_message(inbox_id: int, message_id: str) -> InboundMessageDetails: + return messages_api.get_by_id(inbox_id, message_id) + + +def delete_message(inbox_id: int, message_id: str) -> DeletedObject: + return messages_api.delete(inbox_id, message_id) + + +def reply(inbox_id: int, message_id: str, text: str) -> InboundSendResult: + params = mt.ReplyInboundMessageParams(text=text) + return messages_api.reply(inbox_id, message_id, params) + + +def reply_all(inbox_id: int, message_id: str, text: str) -> InboundSendResult: + params = mt.ReplyInboundMessageParams(text=text) + return messages_api.reply_all(inbox_id, message_id, params) + + +def forward(inbox_id: int, message_id: str, to: str) -> InboundSendResult: + params = mt.ForwardInboundMessageParams(to=[Address(email=to)]) + return messages_api.forward(inbox_id, message_id, params) + + +if __name__ == "__main__": + page = list_messages(INBOX_ID) + print(f"{len(page.data)} of {page.total_count} messages") + + if page.data: + message_id = page.data[0].id + print(get_message(INBOX_ID, message_id)) + print(reply(INBOX_ID, message_id, "Thanks for reaching out!")) + print(reply_all(INBOX_ID, message_id, "Looping everyone in.")) + print(forward(INBOX_ID, message_id, "colleague@example.com")) + print(delete_message(INBOX_ID, message_id)) diff --git a/examples/inbound/threads.py b/examples/inbound/threads.py new file mode 100644 index 0000000..6f4d7fe --- /dev/null +++ b/examples/inbound/threads.py @@ -0,0 +1,33 @@ +import mailtrap as mt +from mailtrap.models.common import DeletedObject +from mailtrap.models.inbound import InboundThread +from mailtrap.models.inbound import InboundThreadsListResponse + +API_TOKEN = "YOUR_API_TOKEN" +INBOX_ID = 9 + +client = mt.MailtrapClient(token=API_TOKEN) +threads_api = client.inbound_api.threads + + +def list_threads(inbox_id: int) -> InboundThreadsListResponse: + # Pass last_id from the previous page to paginate. + return threads_api.get_list(inbox_id) + + +def get_thread(inbox_id: int, thread_id: str) -> InboundThread: + return threads_api.get_by_id(inbox_id, thread_id) + + +def delete_thread(inbox_id: int, thread_id: str) -> DeletedObject: + return threads_api.delete(inbox_id, thread_id) + + +if __name__ == "__main__": + page = list_threads(INBOX_ID) + print(f"{len(page.data)} of {page.total_count} threads") + + if page.data: + thread_id = page.data[0].id + print(get_thread(INBOX_ID, thread_id)) + print(delete_thread(INBOX_ID, thread_id)) From 9439b16a3a292a2fc5b23b16bf8a262c8c2f12c2 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Fri, 31 Jul 2026 12:07:10 +0200 Subject: [PATCH 6/6] Add inbound_inbox_id to webhook models Support inbound_receiving webhooks, which link a webhook to an inbound inbox via inbound_inbox_id. Add the field to Webhook, CreateWebhookParams, and UpdateWebhookParams (webhook_type already accepts any string). --- mailtrap/models/webhooks.py | 3 ++ tests/unit/models/test_webhooks.py | 47 ++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 tests/unit/models/test_webhooks.py diff --git a/mailtrap/models/webhooks.py b/mailtrap/models/webhooks.py index d4482b8..cb06888 100644 --- a/mailtrap/models/webhooks.py +++ b/mailtrap/models/webhooks.py @@ -15,6 +15,7 @@ class Webhook: payload_format: str sending_stream: Optional[str] = None domain_id: Optional[int] = None + inbound_inbox_id: Optional[int] = None event_types: list[str] = Field(default_factory=list) @@ -47,6 +48,7 @@ class CreateWebhookParams(RequestParams): sending_stream: Optional[str] = None event_types: Optional[list[str]] = None domain_id: Optional[int] = None + inbound_inbox_id: Optional[int] = None @dataclass @@ -55,3 +57,4 @@ class UpdateWebhookParams(RequestParams): active: Optional[bool] = None payload_format: Optional[str] = None event_types: Optional[list[str]] = None + inbound_inbox_id: Optional[int] = None diff --git a/tests/unit/models/test_webhooks.py b/tests/unit/models/test_webhooks.py new file mode 100644 index 0000000..14716f9 --- /dev/null +++ b/tests/unit/models/test_webhooks.py @@ -0,0 +1,47 @@ +from mailtrap.models.webhooks import CreateWebhookParams +from mailtrap.models.webhooks import UpdateWebhookParams +from mailtrap.models.webhooks import Webhook + + +class TestWebhook: + def test_parses_inbound_receiving_webhook(self) -> None: + webhook = Webhook( + id=1, + url="https://example.com/hook", + active=True, + webhook_type="inbound_receiving", + payload_format="json", + inbound_inbox_id=665, + ) + assert webhook.webhook_type == "inbound_receiving" + assert webhook.inbound_inbox_id == 665 + + def test_inbound_inbox_id_defaults_to_none(self) -> None: + webhook = Webhook( + id=1, + url="https://example.com/hook", + active=True, + webhook_type="email_sending", + payload_format="json", + ) + assert webhook.inbound_inbox_id is None + + +class TestCreateWebhookParams: + def test_api_data_includes_inbound_inbox_id(self) -> None: + params = CreateWebhookParams( + url="https://example.com/hook", + webhook_type="inbound_receiving", + inbound_inbox_id=665, + ) + assert params.api_data == { + "url": "https://example.com/hook", + "webhook_type": "inbound_receiving", + "inbound_inbox_id": 665, + } + + +class TestUpdateWebhookParams: + def test_api_data_includes_inbound_inbox_id(self) -> None: + params = UpdateWebhookParams(inbound_inbox_id=665) + assert params.api_data == {"inbound_inbox_id": 665}