#MPT-12324 HTTP layer with Httpx

Author: albertsola

mpt_api_client/client.py

@@ -0,0 +1,35 @@
+import httpx
+
+
+class MPTClient(httpx.Client):
+    """A client for interacting with Softwareone Marketplace Platform API."""
+
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        api_token: str,
+        timeout: float = 5.0,
+        retries: int = 0,
+    ):
+        """
+        Initializes the MPTClient.
+
+        Args:
+            base_url: The base URL of the API.
+            api_token: The API token.
+            timeout: The timeout for requests.
+            retries: The number of retries for requests. If set to 0, no retries are performed.
+
+        """
+        self.api_token = api_token
+        base_headers = {
+            "User-Agent": "swo-marketplace-client/1.0",
+            "Authorization": f"Bearer {api_token}",
+        }
+        super().__init__(
+            base_url=base_url,
+            headers=base_headers,
+            timeout=timeout,
+            transport=httpx.HTTPTransport(retries=retries),
+        )

mpt_api_client/http/client.py

@@ -19,7 +19,9 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
     "Topic :: Utilities",
 ]
-dependencies = []
+dependencies = [
+    "httpx>=0.28.1"
+]
 
 [dependency-groups]
 dev = [

pyproject.toml

@@ -0,0 +1,64 @@
+import pytest
+from httpx import ConnectTimeout, MockTransport, Request, Response, codes
+
+from mpt_api_client.http.client import MPTClient
+
+API_TOKEN = "test-token"
+API_URL = "https://api.example.com"
+
+
+@pytest.fixture
+def mock_transport() -> MockTransport:
+    def handler_request(request: Request):  # noqa: WPS430
+        if request.url.path == "/":
+            return Response(codes.OK, json={"message": "Hello, World!"})
+        if request.url.path == "/timeout":
+            raise ConnectTimeout("Mock Timeout")
+        return Response(codes.NOT_FOUND, json={"message": "Not Found"})
+
+    return MockTransport(handler=handler_request)
+
+
+@pytest.fixture
+def mock_api_client(mocker, mock_transport: MockTransport):
+    transport_mock = mocker.patch("mpt_api_client.http.client.httpx.HTTPTransport")
+    transport_mock.return_value = mock_transport
+    return MPTClient(base_url=API_URL, api_token=API_TOKEN)
+
+
+def test_mpt_client_initialization():
+    client = MPTClient(base_url=API_URL, api_token=API_TOKEN)
+
+    assert client.api_token == API_TOKEN
+    assert client.base_url == API_URL
+
+
+def test_mpt_client_headers():
+    client = MPTClient(base_url=API_URL, api_token=API_TOKEN)
+
+    assert client.headers["Authorization"] == "Bearer test-token"
+    assert client.headers["User-Agent"] == "swo-marketplace-client/1.0"
+
+
+def test_mpt_client_timeout_and_retries():
+    client = MPTClient(
+        base_url=API_URL,
+        api_token=API_TOKEN,
+        timeout=12,  # noqa: WPS432
+        retries=2,  # noqa: WPS432
+    )
+
+    assert client._timeout.connect == 12  # noqa: WPS432, SLF001
+
+
+def test_mock(mock_api_client: MPTClient):
+    success_response = mock_api_client.get("/")
+
+    with pytest.raises(ConnectTimeout):
+        mock_api_client.get("/timeout")
+
+    not_found_response = mock_api_client.get("/not-found")
+
+    assert success_response.status_code == codes.OK
+    assert success_response.json() == {"message": "Hello, World!"}
+    assert not_found_response.status_code == codes.NOT_FOUND