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

              flowchart TD
              cyhole.core.client.APIClient[APIClient]
              cyhole.core.client.APIClientInterface[APIClientInterface]

                              cyhole.core.client.APIClientInterface --> cyhole.core.client.APIClient
                


              click cyhole.core.client.APIClient href "" "cyhole.core.client.APIClient"
              click cyhole.core.client.APIClientInterface href "" "cyhole.core.client.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

              flowchart TD
              cyhole.core.client.AsyncAPIClient[AsyncAPIClient]
              cyhole.core.client.APIClientInterface[APIClientInterface]

                              cyhole.core.client.APIClientInterface --> cyhole.core.client.AsyncAPIClient
                


              click cyhole.core.client.AsyncAPIClient href "" "cyhole.core.client.AsyncAPIClient"
              click cyhole.core.client.APIClientInterface href "" "cyhole.core.client.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