Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 15 additions & 25 deletions docs/api_callers.rst
Original file line number Diff line number Diff line change
Expand Up @@ -6,26 +6,22 @@ The :py:mod:`comb_utils.lib.api_callers` module provides base classes for making

The API callers wrap the :py:class:`requests` library, which is a popular library for making HTTP requests in Python. The API callers handle the details of making the requests and parsing the responses, whether successful or failed, allowing you to focus on the logic of your application.

.. attention::

The current implementation requires you to extend the bases classes by at least overriding :code:`_set_url` to set the API URL you are querying. This will soon be deprecated. The plan is to allow you to pass the URL in at :code:`__init__` time, but this is not yet implemented. See https://github.com/crickets-and-comb/comb_utils/issues/38.

Base classes
############

The main classes in this module are:

- :py:class:`comb_utils.lib.api_callers.BaseCaller`: A base class for making API calls with customizable headers and parameters. If using this, you will need to extend it and override :py:meth:`comb_utils.lib.api_callers.BaseCaller._set_request_call`, but the following classes are recommended.
- :py:class:`comb_utils.lib.api_callers.BaseCaller`: A base class for making API calls with customizable headers and parameters. If using this, you will need to extend it and override :py:meth:`comb_utils.lib.api_callers.BaseCaller._request_call`, but the following classes are recommended.

- :py:class:`comb_utils.lib.api_callers.BaseGetCaller`: A base class for making GET API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making GET requests.
- :py:class:`comb_utils.lib.api_callers.GetCaller`: A base class for making GET API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making GET requests.

- :py:class:`comb_utils.lib.api_callers.BasePostCaller`: A base class for making POST API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making POST requests.
- :py:class:`comb_utils.lib.api_callers.PostCaller`: A base class for making POST API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making POST requests.

- :py:class:`comb_utils.lib.api_callers.BaseDeleteCaller`: A base class for making DELETE API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making DELETE requests.
- :py:class:`comb_utils.lib.api_callers.DeleteCaller`: A base class for making DELETE API calls. This class is a subclass of :code:`BaseCaller` and provides a default implementation for making DELETE requests.

- :py:class:`comb_utils.lib.api_callers.BasePagedResponseGetter`: A base class for making paginated GET calls. This class is a subclass of :code:`BaseGetCaller` and provides a default implementation for making paginated GET requests. It returns the next page token and the response data. This is useful for APIs that return large amounts of data in multiple pages. To get the most out of this class, use in conjunction with :py:func:`comb_utils.lib.api_callers.get_responses` and :py:func:`comb_utils.lib.api_callers.concat_response_pages`, which will handle the pagination for you (see :ref:`helper-functions` below).
- :py:class:`comb_utils.lib.api_callers.PagedResponseGetter`: A base class for making paginated GET calls. This class is a subclass of :code:`GetCaller` and provides a default implementation for making paginated GET requests. It returns the next page token and the response data. This is useful for APIs that return large amounts of data in multiple pages. To get the most out of this class, use in conjunction with :py:func:`comb_utils.lib.api_callers.get_responses` and :py:func:`comb_utils.lib.api_callers.concat_response_pages`, which will handle the pagination for you (see :ref:`helper-functions` below).

The are plans to add additional classes for remaining request calls (i.e., PUT, PATCH, etc.) in the future. For now, you can use :code:`BaseCaller` and override the :code:`BaseCaller._set_request_call` method to implement these calls. See https://github.com/crickets-and-comb/comb_utils/issues/40.
There are plans to add additional classes for remaining request calls (i.e., PUT, PATCH, etc.) in the future. For now, you can use :code:`BaseCaller` and override the :code:`BaseCaller._request_call` method to implement these calls. See https://github.com/crickets-and-comb/comb_utils/issues/40.

Example usage
*************
Expand All @@ -34,23 +30,20 @@ To use the API callers, you need to create a subclass of one of the base classes

.. code:: python

from comb_utils import BaseGetCaller
from comb_utils import GetCaller

class MyAPICaller(BaseGetCaller):
class MyAPICaller(GetCaller):

# Optionally create a custom member to store the response value.
# The, override the `_handle_200` method to set this value.
target_response_value

def _set_url(self):
self.url = "https://api.example.com/data"

# Optionally override `_handle_200` to process the response.
def _handle_200(self):
super()._handle_200()
self.target_response_value = self.response_json["target_key"]

my_caller = MyCaller()
my_caller = MyAPICaller(url="https://api.example.com/data")
my_caller.call_api()
target_response_value = my_caller.target_response_value

Expand All @@ -65,17 +58,14 @@ Using a key for authentication is common in APIs. You can override the :code:`_g

.. code:: python

from comb_utils import BaseGetCaller

class MyAPICaller(BaseGetCaller):
from comb_utils import GetCaller

def _set_url(self):
self.url = "https://api.example.com/data"
class MyAPICaller(GetCaller):

def _get_API_key(self):
return my_custom_key_retrieval_function()

my_caller = MyCaller()
my_caller = MyAPICaller(url="https://api.example.com/data")
my_caller.call_api()

.. _helper-functions:
Expand All @@ -87,7 +77,7 @@ The :py:mod:`comb_utils.lib.api_callers` module provides a few helper functions

- :py:func:`comb_utils.lib.api_callers.get_response_dict`: This function takes a response object and returns a dictionary containing the status code, headers, and JSON data. This is useful for debugging and logging purposes. You may wish to use it within your own custom API caller class or within a script to generically handle response data. The base callers use this function to handle errors and timeouts.

- :py:func:`comb_utils.lib.api_callers.get_responses`: This function gets all the responses from a paginated API endpoint using the :py:class:`comb_utils.lib.api_callers.BaseDeleteCaller` class. Returns a list of all the response pages. use this in conjunction with :py:func:`comb_utils.lib.api_callers.concat_response_pages` to get all the data from a paginated API endpoint.
- :py:func:`comb_utils.lib.api_callers.get_responses`: This function gets all the responses from a paginated API endpoint using the :py:class:`comb_utils.lib.api_callers.DeleteCaller` class. Returns a list of all the response pages. use this in conjunction with :py:func:`comb_utils.lib.api_callers.concat_response_pages` to get all the data from a paginated API endpoint.

- :py:func:`comb_utils.lib.api_callers.concat_response_pages`: This function concatenates the response pages from a paginated API endpoint into a single list. This is useful for working with APIs that return large amounts of data in multiple pages. Use this in conjunction with :py:func:`comb_utils.lib.api_callers.get_responses` to get all the data from a paginated API endpoint.

Expand Down Expand Up @@ -116,7 +106,7 @@ Handling paginated responses: ``get_responses`` and ``concat_response_pages``
.. code:: python

from comb_utils import (
BasePagedResponseGetter,
PagedResponseGetter,
concat_response_pages,
get_responses,
)
Expand All @@ -127,7 +117,7 @@ Handling paginated responses: ``get_responses`` and ``concat_response_pages``
def _get_API_key(self):
return my_custom_key_retrieval_function()

class MyAPICaller(BaseKeyRetriever, BasePagedResponseGetter):
class MyAPICaller(BaseKeyRetriever, PagedResponseGetter):

# Get all the responses from a paginated API endpoint.
all_responses = get_responses(url="https://api.example.com/data", paged_response_class=MyAPICaller)
Expand Down
8 changes: 4 additions & 4 deletions src/comb_utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,12 @@

from comb_utils.lib import (
BaseCaller,
BaseDeleteCaller,
BaseGetCaller,
BasePagedResponseGetter,
BasePostCaller,
DeleteCaller,
DocString,
ErrorDocString,
GetCaller,
PagedResponseGetter,
PostCaller,
concat_response_pages,
get_response_dict,
get_responses,
Expand Down
8 changes: 4 additions & 4 deletions src/comb_utils/lib/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,10 @@

from comb_utils.lib.api_callers import (
BaseCaller,
BaseDeleteCaller,
BaseGetCaller,
BasePagedResponseGetter,
BasePostCaller,
DeleteCaller,
GetCaller,
PagedResponseGetter,
PostCaller,
concat_response_pages,
get_response_dict,
get_responses,
Expand Down
108 changes: 36 additions & 72 deletions src/comb_utils/lib/api_callers.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,6 @@
logger = logging.getLogger(__name__)


# TODO: https://github.com/crickets-and-comb/comb_utils/issues/38:
# Why are we using _set_url instead of the url property?
# Why are we using _set_request_call instead of the _request_call property?


class BaseCaller(ABC):
"""An abstract class for making API calls.

Expand All @@ -40,21 +35,15 @@ class MyGetCaller(BaseCaller):
_wait_seconds: float = _min_wait_seconds
_timeout: float = 10

def _set_request_call(self):
self._request_call = requests.get

def _set_url(self):
self._url = "https://example.com/public/v0.2b/"

def _get_API_key(self) -> str | None:
# Optionally wrap your own API key retrieval function here.
return my_custom_key_retrieval_function()
# Optionally wrap your own API key retrieval function here.
return my_custom_key_retrieval_function()

def _handle_200(self):
super()._handle_200()
self.target_response_value = self.response_json["target_key"]

my_caller = MyCaller()
my_caller = MyGetCaller(url="https://api.example.com/data")
my_caller.call_api()
target_response_value = my_caller.target_response_value

Expand All @@ -68,10 +57,6 @@ def _handle_200(self):
this run in the background and will stop it if it runs too long. It will eventually
at least crash the memory, depending on available memory, mean time to failure, and
time left in the universe.

.. note::
The `_set_request_call` and `_set_url` methods will be deprecated in favor of setting
the request call member at child class definition and passing the URL to `__init__`.
"""

# Set by object:
Expand All @@ -80,9 +65,12 @@ def _handle_200(self):
#: The response from the API call.
_response: requests.Response

# Must set in child class with _set*:
#: The requests call method. (get, post, etc.)
_request_call: _Callable[..., requests.Response]
# Must set in child class:
@property
@abstractmethod
def _request_call(self) -> _Callable[..., requests.Response]:
"""The requests call method (get, post, etc.)."""

#: The URL for the API call.
_url: str

Expand All @@ -103,32 +91,13 @@ def _handle_200(self):
_wait_decrease_scalar: float = RateLimits.WAIT_DECREASE_SECONDS

@typechecked
def __init__(self) -> None: # noqa: ANN401
"""Initialize the BaseCaller object."""
self._set_request_call()
self._set_url()

@abstractmethod
@typechecked
def _set_request_call(self) -> None:
"""Set the requests call method.

requests.get, requests.post, etc.

Raises:
NotImplementedError: If not implemented in child class.
"""
raise NotImplementedError
def __init__(self, url: str) -> None: # noqa: ANN401
"""Initialize the BaseCaller object.

@abstractmethod
@typechecked
def _set_url(self) -> None:
"""Set the URL for the API call.

Raises:
NotImplementedError: If not implemented in child class.
Args:
url: The URL for the page. (Optionally contains nextPageToken.)
"""
raise NotImplementedError
self._url = url

@typechecked
def call_api(self) -> None:
Expand Down Expand Up @@ -287,51 +256,51 @@ def _increase_timeout(self) -> None:
cls._timeout = cls._timeout * self._wait_increase_scalar


class BaseGetCaller(BaseCaller):
class GetCaller(BaseCaller):
"""A base class for making GET API calls.

Presets the timeout, initial wait time, and requests method.
"""

@property
def _request_call(self) -> _Callable[..., requests.Response]:
"""The requests call method."""
return requests.get

_timeout: float = RateLimits.READ_TIMEOUT_SECONDS
_min_wait_seconds: float = RateLimits.READ_SECONDS
_wait_seconds: float = _min_wait_seconds

@typechecked
def _set_request_call(self) -> None:
"""Set the requests call method to `requests.get`."""
self._request_call = requests.get


class BasePostCaller(BaseCaller):
class PostCaller(BaseCaller):
"""A base class for making POST API calls.

Presets the timeout, initial wait time, and requests method.
"""

@property
def _request_call(self) -> _Callable[..., requests.Response]:
"""The requests call method."""
return requests.post

_timeout: float = RateLimits.WRITE_TIMEOUT_SECONDS
_min_wait_seconds: float = RateLimits.WRITE_SECONDS
_wait_seconds: float = _min_wait_seconds

@typechecked
def _set_request_call(self) -> None:
"""Set the requests call method to `requests.post`."""
self._request_call = requests.post


class BaseDeleteCaller(BasePostCaller):
class DeleteCaller(PostCaller):
"""A base class for making DELETE API calls.

Presets the timeout, initial wait time, and requests method.
"""

@typechecked
def _set_request_call(self) -> None:
"""Set the requests call method to `requests.delete`."""
self._request_call = requests.delete
@property
def _request_call(self) -> _Callable[..., requests.Response]:
"""The requests call method."""
return requests.delete


class BasePagedResponseGetter(BaseGetCaller):
class PagedResponseGetter(GetCaller):
"""Class for getting paged responses."""

#: The nextPageToken returned, but called salsa to avoid bandit.
Expand All @@ -345,22 +314,17 @@ class BasePagedResponseGetter(BaseGetCaller):

@typechecked
def __init__(self, page_url: str, params: dict[str, str] | None = None) -> None:
"""Initialize the BasePagedResponseGetter object.
"""Initialize the PagedResponseGetter object.

Args:
page_url: The URL for the page. (Optionally contains nextPageToken.)
params: The dictionary of query string parameters.
"""
self._page_url = page_url
self._params = params
super().__init__()

@typechecked
def _set_url(self) -> None:
"""Set the URL for the API call to the `page_url`."""
self._check_duplicates_in_URL()
self._add_params_to_URL()
self._url = self._page_url
super().__init__(self._page_url)

@typechecked
def _check_duplicates_in_URL(self) -> None:
Expand All @@ -376,7 +340,7 @@ def _check_duplicates_in_URL(self) -> None:

@typechecked
def _add_params_to_URL(self) -> None:
"""Add query string parameters to `page_url`."""
"""Add query string parameters to url."""
if self._params:
parsed_url = urlparse(self._page_url)
query_str = parsed_url.query
Expand Down Expand Up @@ -443,7 +407,7 @@ def get_response_dict(response: requests.Response) -> dict[str, Any]:
@typechecked
def get_responses(
url: str,
paged_response_class: type[BasePagedResponseGetter],
paged_response_class: type[PagedResponseGetter],
params: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
"""Get all responses from a paginated API endpoint.
Expand Down
Loading
Loading