Fix a csvimport error 'List index (0) out of bounds'

.pre-commit-config.yaml

@@ -64,7 +64,7 @@ repos:
         exclude: ^tests
 
   - repo: https://github.com/python-jsonschema/check-jsonschema
-    rev: 0.33.0
+    rev: 0.33.2
     hooks:
       - id: check-github-workflows
 
@@ -90,7 +90,7 @@ repos:
           - --ignore-init-module-imports
 
   - repo: https://github.com/pycqa/flake8
-    rev: 7.2.0
+    rev: 7.3.0
     hooks:
     -   id: flake8
         additional_dependencies:
@@ -108,14 +108,14 @@ repos:
           - --exit-zero
 
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.19.1
+    rev: v3.20.0
     hooks:
       - id: pyupgrade
         args: [--py39-plus, --keep-runtime-typing]
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
     # Ruff version.
-    rev: v0.11.8
+    rev: v0.12.2
     hooks:
       # Run the linter.
       - id: ruff
@@ -124,12 +124,13 @@ repos:
       - id: ruff-format
 
   -   repo: https://github.com/dosisod/refurb
-      rev: v2.0.0
+      rev: v2.1.0
       hooks:
       -   id: refurb
+          args: [--ignore, FURB184]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.15.0
+    rev: v1.16.1
     hooks:
     -   id: mypy
         args:
@@ -139,12 +140,12 @@ repos:
         exclude: ^samples/
 
   - repo: https://github.com/RobertCraigie/pyright-python
-    rev: v1.1.400
+    rev: v1.1.403
     hooks:
     - id: pyright
 
   - repo: https://github.com/PyCQA/bandit
-    rev: 1.8.3
+    rev: 1.8.6
     hooks:
       - id: bandit
         args: ["-c", "pyproject.toml", "-r", "."]

mailjet_rest/_version.py

@@ -1 +1 @@
-__version__ = "1.4.0"
+__version__ = "1.4.0"

mailjet_rest/client.py

@@ -253,7 +253,7 @@ def get(
 
     def create(
         self,
-        data: dict | None = None,
+        data: str | bytes | dict[Any, Any] | None = None,
         filters: Mapping[str, str | Any] | None = None,
         id: str | None = None,
         action_id: str | None = None,
@@ -264,7 +264,7 @@ def create(
         """Perform a POST request to create a new resource.
 
         Parameters:
-        - data (dict | None): The data to include in the request body.
+        - data (str | bytes | dict[Any, Any] | None): The data to include in the request body.
         - filters (Mapping[str, str | Any] | None): Filters to be applied in the request.
         - id (str | None): The ID of the specific resource to be created.
         - action_id (str | None): The specific action ID to be performed.
@@ -275,18 +275,20 @@ def create(
         Returns:
         - Response: The response object from the API call.
         """
-        json_data: str | bytes | None = None
         if self.headers.get("Content-type") == "application/json" and data is not None:
-            json_data = json.dumps(data, ensure_ascii=ensure_ascii)
+            data = json.dumps(
+                data,
+                ensure_ascii=ensure_ascii,
+            )
             if not ensure_ascii:
-                json_data = json_data.encode(data_encoding)
+                data = data.encode(data_encoding)
         return api_call(
             self._auth,
             "post",
             self._url,
             headers=self.headers,
             resource_id=id,
-            data=json_data,
+            data=data,  # type: ignore[arg-type]
             action=self.action,
             action_id=action_id,
             filters=filters,
@@ -406,7 +408,7 @@ def __getattr__(self, name: str) -> Any:
         - Endpoint: An instance of the `Endpoint` class, initialized with the constructed URL, headers, action, and authentication details.
         """
         name_regex: str = re.sub(r"[A-Z]", prepare_url, name)
-        split: list[str] = name_regex.split("_")  # noqa: RUF100, FURB184
+        split: list[str] = name_regex.split("_")  # noqa: RUF100
         # identify the resource
         fname: str = split[0]
         action: str | None = None

pyproject.toml

@@ -236,6 +236,8 @@ ignore = [
     "S311",  # S311 Standard pseudo-random generators are not suitable for cryptographic purposes
     # TODO: T201 Replace `print` with logging functions
     "T201",  # T201 `print` found
+    "PLC0207",  # PLC0207 Accessing only the first or last element of `str.split()` without setting `maxsplit=1`
+
 ]
 
 

test.py

@@ -4,7 +4,9 @@
 import random
 import string
 import unittest
+from pathlib import Path
 from typing import Any
+from typing import ClassVar
 
 from mailjet_rest import Client
 
@@ -216,5 +218,106 @@ def test_user_agent(self) -> None:
         self.assertEqual(self.client.config.user_agent, "mailjet-apiv3-python/v1.4.0")
 
 
+class TestCsvImport(unittest.TestCase):
+    """Tests for Mailjet API csv import functionality.
+
+    This class provides setup and teardown functionality for tests involving the
+    csv import functionality, with authentication and client initialization handled
+    in `setUp`. Each test in this suite operates with the configured Mailjet client
+    instance to simulate API interactions.
+
+    Attributes:
+    - _shared_state (dict[str, str]): A dictionary containing values taken from tests to share them in other tests.
+    """
+
+    _shared_state: ClassVar[dict[str, Any]] = {}
+
+    @classmethod
+    def get_shared(cls, key: str) -> Any:
+        """Retrieve a value from shared test state.
+
+        Parameters:
+        - key (str): The key to look up in shared state.
+
+        Returns:
+        - Any: The stored value, or None if key doesn't exist.
+        """
+        return cls._shared_state.get(key)
+
+    @classmethod
+    def set_shared(cls, key: str, value: Any) -> None:
+        """Store a value in shared test state.
+
+        Parameters:
+        - key (str): The key to store the value under.
+        - value (Any): The value to store.
+        """
+        cls._shared_state[key] = value
+
+    def setUp(self) -> None:
+        """Set up the test environment by initializing authentication credentials and the Mailjet client.
+
+        This method is called before each test to ensure a consistent testing
+        environment. It retrieves the API keys and ID_CONTACTSLIST from environment variables and
+        uses them to create an instance of the Mailjet `Client` for authenticated
+        API interactions.
+
+        Attributes:
+        - self.auth (tuple[str, str]): A tuple containing the public and private API keys obtained from the environment variables 'MJ_APIKEY_PUBLIC' and 'MJ_APIKEY_PRIVATE' respectively.
+        - self.client (Client): An instance of the Mailjet Client class, initialized with the provided authentication credentials.
+        - self.id_contactslist (str): A string of the contacts list ID from https://app.mailjet.com/contacts
+        """
+        self.auth: tuple[str, str] = (
+            os.environ["MJ_APIKEY_PUBLIC"],
+            os.environ["MJ_APIKEY_PRIVATE"],
+        )
+        self.client: Client = Client(auth=self.auth)
+        self.id_contactslist: str = os.environ["ID_CONTACTSLIST"]
+
+    def test_01_upload_the_csv(self) -> None:
+        """Test uploading a csv file.
+
+        POST https://api.mailjet.com/v3/DATA/contactslist
+        /$ID_CONTACTLIST/CSVData/text:plain
+        """
+        result = self.client.contactslist_csvdata.create(
+            id=self.id_contactslist,
+            data=Path("tests/doc_tests/files/data.csv").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(result.status_code, 200)
+
+        self.set_shared("data_id", result.json().get("ID"))
+        data_id = self.get_shared("data_id")
+        self.assertIsNotNone(data_id)
+
+    def test_02_import_csv_content_to_a_list(self) -> None:
+        """Test importing a csv content to a list.
+
+        POST https://api.mailjet.com/v3/REST/csvimport
+        """
+        data_id = self.get_shared("data_id")
+        self.assertIsNotNone(data_id)
+        data = {
+            "Method": "addnoforce",
+            "ContactsListID": self.id_contactslist,
+            "DataID": data_id,
+        }
+        result = self.client.csvimport.create(data=data)
+        self.assertEqual(result.status_code, 201)
+        self.assertIn("ID", result.json()["Data"][0])
+
+        self.set_shared("id_value", result.json()["Data"][0]["ID"])
+
+    def test_03_monitor_the_import_progress(self) -> None:
+        """Test getting a csv content import.
+
+        GET https://api.mailjet.com/v3/REST/csvimport/$importjob_ID
+        """
+        result = self.client.csvimport.get(id=self.get_shared("id_value"))
+        self.assertEqual(result.status_code, 200)
+        self.assertIn("ID", result.json()["Data"][0])
+        self.assertEqual(0, result.json()["Data"][0]["Errcount"])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/doc_tests/files/data.csv

@@ -0,0 +1,5 @@
+address,name,subscribed,vars
+bob@mywebsite.com,Bob3,TRUE,34
+jane@example.com,Janen,TRUE,21
+pete@example.com,Pete,TRUE,44
+foo@example.com,Foo,TRUE,20