70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
import httpx
|
|
|
|
class BaseAPI():
|
|
def __init__(self):
|
|
self._client = httpx.AsyncClient()
|
|
self.access_token = None # Placeholder, will be set by derived class or external config
|
|
|
|
async def __aenter__(self):
|
|
"""Allows using the client with async with."""
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
"""Ensures the client is closed when exiting async with."""
|
|
await self.close()
|
|
|
|
async def close(self):
|
|
"""Closes the underlying asynchronous HTTP client."""
|
|
if self._client: # Ensure _client exists before trying to close
|
|
await self._client.close()
|
|
|
|
async def _post(self, endpoint: str, data: dict = None):
|
|
"""
|
|
Asynchronous helper method for making POST requests.
|
|
This method will now implicitly use the _request method,
|
|
so authentication logic is centralized.
|
|
"""
|
|
return await self._request("POST", endpoint, json=data)
|
|
|
|
async def _get(self, endpoint: str):
|
|
"""
|
|
Asynchronous helper method for making GET requests.
|
|
This method will now implicitly use the _request method,
|
|
so authentication logic is centralized.
|
|
"""
|
|
return await self._request("GET", endpoint)
|
|
|
|
async def _request(self, method, endpoint, **kwargs):
|
|
"""
|
|
Helper method to make an asynchronous HTTP request.
|
|
This is where the access_token will be injected.
|
|
|
|
Args:
|
|
method (str): The HTTP method (e.g., 'GET', 'POST').
|
|
endpoint (str): The API endpoint path (e.g., '/systems').
|
|
**kwargs: Additional arguments for httpx.AsyncClient.request.
|
|
|
|
Returns:
|
|
dict: The JSON response from the API.
|
|
|
|
Raises:
|
|
httpx.HTTPStatusError: If the request returns a non-2xx status code.
|
|
httpx.RequestError: For other request-related errors.
|
|
"""
|
|
url = f"{self.base_url}{endpoint}" # base_url must be defined in derived class or passed to BaseAPI
|
|
|
|
headers = kwargs.pop("headers", {})
|
|
if self.access_token: # Check if access_token is set
|
|
headers["Authorization"] = f"Bearer {self.access_token}"
|
|
kwargs["headers"] = headers
|
|
|
|
try:
|
|
response = await self._client.request(method, url, **kwargs)
|
|
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
|
|
return response.json()
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"HTTP error occurred: {e}")
|
|
raise
|
|
except httpx.RequestError as e:
|
|
print(f"An error occurred while requesting {e.request.url!r}: {e}")
|
|
raise |