Fix issue #29. Add support of Emails Sandbox (Testing) API: Projects

examples/testing/projects.py

@@ -0,0 +1,30 @@
+import os
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
+
+from typing import Optional
+from mailtrap import MailtrapClient
+from mailtrap.models.projects import Project
+
+API_TOKEN = "YOU_API_TOKEN"
+ACCOUNT_ID = "YOU_ACCOUNT_ID"
+
+
+def find_project_by_name(project_name: str, projects: list[Project]) -> Optional[str]:
+    filtered_projects = [project for project in projects if project.name == project_name]
+    if filtered_projects:
+        return filtered_projects[0].id
+    return None
+
+
+client = MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID)
+api = client.testing.projects
+
+project_name = "Example-project"
+
+created_project = api.create(project_name=project_name)
+projects = api.get_list()
+project_id = find_project_by_name(project_name, projects)
+project = api.get_by_id(project_id)
+updated_projected = api.update(project_id, "Updated-project-name")
+api.delete(project_id)

mailtrap/api/resources/projects.py

@@ -0,0 +1,37 @@
+from mailtrap.http import HttpClient
+from mailtrap.models.base import DeletedObject
+from mailtrap.models.projects import Project
+
+
+class ProjectsApi:
+    def __init__(self, account_id: str, client: HttpClient) -> None:
+        self.account_id = account_id
+        self.client = client
+
+    def get_list(self) -> list[Project]:
+        response = self.client.list(f"/api/accounts/{self.account_id}/projects")
+        return [Project.from_dict(project) for project in response]
+
+    def get_by_id(self, project_id: str) -> Project:
+        response = self.client.get(f"/api/accounts/{self.account_id}/projects/{project_id}")
+        return Project.from_dict(response)
+
+    def create(self, project_name: str) -> Project:
+        response = self.client.post(
+            f"/api/accounts/{self.account_id}/projects",
+            json={"project": {"name": project_name}},
+        )
+        return Project.from_dict(response)
+
+    def update(self, project_id: str, project_name: str) -> Project:
+        response = self.client.patch(
+            f"/api/accounts/{self.account_id}/projects/{project_id}",
+            json={"project": {"name": project_name}},
+        )
+        return Project.from_dict(response)
+
+    def delete(self, project_id: str) -> DeletedObject:
+        response = self.client.delete(
+            f"/api/accounts/{self.account_id}/projects/{project_id}",
+        )
+        return DeletedObject(response["id"])

mailtrap/api/testing.py

@@ -0,0 +1,13 @@
+
+from mailtrap.api.resources.projects import ProjectsApi 
+from mailtrap.http import HttpClient
+
+
+class TestingApi:
+    def __init__(self, account_id: str, client: HttpClient) -> None:
+        self.account_id = account_id
+        self.client = client
+
+    @property
+    def projects(self) -> ProjectsApi:
+        return ProjectsApi(account_id=self.account_id, client=self.client)

mailtrap/client.py

@@ -4,9 +4,12 @@
 
 import requests
 
+from mailtrap.api.testing import TestingApi
+from mailtrap.config import MAILTRAP_HOST
 from mailtrap.exceptions import APIError
 from mailtrap.exceptions import AuthorizationError
 from mailtrap.exceptions import ClientConfigurationError
+from mailtrap.http import HttpClient
 from mailtrap.mail.base import BaseMail
 
 
@@ -24,16 +27,26 @@ def __init__(
         bulk: bool = False,
         sandbox: bool = False,
         inbox_id: Optional[str] = None,
+        account_id: Optional[str] = None,
     ) -> None:
         self.token = token
         self.api_host = api_host
         self.api_port = api_port
         self.bulk = bulk
         self.sandbox = sandbox
         self.inbox_id = inbox_id
+        self.account_id = account_id
 
         self._validate_itself()
 
+    @property
+    def testing(self) -> TestingApi:
+        if not self.account_id:
+            raise ClientConfigurationError("`account_id` is required for Testing API")
+
+        http_client = HttpClient(host=MAILTRAP_HOST, headers=self.headers)
+        return TestingApi(account_id=self.account_id, client=http_client)
+
     def send(self, mail: BaseMail) -> dict[str, Union[bool, list[str]]]:
         response = requests.post(
             self.api_send_url, headers=self.headers, json=mail.api_data

mailtrap/config.py

@@ -0,0 +1,3 @@
+MAILTRAP_HOST = "mailtrap.io"
+
+DEFAULT_REQUEST_TIMEOUT = 30  # in seconds

mailtrap/http.py

@@ -0,0 +1,127 @@
+from typing import Any, Optional, Type, TypeVar
+from typing import NoReturn
+
+from requests import Response, Session
+
+from mailtrap.config import DEFAULT_REQUEST_TIMEOUT
+from mailtrap.exceptions import APIError
+from mailtrap.exceptions import AuthorizationError
+
+T = TypeVar("T")
+
+
+class HttpClient:
+    def __init__(
+        self, 
+        host: str, 
+        headers: Optional[dict[str, str]] = None, 
+        timeout: int = DEFAULT_REQUEST_TIMEOUT
+    ):
+        self._host = host
+        self._session = Session()
+        self._session.headers.update(headers or {})
+        self._timeout = timeout
+
+    def _url(self, path: str) -> str:
+        return f"https://{self._host}/{path.lstrip('/')}"
+
+    def _handle_failed_response(self, response: Response) -> NoReturn:
+        status_code = response.status_code
+        try:
+            data = response.json()
+        except ValueError:
+            raise APIError(status_code, errors=["Unknown Error"])
+
+        errors = _extract_errors(data)
+
+        if status_code == 401:
+            raise AuthorizationError(errors=errors)
+
+        raise APIError(status_code, errors=errors)
+
+    def _process_response(
+        self, 
+        response: Response, 
+        expected_type: Type[T]
+    ) -> T:
+        if not response.ok:
+            self._handle_failed_response(response)
+        data = response.json()
+        if not isinstance(data, expected_type):
+            raise APIError(response.status_code, errors=[f"Expected response type {expected_type.__name__}"])
+        return data
+
+    def _process_response_dict(self, response: Response) -> dict[str, Any]:
+        return self._process_response(response, dict)
+
+    def _process_response_list(self, response: Response) -> list[dict[str, Any]]:
+        return self._process_response(response, list)
+
+    def get(
+        self, 
+        path: str, 
+        params: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        response = self._session.get(self._url(path), params=params, timeout=self._timeout)
+        return self._process_response_dict(response)
+
+    def list(
+        self, 
+        path: str, 
+        params: Optional[dict[str, Any]] = None
+    ) -> list[dict[str, Any]]:
+        response = self._session.get(self._url(path), params=params, timeout=self._timeout)
+        return self._process_response_list(response)
+
+    def post(
+        self, 
+        path: str, 
+        json: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        response = self._session.post(self._url(path), json=json, timeout=self._timeout)
+        return self._process_response_dict(response)
+
+    def put(
+        self, 
+        path: str, 
+        json: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        response = self._session.put(self._url(path), json=json, timeout=self._timeout)
+        return self._process_response_dict(response)
+
+    def patch(
+        self, 
+        path: str, 
+        json: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        response = self._session.patch(self._url(path), json=json, timeout=self._timeout)
+        return self._process_response_dict(response)
+
+    def delete(self, path: str) -> dict[str, Any]:
+        response = self._session.delete(self._url(path), timeout=self._timeout)
+        return self._process_response_dict(response)
+
+
+def _extract_errors(data: dict[str, Any]) -> list[str]:    
+    def flatten_errors(errors: Any) -> list[str]:
+        if isinstance(errors, list):
+            return [str(error) for error in errors]
+
+        if isinstance(errors, dict):
+            flat_errors = []
+            for key, value in errors.items():
+                if isinstance(value, list):
+                    flat_errors.extend([f"{key}: {v}" for v in value])
+                else:
+                    flat_errors.append(f"{key}: {value}")
+            return flat_errors
+
+        return [str(errors)]
+
+    if "errors" in data:
+        return flatten_errors(data["errors"])
+
+    if "error" in data:
+        return flatten_errors(data["error"])
+
+    return ["Unknown error"]

mailtrap/models/base.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass, fields, is_dataclass
+from typing import Any, Type, TypeVar, get_args, get_origin
+
+
+T = TypeVar("T", bound="BaseModel")
+
+
+@dataclass
+class BaseModel:
+    @classmethod
+    def from_dict(cls: Type[T], data: dict[str, Any]) -> T:
+        values: dict[str, Any] = {}
+        for field in fields(cls):
+            value = data.get(field.name)
+
+            if value is None:
+                values[field.name] = None
+                continue
+
+            field_type = field.type
+            values[field.name] = cls._parse_value(value, field_type)
+
+        return cls(**values)
+
+    @classmethod
+    def _parse_value(cls, value: Any, field_type: Any) -> Any:
+        origin = get_origin(field_type)
+
+        if origin is list:
+            item_type = get_args(field_type)[0]
+            return [cls._parse_value(item, item_type) for item in value]
+
+        if is_dataclass(field_type) and hasattr(field_type, "from_dict"):
+            return field_type.from_dict(value)
+
+        return value
+
+
+
+@dataclass
+class DeletedObject(BaseModel):
+    id: str

mailtrap/models/inboxes.py

@@ -0,0 +1,31 @@
+
+from dataclasses import dataclass
+from mailtrap.models.base import BaseModel
+from mailtrap.models.permissions import Permissions
+
+
+@dataclass
+class Inbox(BaseModel):
+    id: int
+    name: str
+    username: str
+    max_size: int
+    status: str
+    email_username: str
+    email_username_enabled: bool
+    sent_messages_count: int
+    forwarded_messages_count: int
+    used: bool
+    forward_from_email_address: str
+    project_id: int
+    domain: str
+    pop3_domain: str
+    email_domain: str
+    emails_count: int
+    emails_unread_count: int
+    smtp_ports: list[int]
+    pop3_ports: list[int]
+    max_message_size: int
+    permissions: Permissions
+    password: str | None = None  # Password is only available if you have admin permissions for the inbox.
+    last_message_sent_at: str | None = None

mailtrap/models/permissions.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+from mailtrap.models.base import BaseModel
+
+
+@dataclass
+class Permissions(BaseModel):
+    can_read: bool
+    can_update: bool
+    can_destroy: bool
+    can_leave: bool

mailtrap/models/projects.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+from mailtrap.models.base import BaseModel
+from mailtrap.models.inboxes import Inbox
+from mailtrap.models.permissions import Permissions
+
+
+@dataclass
+class ShareLinks(BaseModel):
+    admin: str
+    viewer: str
+
+
+@dataclass
+class Project(BaseModel):
+    id: str
+    name: str
+    share_links: ShareLinks
+    inboxes: list[Inbox]
+    permissions: Permissions