feat: update APIClient class

AlexAlexandreAlves · AlexAlexandreAlves · commit 0289d1cea288 · 2025-12-02T11:51:09.000-03:00
diff --git a/api-batch-test/src/api_client.py b/api-batch-test/src/api_client.py
@@ -1,32 +1,123 @@
+import time
+import pytest
 import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+from config import settings
+
 
 class APIClient:
-    def __init__(self, base_url):
-        self.base_url = base_url
+    """Cliente API com suporte a retry, rate-limit e formatação de URLs."""
+
+    def __init__(self, base_url: str, connect_timeout: float, read_timeout: float, retry_total: int, rate_sleep: float):
+        """
+        Inicializa o cliente API.
+
+        Args:
+            base_url: URL base da API
+            connect_timeout: Timeout de conexão em segundos
+            read_timeout: Timeout de leitura em segundos
+            retry_total: Total de tentativas para requisições
+            rate_sleep: Tempo de espera entre requisições (rate-limit)
+        """
+        self.base_url = base_url.rstrip("/")
+        self.session = requests.Session()
+        retry = Retry(
+            total=retry_total,
+            backoff_factor=0.5,
+            status_forcelist=[429, 500, 502, 503, 504],
+            allowed_methods=["GET", "POST", "PUT", "DELETE"],
+            respect_retry_after_header=True,
+        )
+        adapter = HTTPAdapter(
+            max_retries=retry, pool_connections=10, pool_maxsize=10)
+        self.session.mount("https://", adapter)
+        self.session.mount("http://", adapter)
+        self.session.headers.update({"Accept": "application/json"})
+        self.timeout = (connect_timeout, read_timeout)
+        self.rate_sleep = rate_sleep
+
+    def _build_url(self, endpoint: str, **path_params) -> str:
+        """Constrói a URL completa formatando parametros de caminho."""
+        if path_params:
+            endpoint = endpoint.format(**path_params)
+        return f"{self.base_url}/{endpoint.lstrip('/')}"
+
+    def get(self, endpoint: str, params=None, **kwargs) -> requests.Response:
+        """
+        Faz uma requisição GET.
+
+        Args:
+            endpoint: Endpoint da API (pode ter placeholders como {id})
+            params: Parâmetros de query
+            **kwargs: Parâmetros de caminho como id=123
+
+        Returns:
+            Response object
+        """
+        url = self._build_url(endpoint, **kwargs)
+        time.sleep(self.rate_sleep)
+        return self.session.get(url, params=params, timeout=self.timeout)
+
+    def post(self, endpoint: str, data=None, **kwargs) -> requests.Response:
+        """
+        Faz uma requisição POST.
+
+        Args:
+            endpoint: Endpoint da API
+            data: Dados a enviar no body (será convertido para JSON)
+            **kwargs: Parâmetros de caminho
+
+        Returns:
+            Response object
+        """
+        url = self._build_url(endpoint, **kwargs)
+        time.sleep(self.rate_sleep)
+        return self.session.post(url, json=data, timeout=self.timeout)
+
+    def put(self, endpoint: str, data=None, **kwargs) -> requests.Response:
+        """
+        Faz uma requisição PUT.
+
+        Args:
+            endpoint: Endpoint da API
+            data: Dados a enviar no body (será convertido para JSON)
+            **kwargs: Parâmetros de caminho
+
+        Returns:
+            Response object
+        """
+        url = self._build_url(endpoint, **kwargs)
+        time.sleep(self.rate_sleep)
+        return self.session.put(url, json=data, timeout=self.timeout)
 
-    def get(self, endpoint, params=None, **kwargs):
+    def delete(self, endpoint: str, **kwargs) -> requests.Response:
         """
-        Makes a GET request.
-        
+        Faz uma requisição DELETE.
+
         Args:
-            endpoint: API endpoint (can have {id} placeholders)
-            params: Query parameters
-            **kwargs: Path parameters like id=123
-        """
-        url = endpoint.format(**kwargs) if kwargs else endpoint
-        response = requests.get(f"{self.base_url}{url}", params=params)
-        return response
-
-    def post(self, endpoint, data=None):
-        response = requests.post(f"{self.base_url}{endpoint}", json=data)
-        return response
-    
-    def put(self, endpoint, data=None, **kwargs):
-        url = endpoint.format(**kwargs) if kwargs else endpoint
-        response = requests.put(f"{self.base_url}{url}", json=data)
-        return response
-    
-    def delete(self, endpoint, **kwargs):
-        url = endpoint.format(**kwargs) if kwargs else endpoint
-        response = requests.delete(f"{self.base_url}{url}")
-        return response
+            endpoint: Endpoint da API
+            **kwargs: Parâmetros de caminho
+
+        Returns:
+            Response object
+        """
+        url = self._build_url(endpoint, **kwargs)
+        time.sleep(self.rate_sleep)
+        return self.session.delete(url, timeout=self.timeout)
+
+
+@pytest.fixture(scope="session")
+def base_url() -> str:
+    return settings.BASE_URL
+
+
+@pytest.fixture(scope="session")
+def http() -> APIClient:
+    return APIClient(
+        base_url=settings.BASE_URL,
+        connect_timeout=settings.CONNECT_TIMEOUT,
+        read_timeout=settings.READ_TIMEOUT,
+        retry_total=settings.RETRY_TOTAL,
+        rate_sleep=settings.RATE_LIMIT_SLEEP,
+    )
diff --git a/api-batch-test/src/api_endpoints.py b/api-batch-test/src/api_endpoints.py
@@ -1,3 +1,3 @@
 class Endpoints:
-    ACTIVITIES = "/api/v1/Activities"
-    ACTIVITIES_BY_ID = "/api/v1/Activities/{id}"
+    PEOPLE = "/people/"
+    PEOPLE_ID = "/people/{id}"
diff --git a/api-batch-test/tests/conftest.py b/api-batch-test/tests/conftest.py
@@ -1,54 +1,4 @@
-import time
-import pytest
-import requests
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
-from config import settings
-
-
-class HttpClient:
-    """Cliente HTTP focado em testes: timeout, retry e rate-limit simples."""
-
-    def __init__(self, base_url: str, connect_timeout: float, read_timeout: float, retry_total: int, rate_sleep: float):
-        self.base_url = base_url.rstrip("/")
-        self.session = requests.Session()
-        retry = Retry(
-            total=retry_total,
-            backoff_factor=0.5,
-            status_forcelist=[429, 500, 502, 503, 504],
-            allowed_methods=["GET"],
-            respect_retry_after_header=True,
-        )
-        adapter = HTTPAdapter(
-            max_retries=retry, pool_connections=10, pool_maxsize=10)
-        self.session.mount("https://", adapter)
-        self.session.mount("http://", adapter)
-        self.session.headers.update({"Accept": "application/json"})
-        self.timeout = (connect_timeout, read_timeout)
-        self.rate_sleep = rate_sleep
-
-    def get(self, path_or_url: str, **kwargs) -> requests.Response:
-        """GET com base_url opcional e respeito a rate-limit."""
-        if path_or_url.startswith("http://") or path_or_url.startswith("https://"):
-            url = path_or_url
-        else:
-            url = f"{self.base_url}/{path_or_url.lstrip('/')}"
-        time.sleep(self.rate_sleep)  # rate-limit simples
-        return self.session.get(url, timeout=self.timeout, **kwargs)
-
-
-@pytest.fixture(scope="session")
-def base_url() -> str:
-    return settings.BASE_URL
-
-
-@pytest.fixture(scope="session")
-def http() -> HttpClient:
-    return HttpClient(
-        base_url=settings.BASE_URL,
-        connect_timeout=settings.CONNECT_TIMEOUT,
-        read_timeout=settings.READ_TIMEOUT,
-        retry_total=settings.RETRY_TOTAL,
-        rate_sleep=settings.RATE_LIMIT_SLEEP,
-    )
+from src.api_client import base_url, http
 
+# Re-exporting fixtures from src.api_client
+__all__ = ["base_url", "http"]
diff --git a/api-batch-test/tests/helpers/pagination.py b/api-batch-test/tests/helpers/pagination.py
@@ -1,7 +1,8 @@
 from typing import Dict, Generator
-from tests.conftest import HttpClient
+from src.api_client import APIClient
 
-def iter_pages(http: HttpClient, first_path: str) -> Generator[Dict, None, None]:
+
+def iter_pages(http: APIClient, first_path: str) -> Generator[Dict, None, None]:
     # Iterate through pages starting from first_path
     resp = http.get(first_path)
     resp.raise_for_status()
diff --git a/api-batch-test/tests/test_people_contract.py b/api-batch-test/tests/test_people_contract.py
@@ -2,10 +2,12 @@
 from tests.schemas.page_schema import PAGE_SCHEMA
 from tests.schemas.people_schema import PEOPLE_ITEM_SCHEMA
 import pytest
+from src.api_endpoints import Endpoints
+
 
 @pytest.mark.contract
 def test_people_page_schema(http):
-    r = http.get("/people/")
+    r = http.get(Endpoints.PEOPLE)
     r.raise_for_status()
     data = r.json()
     validate(instance=data, schema=PAGE_SCHEMA)
diff --git a/api-batch-test/tests/test_smoke.py b/api-batch-test/tests/test_smoke.py
@@ -1,7 +1,8 @@
 import pytest
+from src.api_endpoints import Endpoints
 
 @pytest.mark.smoke
 def test_api_available(http, base_url):
-    r = http.get("/people/")
+    r = http.get(Endpoints.PEOPLE)
     assert r.status_code == 200
     assert r.headers.get("Content-Type", "").startswith("application/json")
