diff --git a/app/base_api.py b/app/base_api.py new file mode 100644 index 0000000..389aad5 --- /dev/null +++ b/app/base_api.py @@ -0,0 +1,100 @@ + + +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 diff --git a/app/drb_cdb_api.py b/app/drb_cdb_api.py index 72a2cdb..4a675fd 100644 --- a/app/drb_cdb_api.py +++ b/app/drb_cdb_api.py @@ -2,8 +2,9 @@ import httpx import json import asyncio # Import asyncio for running the example usage from drb_cdb_types import ConfigGenerator +from base_api import BaseAPI -class DRBCDBAPI: +class DRBCDBAPI(BaseAPI): """ An asynchronous Python wrapper for interacting with the FastAPI application. Uses httpx for asynchronous HTTP requests. @@ -15,79 +16,11 @@ class DRBCDBAPI: Args: base_url: The base URL of the FastAPI application (e.g., "http://localhost:8000"). """ + super().__init__() self.base_url = base_url # Use an AsyncClient for making asynchronous requests self._client = httpx.AsyncClient() - 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 - # --- OP25 Endpoints --- async def start_op25(self): diff --git a/app/server_api.py b/app/server_api.py index cd53663..010ad29 100644 --- a/app/server_api.py +++ b/app/server_api.py @@ -1,7 +1,8 @@ import httpx import json +from base_api import BaseAPI -class RadioAPIClient: +class RadioAPIClient(BaseAPI): """ A client wrapper for interacting with the Radio App Server API. Uses httpx for asynchronous HTTP requests. @@ -13,51 +14,11 @@ class RadioAPIClient: Args: base_url (str): The base URL of the server's API (default is http://localhost:5000). """ + super().__init__() self.base_url = base_url # Use an AsyncClient for making asynchronous requests self._client = httpx.AsyncClient() - 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 HTTP client.""" - await self._client.close() - - 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 - async def get_systems(self): """ Retrieves a summary list of all available radio systems.