Client

pokelance.http.HttpClient(*, cache_size, client, session=None)

The HTTP client for PokeLance.

Parameters:

Name	Type	Description	Default
client	'PokeLance'	The client that this HTTP client is for.	required
cache_size	int	The size of the cache.	required
session	Optional[ClientSession]	The session to use for the HTTP client. If not provided, one is created internally on the first request and owned by this client.	None

Attributes:

Name	Type	Description
session	ClientSession	The session to use for the HTTP client.
_is_ready	bool	Whether the HTTP client is ready.
_cache	Cache	The cache to use for the HTTP client.
_client	PokeLance	The client that this HTTP client is for.
_tasks_queue	List[Task]	The queue for the tasks.
_session_owner	bool	Whether the session was created internally (True) or passed in by the caller (False). Only owned sessions are closed by close().

Initializes the HTTP client.

Parameters:

Name	Type	Description	Default
cache_size	int	The size of the cache.	required
client	'PokeLance'	The client that this HTTP client is for.	required
session	Optional[ClientSession]	The session to use for the HTTP client.	None

Source code in pokelance/http/__init__.py

def __init__(
    self, *, cache_size: int, client: "PokeLance", session: t.Optional[aiohttp.ClientSession] = None
) -> None:
    """Initializes the HTTP client.

    Parameters
    ----------
    cache_size: int
        The size of the cache.
    client: pokelance.PokeLance
        The client that this HTTP client is for.
    session: aiohttp.ClientSession
        The session to use for the HTTP client.
    """
    self._client = client
    self.session = session
    self._is_ready = False
    self._cache = Cache(max_size=cache_size, client=self._client)
    self._tasks_queue: t.List[asyncio.Task[None]] = []
    self._session_owner = session is None

cache property

The cache to use for the HTTP client.

Returns:

Type	Description
Cache	The cache.

logger property

The logger to use for the HTTP client.

Returns:

Type	Description
Union[Logger, Logger]	The logger.

close() async

Closes the HTTP client.

Only closes the underlying session if it was created internally. A session passed in by the caller is left open, since the caller owns its lifecycle and may still be using it elsewhere.

Source code in pokelance/http/__init__.py

async def close(self) -> None:
    """Closes the HTTP client.

    Only closes the underlying session if it was created internally.
    A session passed in by the caller is left open, since the caller
    owns its lifecycle and may still be using it elsewhere.
    """
    for task in self._tasks_queue:
        if not task.done():
            task.cancel()
            self._client.logger.warning(f"Cancelled task {task.get_name()}")
    if self.session and self._session_owner:
        await self.session.close()
    elif self.session:
        self._client.logger.debug("Session was provided externally, not closing it.")

connect() async

Connects the HTTP client.

Source code in pokelance/http/__init__.py

async def connect(self) -> None:
    """Connects the HTTP client."""
    if self.session is None:
        self.session = aiohttp.ClientSession()
        self._session_owner = True
    if not self._is_ready:
        if self._client.cache_endpoints:
            await self._schedule_tasks()
        self._is_ready = True

load_audio(url) async

Loads an audio from the url.

Parameters:

Name	Type	Description	Default
url	str	The URL to load the audio from.	required

Returns:

Type	Description
bytes	The audio.

Raises:

Type	Description
AudioNotFound	The audio was not found.

Source code in pokelance/http/__init__.py

async def load_audio(self, url: str) -> bytes:
    """
    Loads an audio from the url.

    Parameters
    ----------
    url: str
        The URL to load the audio from.

    Returns
    -------
    bytes
        The audio.

    Raises
    ------
    pokelance.exceptions.AudioNotFound
        The audio was not found.
    """
    await self.connect()
    _cry_formats = ("ogg", "wav", "mp3")
    if self.session is not None:
        async with self.session.get(url) as response:
            is_cry = any(f_ in response.content_type for f_ in _cry_formats)
            if 300 > response.status >= 200 and is_cry:
                self._client.logger.debug(f"Request to {url} was successful.")
                return await response.read()
            else:
                self._client.logger.error(f"Request to {url} was unsuccessful.")
                message = f"Request to {url} was unsuccessful or the URL is not a cry."
                raise AudioNotFound(f"{message} ({response.content_type})", Route(), response.status)
    return b""

load_image(url) async

Loads an image from the url.

Parameters:

Name	Type	Description	Default
url	str	The URL to load the image from.	required

Returns:

Type	Description
bytes	The image.

Raises:

Type	Description
ImageNotFound	The image was not found.

Source code in pokelance/http/__init__.py

async def load_image(self, url: str) -> bytes:
    """Loads an image from the url.

    Parameters
    ----------
    url: str
        The URL to load the image from.

    Returns
    -------
    bytes
        The image.

    Raises
    ------
    pokelance.exceptions.ImageNotFound
        The image was not found.
    """
    await self.connect()
    _image_formats = ("png", "jpg", "jpeg", "gif", "webp", "svg")
    if self.session is not None:
        async with self.session.get(url) as response:
            is_image = any(f_ in response.content_type for f_ in _image_formats)
            if 300 > response.status >= 200 and is_image:
                self._client.logger.debug(f"Request to {url} was successful.")
                return await response.read()
            else:
                self._client.logger.error(f"Request to {url} was unsuccessful.")
                message = f"Request to {url} was unsuccessful or the URL is not an image."
                raise ImageNotFound(f"{message} ({response.content_type})", Route(), response.status)
    return b""

ping() async

Pings the PokeAPI and returns the latency.

Returns:

Type	Description
float	The latency of the PokeAPI.

Source code in pokelance/http/__init__.py

async def ping(self) -> float:
    """Pings the PokeAPI and returns the latency.

    Returns
    -------
    float
        The latency of the PokeAPI.
    """
    start = time.perf_counter()
    await self.request(Route())
    return time.perf_counter() - start

request(route) async

Makes a request to the PokeAPI.

Parameters:

Name	Type	Description	Default
route	Route	The route to use for the request.	required

Returns:

Type	Description
Any	The response from the PokeAPI.

Raises:

Type	Description
HTTPException	An error occurred while making the request.

Source code in pokelance/http/__init__.py

async def request(self, route: Route) -> t.Any:
    """Makes a request to the PokeAPI.

    Parameters
    ----------
    route: pokelance.http.Route
        The route to use for the request.

    Returns
    -------
    t.Any
        The response from the PokeAPI.

    Raises
    ------
    pokelance.exceptions.HTTPException
        An error occurred while making the request.
    """
    await self.connect()
    if self.session is not None:
        async with self.session.request(route.method, route.url, params=route.payload) as response:
            if 300 > response.status >= 200:
                self._client.logger.debug(f"Request to {route.url} was successful.")
                return await response.json()
            else:
                self._client.logger.error(f"Request to {route.url} was unsuccessful.")
                raise HTTPException(str(response.reason), route, response.status).create()
    else:
        raise HTTPException("No session was provided.", route, -1).create()

Comments