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

Author: Ihor Bilous

mailtrap/api/base.py

@@ -0,0 +1,71 @@
+from abc import ABC
+from enum import Enum
+from typing import Any, Dict, List, NoReturn, Union, cast
+from requests import Response, Session
+
+from mailtrap.exceptions import APIError, AuthorizationError
+
+RESPONSE_TYPE = Dict[str, Any]
+LIST_RESPONSE_TYPE = List[Dict[str, Any]]
+
+
+class HttpMethod(Enum):
+    GET = "GET"
+    POST = "POST"
+    PUT = "PUT"
+    PATCH = "PATCH"
+    DELTE = "DELETE"
+
+
+def _extract_errors(data: Dict[str, Any]) -> List[str]:
+    if "errors" in data:
+        errors = data["errors"]
+
+        if isinstance(errors, list):
+            return [str(err) for err 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)]
+
+    elif "error" in data:
+        return [str(data["error"])]
+
+    return ["Unknown error"]
+
+
+class BaseHttpApiClient(ABC):
+    def __init__(self, session: Session):
+        self.session = session
+
+    def _request(self, method: HttpMethod, url: str, **kwargs: Any) -> Union[RESPONSE_TYPE, LIST_RESPONSE_TYPE]:
+        response = self.session.request(method.value, url, **kwargs)
+        if response.ok:
+            data = cast(Union[RESPONSE_TYPE, LIST_RESPONSE_TYPE], response.json())
+            return data
+
+        self._handle_failed_response(response)
+
+    @staticmethod
+    def _handle_failed_response(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)
+

mailtrap/api/projects.py

@@ -0,0 +1,50 @@
+from typing import List, cast
+
+from mailtrap.api.base import LIST_RESPONSE_TYPE, RESPONSE_TYPE, BaseHttpApiClient, HttpMethod
+from mailtrap.constants import MAILTRAP_HOST
+from mailtrap.models.common import DeletedObject
+from mailtrap.models.projects import Project
+
+
+class ProjectsApiClient(BaseHttpApiClient):
+
+    def _build_url(self, account_id: str, *parts: str) -> str:
+        base_url = f"https://{MAILTRAP_HOST}/api/accounts/{account_id}/projects"
+        return "/".join([base_url, *parts])
+
+    def get_list(self, account_id: str) -> List[Project]:
+        response: LIST_RESPONSE_TYPE = cast(LIST_RESPONSE_TYPE, self._request(
+            HttpMethod.GET, 
+            self._build_url(account_id)
+        ))
+        return [Project.from_dict(proj) for proj in response]
+
+    def get_by_id(self, account_id: str, project_id: str) -> Project:
+        response: RESPONSE_TYPE = cast(RESPONSE_TYPE, self._request(
+            HttpMethod.GET, 
+            self._build_url(account_id, project_id)
+        ))
+        return Project.from_dict(response)
+    
+    def create(self, account_id: str, name: str) -> Project:
+        response: RESPONSE_TYPE = cast(RESPONSE_TYPE, self._request(
+            HttpMethod.POST, 
+            self._build_url(account_id), 
+            json={"project": {"name": name}},
+        ))
+        return Project.from_dict(response)
+    
+    def update(self, account_id: str, project_id: str, name: str) -> Project:
+        response: RESPONSE_TYPE = cast(RESPONSE_TYPE, self._request(
+            HttpMethod.PATCH, 
+            self._build_url(account_id, project_id), 
+            json={"project": {"name": name}},
+        ))
+        return Project.from_dict(response)
+    
+    def delete(self, account_id: str, project_id: str) -> DeletedObject:
+        response: RESPONSE_TYPE = cast(RESPONSE_TYPE, self._request(
+            HttpMethod.DELTE, 
+            self._build_url(account_id, project_id),
+        ))
+        return DeletedObject(response["id"])

mailtrap/client.py

@@ -4,6 +4,7 @@
 
 import requests
 
+from mailtrap.api.projects import ProjectsApiClient
 from mailtrap.exceptions import APIError
 from mailtrap.exceptions import AuthorizationError
 from mailtrap.exceptions import ClientConfigurationError
@@ -34,10 +35,15 @@ def __init__(
 
         self._validate_itself()
 
+        self._http_client = requests.Session()
+        self._http_client.headers.update(self.headers)
+
+    @property
+    def projects_api(self) -> ProjectsApiClient:
+        return ProjectsApiClient(self._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
-        )
+        response = self._http_client.post(self.api_send_url, json=mail.api_data)
 
         if response.ok:
             data: dict[str, Union[bool, list[str]]] = response.json()

mailtrap/constants.py

@@ -0,0 +1 @@
+MAILTRAP_HOST = "mailtrap.io"

mailtrap/models/common.py

@@ -0,0 +1,3 @@
+class DeletedObject:
+    def __init__(self, id: str):
+        self.id = id

mailtrap/models/projects.py

@@ -0,0 +1,50 @@
+from typing import Any, List, Dict
+
+
+class ShareLinks:
+    def __init__(self, admin: str, viewer: str):
+        self.admin = admin
+        self.viewer = viewer
+
+
+class Permissions:
+    def __init__(
+        self, 
+        can_read: bool, 
+        can_update: bool, 
+        can_destroy: bool, 
+        can_leave: bool
+    ):
+        self.can_read = can_read
+        self.can_update = can_update
+        self.can_destroy = can_destroy
+        self.can_leave = can_leave
+
+
+class Project:
+    def __init__(
+        self, 
+        id: str, 
+        name: str, 
+        share_links: ShareLinks, 
+        inboxes: List[Dict[str, Any]], 
+        permissions: Permissions
+    ):
+        self.id = id
+        self.name = name
+        self.share_links = share_links
+        self.inboxes = inboxes
+        self.permissions = permissions
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "Project":
+        share_links = ShareLinks(**data["share_links"])
+        permissions = Permissions(**data["permissions"])
+        inboxes = data.get("inboxes", [])
+        return cls(
+            id=data["id"],
+            name=data["name"],
+            share_links=share_links,
+            inboxes=inboxes,
+            permissions=permissions,
+        )