borbann-platform/backend-api
1
0
Fork 0
mirror of https://github.com/borbann-platform/backend-api.git synced 2025-12-19 12:44:04 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: use generic typehint style

Browse Source
This commit is contained in:
Sosokker 2025-05-12 15:12:42 +07:00
parent 6a5fdf2c09
commit ec62a75b15
1 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pipeline/ingestion/adapters/api_adapter.py
View File

@ -2,7 +2,7 @@
API adapter to fetch JSON data from HTTP endpoints. API adapter to fetch JSON data from HTTP endpoints.
""" """
from typing import List, Dict, Any, Optional from typing import Any
import requests import requests
from requests.adapters import HTTPAdapter from requests.adapters import HTTPAdapter
@ -20,9 +20,9 @@ class ApiAdapter(DataSourceAdapter):
def __init__( def __init__(
self, self,
url: str, url: str,
headers: Optional[Dict[str, str]] = None, headers: dict[str, str] | None = None,
timeout: float = 30, timeout: float = 30,
token: Optional[str] = None, token: str | None = None,
): ):
""" """
Initialize the API adapter. Initialize the API adapter.
@ -50,7 +50,7 @@ class ApiAdapter(DataSourceAdapter):
total=3, total=3,
backoff_factor=0.3, backoff_factor=0.3,
status_forcelist=[500, 502, 503, 504], status_forcelist=[500, 502, 503, 504],
allowed_methods=["GET"] allowed_methods=["GET"],
) )
adapter = HTTPAdapter(max_retries=retries) adapter = HTTPAdapter(max_retries=retries)
session.mount("https://", adapter) session.mount("https://", adapter)
@ -58,7 +58,7 @@ class ApiAdapter(DataSourceAdapter):
logger.debug("HTTP session initialized with retry strategy.") logger.debug("HTTP session initialized with retry strategy.")
return session return session
def fetch(self) -> List[Dict[str, Any]]: def fetch(self) -> list[dict[str, Any]]:
""" """
Perform a GET request and return JSON data as a list of records. Perform a GET request and return JSON data as a list of records.