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

The HTTP client for PokeLance.

Parameters:
  • client (PokeLance) –

    The client that this HTTP client is for.

  • cache_size (int) –

    The size of the cache.

  • session (Optional[ClientSession], default: None ) –

    The session to use for the HTTP client.
Attributes:
  • 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.

Initializes the HTTP client.

Parameters:
  • cache_size (int) –

    The size of the cache.

  • client (PokeLance) –

    The client that this HTTP client is for.

  • session (Optional[ClientSession], default: None ) –

    The session to use for the HTTP client.
Returns:
Source code in pokelance/http/__init__.py
Python
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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.

    Returns
    -------
    pokelance.http.HttpClient
        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]] = []

cache: Cache property ⚓︎

The cache to use for the HTTP client.

Returns:

close() async ⚓︎

Closes the HTTP client.

Source code in pokelance/http/__init__.py
Python
110
111
112
113
114
115
116
117
async def close(self) -> None:
    """Closes the HTTP client."""
    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:
        await self.session.close()

connect() async ⚓︎

Connects the HTTP client.

Source code in pokelance/http/__init__.py
Python
119
120
121
122
123
124
125
126
async def connect(self) -> None:
    """Connects the HTTP client."""
    if self.session is None:
        self.session = self.session or aiohttp.ClientSession()
    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:
  • url (str) –

    The URL to load the audio from.
Returns:
Raises:
Source code in pokelance/http/__init__.py
Python
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
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:
  • url (str) –

    The URL to load the image from.
Returns:
Raises:
Source code in pokelance/http/__init__.py
Python
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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:
  • float –

    The latency of the PokeAPI.
Source code in pokelance/http/__init__.py
Python
223
224
225
226
227
228
229
230
231
232
233
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:
  • route (Route) –

    The route to use for the request.
Returns:
  • Any –

    The response from the PokeAPI.
Raises:
Source code in pokelance/http/__init__.py
Python
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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()