101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
|
|
|
|
class BaseAPI():
|
|
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."""
|
|
await self._client.close()
|
|
|
|
async def _post(self, endpoint: str, data: dict = None):
|
|
"""
|
|
Asynchronous helper method for making POST requests.
|
|
|
|
Args:
|
|
endpoint: The API endpoint (e.g., "/op25/start").
|
|
data: The data to send in the request body (as a dictionary).
|
|
|
|
Returns:
|
|
The JSON response from the API.
|
|
|
|
Raises:
|
|
httpx.RequestError: If the request fails.
|
|
httpx.HTTPStatusError: If the API returns an error status code (4xx or 5xx).
|
|
"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
try:
|
|
# Use await with the asynchronous httpx client
|
|
response = await self._client.post(url, json=data)
|
|
response.raise_for_status() # Raise HTTPStatusError for bad responses (4xx or 5xx)
|
|
return response.json()
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"HTTP error occurred: {e}")
|
|
print(f"Response body: {e.response.text}") # Access response text from the exception
|
|
raise
|
|
except httpx.RequestError as e:
|
|
print(f"Request to '{url}' failed: {e}")
|
|
raise
|
|
|
|
async def _get(self, endpoint: str):
|
|
"""
|
|
Asynchronous helper method for making GET requests.
|
|
|
|
Args:
|
|
endpoint: The API endpoint (e.g., "/op25/status").
|
|
|
|
Returns:
|
|
The JSON response from the API.
|
|
|
|
Raises:
|
|
httpx.RequestError: If the request fails.
|
|
httpx.HTTPStatusError: If the API returns an error status code (4xx or 5xx).
|
|
"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
try:
|
|
# Use await with the asynchronous httpx client
|
|
response = await self._client.get(url)
|
|
response.raise_for_status() # Raise HTTPStatusError for bad responses (4xx or 5xx)
|
|
return response.json()
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"HTTP error occurred: {e}")
|
|
print(f"Response body: {e.response.text}") # Access response text from the exception
|
|
raise
|
|
except httpx.RequestError as e:
|
|
print(f"Request to '{url}' failed: {e}")
|
|
raise
|
|
|
|
async def _request(self, method, endpoint, **kwargs):
|
|
"""
|
|
Helper method to make an asynchronous HTTP request.
|
|
|
|
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}"
|
|
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}")
|
|
# You might want to return the error response body or raise the exception
|
|
raise
|
|
except httpx.RequestError as e:
|
|
print(f"An error occurred while requesting {e.request.url!r}: {e}")
|
|
raise
|