Connector

cyhole.core.client

APIClientInterface

The following abstract class defines a general Client API. A client is used in order to connect and interact with an external API.

The key method of an API client is the api function that is used to execute the actual requests.

api abstractmethod

api(
    type: str,
    url: str,
    *args: tuple,
    **kwargs: dict[str, Any]
) -> (
    requests.Response
    | Coroutine[None, None, requests.Response]
)

Function in charge to execute a request to an API endpoint.

Parameters:

Name	Type	Description	Default
type	str	request's type (RequestType).	required
url	str	API endpoint.	required
args	tuple	additional input parameters to provided.	()
kwargs	dict[str, Any]	additional input parameters to provided.	{}

Returns:

Type	Description
requests.Response | Coroutine[None, None, requests.Response]	The response object structured as request library. Observe that the return type could be a Coroutine in case of implementation of an async client.

Raises:

Type	Description
RequestTypeNotSupported	if the request type is not a valid value of RequestType.
AuthorizationAPIKeyError	if the response return a 401 code.

Source code in src/cyhole/core/client.py

@abc.abstractmethod
def api(self, type: str, url: str, *args: tuple, **kwargs: dict[str, Any]) -> requests.Response  | Coroutine[None, None, requests.Response]:
    """
        Function in charge to execute a request to an API endpoint.

        Parameters:
            type: request's type ([`RequestType`][cyhole.core.param.RequestType]).
            url: API endpoint.
            args: additional input parameters to provided.
            kwargs: additional input parameters to provided.

        Returns:
            The response object structured as `request` library. Observe that the return type could be a `Coroutine` in case of implementation  of an `async` client.

        Raises:
            RequestTypeNotSupported: if the request type is not a valid value of [`RequestType`][cyhole.core.param.RequestType].
            AuthorizationAPIKeyError: if the response return a 401 code.
    """
    raise NotImplementedError

APIClient

APIClient(
    interaction: Interaction, headers: Any | None = None
)

Bases: APIClientInterface

This client is designed to manage all the REST calls to an external API by using requests module.

Info

As a conseguence of using requests module, this client can be used to perform the API requests in synchronous logic and achieve the parallelism by using threads.

Use this class as middlelayer to manage all the requests to an external API. By default, all new Interaction should have the synchronous client that inherits from this class.

During the creation of the object is possible to specify some global configurations.

Parameters:

Name	Type	Description	Default
headers	Any | None	headers used globally in all API requests.	None

Source code in src/cyhole/core/client.py

def __init__(self, interaction: Interaction, headers: Any | None = None) -> None:
    self._interaction = interaction
    self.headers = headers
    return

AsyncAPIClient

AsyncAPIClient(
    interaction: Interaction, headers: Any | None = None
)

Bases: APIClientInterface

This client is designed to manage all the REST calls to an external API by using aiohttp module.

Info

As a conseguence of using aiohttp module, this client can be used to perform the API requests in asynchronous logic of async paradigm.

Use this class as middlelayer to manage all the requests to an external API. By default, all new Interaction should have the asynchronous client that inherits from this class.

During the creation of the object is possible to specify some global configurations.

Parameters:

Name	Type	Description	Default
headers	Any | None	headers used globally in all API requests.	None

Source code in src/cyhole/core/client.py

def __init__(self, interaction: Interaction, headers: Any | None = None) -> None:
    self._session: aiohttp.ClientSession | None = None
    self._interaction = interaction
    self.headers = headers
    return

__aenter__ async

__aenter__()

Open a new session.

Source code in src/cyhole/core/client.py

async def __aenter__(self):
    """Open a new session."""
    self.connect()
    return self

__aexit__ async

__aexit__(exc_type, exc_value, exc_traceback)

Exits from the session.

Source code in src/cyhole/core/client.py

async def __aexit__(self, exc_type, exc_value, exc_traceback):
    """Exits from the session."""
    await self.close()
    return

is_connected

is_connected() -> bool

Check if the session is available.

Source code in src/cyhole/core/client.py

def is_connected(self) -> bool:
    """Check if the session is available."""
    if self._session is None:
        return False
    return not self._session.closed

connect

connect() -> None

Init a new session.

Source code in src/cyhole/core/client.py

def connect(self) -> None:
    """Init a new session."""
    if self._session is None:
        self._session = aiohttp.ClientSession()
    return

close async

close() -> None

Close current available session.

Source code in src/cyhole/core/client.py

async def close(self) -> None:
    """Close current available session."""

    if self._session is None:
        raise AsyncClientAPISessionNotAvailable("No session currently available.")

    # close and reset
    await self._session.close()
    self._session = None
    return

_to_requests_response async

_to_requests_response(
    response: aiohttp.ClientResponse,
) -> requests.Response

This internal function is used to convert a response obtained from aiohttp to the one from requests library.

Parameters:

Name	Type	Description	Default
response	aiohttp.ClientResponse	response from aiohttp.	required

Return

response from requests.

Source code in src/cyhole/core/client.py

async def _to_requests_response(self, response: aiohttp.ClientResponse) -> requests.Response:
    """
        This internal function is used to convert a response obtained 
        from `aiohttp` to the one from `requests` library.

        Parameters:
            response: response from `aiohttp`.

        Return:
            response from `requests`.
    """
    _response = requests.Response()
    _response.url = str(response.url)
    _response.headers = requests.structures.CaseInsensitiveDict(response.headers)
    _response.status_code = response.status
    _response.reason = response.reason if response.reason else "Not Found"
    _response.encoding = response.get_encoding()
    _response._content = await response.read()

    return _response

_clean_params

_clean_params(params: dict[str, Any]) -> dict[str, Any]

The aiohttp requires to remove the keys with None value. This function takes as input the original params and returns a new object removed from Nonekeys.

arameters

params: original params.

Return

new params removed from the keys with Nonevaliues.

Source code in src/cyhole/core/client.py

def _clean_params(self, params: dict[str, Any]) -> dict[str, Any]:
    """
        The `aiohttp` requires to remove the keys with `None` value. 
        This function takes as input the original `params` and returns a new 
        object removed from `None`keys.

        arameters:
            params: original params.

        Return:
            new params removed from the keys with `None`valiues.
    """
    return {key: value for key, value in params.items() if value is not None}