Fix issue #22: Add ContactImportsApi, related models, tests, examples

examples/contacts/contact_imports.py

@@ -0,0 +1,31 @@
+import mailtrap as mt
+from mailtrap.models.contacts import ContactImport
+
+API_TOKEN = "YOU_API_TOKEN"
+ACCOUNT_ID = "YOU_ACCOUNT_ID"
+
+client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID)
+contact_imports_api = client.contacts_api.contact_imports
+
+
+def import_contacts(contacts: list[mt.ImportContactParams]) -> ContactImport:
+    return contact_imports_api.import_contacts(contacts=contacts)
+
+
+def get_contact_import(import_id: int) -> ContactImport:
+    return contact_imports_api.get_by_id(import_id)
+
+
+if __name__ == "__main__":
+    contact_import = import_contacts(
+        contacts=[
+            mt.ImportContactParams(
+                email="[email protected]",
+                fields={"first_name": "Test", "last_name": "Test"},
+            )
+        ]
+    )
+    print(contact_import)
+
+    contact_import = get_contact_import(contact_import.id)
+    print(contact_import)

mailtrap/__init__.py

@@ -7,6 +7,7 @@
 from .models.contacts import ContactListParams
 from .models.contacts import CreateContactFieldParams
 from .models.contacts import CreateContactParams
+from .models.contacts import ImportContactParams
 from .models.contacts import UpdateContactFieldParams
 from .models.contacts import UpdateContactParams
 from .models.mail import Address

mailtrap/api/contacts.py

@@ -1,4 +1,5 @@
 from mailtrap.api.resources.contact_fields import ContactFieldsApi
+from mailtrap.api.resources.contact_imports import ContactImportsApi
 from mailtrap.api.resources.contact_lists import ContactListsApi
 from mailtrap.api.resources.contacts import ContactsApi
 from mailtrap.http import HttpClient
@@ -17,6 +18,10 @@ def contact_fields(self) -> ContactFieldsApi:
     def contact_lists(self) -> ContactListsApi:
         return ContactListsApi(account_id=self._account_id, client=self._client)
 
+    @property
+    def contact_imports(self) -> ContactImportsApi:
+        return ContactImportsApi(account_id=self._account_id, client=self._client)
+
     @property
     def contacts(self) -> ContactsApi:
         return ContactsApi(account_id=self._account_id, client=self._client)

mailtrap/api/resources/contact_imports.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+from mailtrap.http import HttpClient
+from mailtrap.models.contacts import ContactImport
+from mailtrap.models.contacts import ImportContactParams
+
+
+class ContactImportsApi:
+    def __init__(self, client: HttpClient, account_id: str) -> None:
+        self._account_id = account_id
+        self._client = client
+
+    def import_contacts(self, contacts: list[ImportContactParams]) -> ContactImport:
+        response = self._client.post(
+            self._api_path(),
+            json={"contacts": [contact.api_data for contact in contacts]},
+        )
+        return ContactImport(**response)
+
+    def get_by_id(self, import_id: int) -> ContactImport:
+        response = self._client.get(self._api_path(import_id))
+        return ContactImport(**response)
+
+    def _api_path(self, import_id: Optional[int] = None) -> str:
+        path = f"/api/accounts/{self._account_id}/contacts/imports"
+        if import_id is not None:
+            return f"{path}/{import_id}"
+        return path

mailtrap/models/contacts.py

@@ -95,3 +95,29 @@ class Contact:
 @dataclass
 class ContactResponse:
     data: Contact
+
+
+class ContactImportStatus(str, Enum):
+    CREATED = "created"
+    STARTED = "started"
+    FINISHED = "finished"
+    FAILED = "failed"
+
+
+@dataclass
+class ContactImport:
+    id: int
+    status: ContactImportStatus
+    created_contacts_count: Optional[int] = None
+    updated_contacts_count: Optional[int] = None
+    contacts_over_limit_count: Optional[int] = None
+
+
+@dataclass
+class ImportContactParams(RequestParams):
+    email: str
+    fields: Optional[dict[str, Union[str, int, float, bool]]] = (
+        None  # field_merge_tag: value
+    )
+    list_ids_included: Optional[list[int]] = None
+    list_ids_excluded: Optional[list[int]] = None

tests/unit/api/test_contact_imports.py

@@ -0,0 +1,214 @@
+from typing import Any
+
+import pytest
+import responses
+
+from mailtrap.api.resources.contact_imports import ContactImportsApi
+from mailtrap.config import GENERAL_HOST
+from mailtrap.exceptions import APIError
+from mailtrap.http import HttpClient
+from mailtrap.models.contacts import ContactImport
+from mailtrap.models.contacts import ContactImportStatus
+from mailtrap.models.contacts import ImportContactParams
+from tests import conftest
+
+ACCOUNT_ID = "321"
+IMPORT_ID = 1234
+BASE_CONTACT_IMPORTS_URL = (
+    f"https://{GENERAL_HOST}/api/accounts/{ACCOUNT_ID}/contacts/imports"
+)
+
+
+@pytest.fixture
+def contact_imports_api() -> ContactImportsApi:
+    return ContactImportsApi(account_id=ACCOUNT_ID, client=HttpClient(GENERAL_HOST))
+
+
+@pytest.fixture
+def sample_contact_import_dict() -> dict[str, Any]:
+    return {
+        "id": IMPORT_ID,
+        "status": "started",
+    }
+
+
+@pytest.fixture
+def sample_finished_contact_import_dict() -> dict[str, Any]:
+    return {
+        "id": IMPORT_ID,
+        "status": "finished",
+        "created_contacts_count": 1,
+        "updated_contacts_count": 3,
+        "contacts_over_limit_count": 3,
+    }
+
+
+@pytest.fixture
+def import_contacts_params() -> list[ImportContactParams]:
+    return [
+        ImportContactParams(
+            email="[email protected]",
+            fields={"first_name": "John", "last_name": "Smith"},
+            list_ids_included=[1],
+            list_ids_excluded=[2],
+        ),
+        ImportContactParams(
+            email="[email protected]",
+            fields={"first_name": "John", "last_name": "Doe"},
+            list_ids_included=[3],
+            list_ids_excluded=[4],
+        ),
+    ]
+
+
+class TestContactImportsApi:
+
+    @pytest.mark.parametrize(
+        "status_code,response_json,expected_error_message",
+        [
+            (
+                conftest.UNAUTHORIZED_STATUS_CODE,
+                conftest.UNAUTHORIZED_RESPONSE,
+                conftest.UNAUTHORIZED_ERROR_MESSAGE,
+            ),
+            (
+                conftest.FORBIDDEN_STATUS_CODE,
+                conftest.FORBIDDEN_RESPONSE,
+                conftest.FORBIDDEN_ERROR_MESSAGE,
+            ),
+            (
+                conftest.VALIDATION_ERRORS_STATUS_CODE,
+                {
+                    "errors": [
+                        {
+                            "email": "[email protected]",
+                            "errors": {
+                                "base": [
+                                    "contacts limit reached",
+                                    "cannot import more than 50000 contacts at once",
+                                ],
+                            },
+                        }
+                    ]
+                },
+                "contacts limit reached",
+            ),
+        ],
+    )
+    @responses.activate
+    def test_import_contacts_should_raise_api_errors(
+        self,
+        contact_imports_api: ContactImportsApi,
+        import_contacts_params: list[ImportContactParams],
+        status_code: int,
+        response_json: dict,
+        expected_error_message: str,
+    ) -> None:
+        responses.post(
+            BASE_CONTACT_IMPORTS_URL,
+            status=status_code,
+            json=response_json,
+        )
+
+        with pytest.raises(APIError) as exc_info:
+            _ = contact_imports_api.import_contacts(import_contacts_params)
+
+        assert expected_error_message in str(exc_info.value)
+
+    @responses.activate
+    def test_import_contacts_should_return_started_import(
+        self,
+        contact_imports_api: ContactImportsApi,
+        import_contacts_params: list[ImportContactParams],
+    ) -> None:
+        expected_response = {
+            "id": IMPORT_ID,
+            "status": "started",
+        }
+        responses.post(
+            BASE_CONTACT_IMPORTS_URL,
+            json=expected_response,
+            status=201,
+        )
+
+        contact_import = contact_imports_api.import_contacts(import_contacts_params)
+
+        assert isinstance(contact_import, ContactImport)
+        assert contact_import.id == IMPORT_ID
+        assert contact_import.status == ContactImportStatus.STARTED
+
+    @pytest.mark.parametrize(
+        "status_code,response_json,expected_error_message",
+        [
+            (
+                conftest.UNAUTHORIZED_STATUS_CODE,
+                conftest.UNAUTHORIZED_RESPONSE,
+                conftest.UNAUTHORIZED_ERROR_MESSAGE,
+            ),
+            (
+                conftest.FORBIDDEN_STATUS_CODE,
+                conftest.FORBIDDEN_RESPONSE,
+                conftest.FORBIDDEN_ERROR_MESSAGE,
+            ),
+            (
+                conftest.NOT_FOUND_STATUS_CODE,
+                conftest.NOT_FOUND_RESPONSE,
+                conftest.NOT_FOUND_ERROR_MESSAGE,
+            ),
+        ],
+    )
+    @responses.activate
+    def test_get_contact_import_should_raise_api_errors(
+        self,
+        contact_imports_api: ContactImportsApi,
+        status_code: int,
+        response_json: dict,
+        expected_error_message: str,
+    ) -> None:
+        responses.get(
+            f"{BASE_CONTACT_IMPORTS_URL}/{IMPORT_ID}",
+            status=status_code,
+            json=response_json,
+        )
+
+        with pytest.raises(APIError) as exc_info:
+            contact_imports_api.get_by_id(IMPORT_ID)
+
+        assert expected_error_message in str(exc_info.value)
+
+    @responses.activate
+    def test_get_contact_import_should_return_started_import(
+        self, contact_imports_api: ContactImportsApi, sample_contact_import_dict: dict
+    ) -> None:
+        responses.get(
+            f"{BASE_CONTACT_IMPORTS_URL}/{IMPORT_ID}",
+            json=sample_contact_import_dict,
+            status=200,
+        )
+
+        contact_import = contact_imports_api.get_by_id(IMPORT_ID)
+
+        assert isinstance(contact_import, ContactImport)
+        assert contact_import.id == IMPORT_ID
+        assert contact_import.status == ContactImportStatus.STARTED
+
+    @responses.activate
+    def test_get_contact_import_should_return_finished_import(
+        self,
+        contact_imports_api: ContactImportsApi,
+        sample_finished_contact_import_dict: dict,
+    ) -> None:
+        responses.get(
+            f"{BASE_CONTACT_IMPORTS_URL}/{IMPORT_ID}",
+            json=sample_finished_contact_import_dict,
+            status=200,
+        )
+
+        contact_import = contact_imports_api.get_by_id(IMPORT_ID)
+
+        assert isinstance(contact_import, ContactImport)
+        assert contact_import.id == IMPORT_ID
+        assert contact_import.status == ContactImportStatus.FINISHED
+        assert contact_import.created_contacts_count == 1
+        assert contact_import.updated_contacts_count == 3
+        assert contact_import.contacts_over_limit_count == 3

tests/unit/models/test_contacts.py

@@ -3,6 +3,7 @@
 from mailtrap.models.contacts import ContactListParams
 from mailtrap.models.contacts import CreateContactFieldParams
 from mailtrap.models.contacts import CreateContactParams
+from mailtrap.models.contacts import ImportContactParams
 from mailtrap.models.contacts import UpdateContactFieldParams
 from mailtrap.models.contacts import UpdateContactParams
 
@@ -120,3 +121,29 @@ def test_update_contact_params_api_data_should_return_correct_dicts(
             "list_ids_excluded": [1],
             "unsubscribed": False,
         }
+
+
+class TestImportContactParams:
+    def test_import_contact_params_api_data_should_exclude_none_values(
+        self,
+    ) -> None:
+        params = ImportContactParams(email="[email protected]")
+        api_data = params.api_data
+        assert api_data == {"email": "[email protected]"}
+
+    def test_import_contact_params_api_data_should_return_correct_dicts(
+        self,
+    ) -> None:
+        params = ImportContactParams(
+            email="[email protected]",
+            fields={"first_name": "Test"},
+            list_ids_included=[1],
+            list_ids_excluded=[2],
+        )
+        api_data = params.api_data
+        assert api_data == {
+            "email": "[email protected]",
+            "fields": {"first_name": "Test"},
+            "list_ids_included": [1],
+            "list_ids_excluded": [2],
+        }