Client

http ¶

BaseHttpClient ¶

BaseHttpClient(
    *, client: _ClientT, session: _SessionT | None = None
)

Bases: Generic[_ClientT, _SessionT, _CacheManagerT]

Base class containing shared HTTP logic, validation, and media checks.

Source code in pokelance/http/_base.py

def __init__(
    self,
    *,
    client: _ClientT,
    session: _SessionT | None = None,
) -> None:
    self._client = client
    self.session = session
    self._is_ready = False
    self._session_owner = session is None
    self._closing = False

is_closing `property` ¶

is_closing: bool

Whether the HTTP client is currently closing.

is_connected `property` ¶

is_connected: bool

Whether the HTTP client currently has an active session.

cache_manager `property` ¶

cache_manager: _CacheManagerT

The cache manager used by this HTTP client.

AsyncHttpClient ¶

AsyncHttpClient(
    *,
    cache_size: int,
    client: PokeLanceAsyncClient,
    session: AsyncSession | None = None,
)

Bases: BaseHttpClient['PokeLanceAsyncClient', AsyncSession, AsyncCacheManager]

The asynchronous HTTP client for PokeLance.

Parameters:

Name	Type	Description	Default
`client`	`PokeLanceAsyncClient`	The client that this HTTP client is for.	required
`cache_size`	`int`	The size of the cache.	required
`session`	`AsyncSession \| None`	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`

Source code in pokelance/http/_async.py

def __init__(
    self,
    *,
    cache_size: int,
    client: PokeLanceAsyncClient,
    session: niquests.AsyncSession | None = None,
) -> None:
    super().__init__(client=client, session=session)
    self._cache_manager = AsyncCacheManager(max_size=cache_size, client=self._client)
    self._loader = AsyncEndpointLoader(client=self._client)

loader `property` ¶

loader: AsyncEndpointLoader

The endpoint loader composition helper.

close `async` ¶

close() -> None

Closes the HTTP client session and cancels pending endpoint loaders.

Source code in pokelance/http/_async.py

async def close(self) -> None:
    """Closes the HTTP client session and cancels pending endpoint loaders."""
    if self._closing:
        return
    self._closing = True
    try:
        await self._loader.cancel_tasks()
        session_to_close = self.session if self._session_owner else None
        self.session = None
        if session_to_close:
            logger.debug("Closing internal async HTTP session...")
            await session_to_close.close()
    finally:
        self._is_ready = False
        self._closing = False

connect `async` ¶

connect() -> None

Connects the HTTP client and sets up the session.

Source code in pokelance/http/_async.py

async def connect(self) -> None:
    """Connects the HTTP client and sets up the session."""
    if self._closing:
        raise RuntimeError("Cannot connect while the HTTP client is closing.")
    if self.session is None:
        logger.debug("Initializing internal async HTTP session (niquests)...")
        self.session = niquests.AsyncSession(resolver="system://")
        self._session_owner = True
    if not self._is_ready:
        self._is_ready = True
        if self._client.cache_endpoints:
            await self._loader.schedule_tasks()

request `async` ¶

request(route: Route) -> dict[str, Any]

Makes an asynchronous request to the PokeAPI.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	The response from the PokeAPI parsed as JSON.

Raises:

Type	Description
`HTTPException`	An error occurred while making the request.

Source code in pokelance/http/_async.py

async def request(self, route: Route) -> dict[str, t.Any]:
    """Makes an asynchronous request to the PokeAPI.

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

    Returns
    -------
    dict[str, t.Any]
        The response from the PokeAPI parsed as JSON.

    Raises
    ------
    HTTPException
        An error occurred while making the request.
    """
    if self._closing:
        raise RuntimeError("Cannot make a request while the HTTP client is closing.")
    await self.connect()
    if self.session is not None:
        logger.debug(f"Sending {route.method} request to {route.url}")
        response = await self.session.request(route.method, route.url, params=route.payload)
        return self._validate_response(response, route)
    raise HTTPException("No session was provided.", route, -1).create()

load_image `async` ¶

load_image(url: str) -> bytes

Loads an image from the url asynchronously.

Parameters:

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

Returns:

Type	Description
`bytes`	The raw image bytes.

Source code in pokelance/http/_async.py

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

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

    Returns
    -------
    bytes
        The raw image bytes.
    """
    await self.connect()
    if self.session is not None:
        logger.debug(f"Fetching image from {url}")
        response = await self.session.get(url)
        return self._validate_image(response, url)
    return b""

load_audio `async` ¶

load_audio(url: str) -> bytes

Loads an audio from the url asynchronously.

Parameters:

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

Returns:

Type	Description
`bytes`	The raw audio bytes.

Source code in pokelance/http/_async.py

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

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

    Returns
    -------
    bytes
        The raw audio bytes.
    """
    await self.connect()
    if self.session is not None:
        logger.debug(f"Fetching audio from {url}")
        response = await self.session.get(url)
        return self._validate_audio(response, url)
    return b""

ping `async` ¶

ping() -> float

Pings the PokeAPI and returns the latency.

Source code in pokelance/http/_async.py

async def ping(self) -> float:
    """Pings the PokeAPI and returns the latency."""
    start = time.perf_counter()
    await self.request(Route())
    latency = time.perf_counter() - start
    logger.debug(f"PokeAPI async ping latency: {latency:.4f}s")
    return latency

AsyncEndpointLoader ¶

AsyncEndpointLoader(client: PokeLanceAsyncClient)

Composition helper for scheduling and tracking async endpoint pre-population tasks.

Source code in pokelance/http/_async.py

def __init__(self, client: PokeLanceAsyncClient) -> None:
    self._client = client
    self._tasks: set[asyncio.Task[None]] = set()
    self._remaining: int = 0
    self._ready_event: asyncio.Event = asyncio.Event()
    self._ready_event.set()
    self._scheduled: bool = False

is_ready `property` ¶

is_ready: bool

Whether all background loading tasks have finished.

wait_until_ready `async` ¶

wait_until_ready() -> None

Wait until all background endpoint tasks have completed.

Source code in pokelance/http/_async.py

async def wait_until_ready(self) -> None:
    """Wait until all background endpoint tasks have completed."""
    await self._ready_event.wait()

schedule_tasks `async` ¶

schedule_tasks() -> None

Schedules background endpoint-loading tasks using asyncio.create_task.

Source code in pokelance/http/_async.py

async def schedule_tasks(self) -> None:
    """Schedules background endpoint-loading tasks using asyncio.create_task."""
    if self._scheduled:
        return
    self._scheduled = True
    self._ready_event.clear()
    if not self._client.cache_endpoints:
        self._ready_event.set()
        self._client.ext_tasks.clear()
        return
    total = len(self._client.ext_tasks)
    self._remaining = total
    logger.info(f"Scheduling {total} endpoint pre-population task(s)...")
    for num, (coroutine, name) in enumerate(self._client.ext_tasks):
        message = f"Extension {name} endpoints ({num + 1}/{total})"
        task = asyncio.create_task(coro=self._load_ext(coroutine, message), name=name)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
    self._client.ext_tasks.clear()
    if self._remaining == 0:
        self._ready_event.set()

cancel_tasks `async` ¶

cancel_tasks() -> None

Cancels and awaits all in-flight endpoint loading tasks.

Source code in pokelance/http/_async.py

async def cancel_tasks(self) -> None:
    """Cancels and awaits all in-flight endpoint loading tasks."""
    tasks = [task for task in self._tasks if not task.done()]
    if tasks:
        logger.warning(f"Cancelling {len(tasks)} in-flight endpoint loading task(s)...")
        for task in tasks:
            task.cancel()
            logger.warning(f"Cancelled task {task.get_name()}")
        try:
            await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout=1.5)
        except asyncio.TimeoutError:
            logger.warning("Timed out waiting for endpoint loading tasks to cancel.")
    self._tasks.clear()
    self._scheduled = False
    self._remaining = 0
    self._ready_event.set()

SyncHttpClient ¶

SyncHttpClient(
    *,
    cache_size: int,
    client: PokeLanceSyncClient,
    session: Session | None = None,
)

Bases: BaseHttpClient['PokeLanceSyncClient', Session, SyncCacheManager]

The synchronous HTTP client for PokeLance.

Parameters:

Name	Type	Description	Default
`client`	`PokeLanceSyncClient`	The client that this HTTP client is for.	required
`cache_size`	`int`	The size of the cache.	required
`session`	`Session \| None`	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`

Source code in pokelance/http/_sync.py

def __init__(
    self,
    *,
    cache_size: int,
    client: PokeLanceSyncClient,
    session: niquests.Session | None = None,
) -> None:
    super().__init__(client=client, session=session)
    self._cache_manager = SyncCacheManager(max_size=cache_size, client=self._client)
    self._loader = SyncEndpointLoader(client=self._client)
    self._lock = threading.Lock()

loader `property` ¶

loader: SyncEndpointLoader

The endpoint loader composition helper.

close ¶

close() -> None

Closes the HTTP client and shuts down the endpoint loader thread pool.

Source code in pokelance/http/_sync.py

def close(self) -> None:
    """Closes the HTTP client and shuts down the endpoint loader thread pool."""
    with self._lock:
        if self._closing:
            return
        self._closing = True
        try:
            session_to_close = self.session if self._session_owner else None
            self.session = None
            if session_to_close:
                logger.debug("Closing internal sync HTTP session...")
                session_to_close.close()
            self._loader.shutdown()
        finally:
            self._is_ready = False
            self._closing = False

connect ¶

connect() -> None

Connects the HTTP client and sets up the session.

Source code in pokelance/http/_sync.py

def connect(self) -> None:
    """Connects the HTTP client and sets up the session."""
    with self._lock:
        if self._closing:
            raise RuntimeError("Cannot connect while the HTTP client is closing.")
        if self.session is None:
            logger.debug("Initializing internal sync HTTP session (niquests)...")
            self.session = niquests.Session(resolver="system://")
            self._session_owner = True
        if not self._is_ready:
            self._is_ready = True
            if self._client.cache_endpoints:
                self._loader.schedule_tasks()

request ¶

request(route: Route) -> dict[str, Any]

Makes a synchronous request to the PokeAPI.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	The response from the PokeAPI parsed as JSON.

Raises:

Type	Description
`HTTPException`	An error occurred while making the request.

Source code in pokelance/http/_sync.py

def request(self, route: Route) -> dict[str, t.Any]:
    """Makes a synchronous request to the PokeAPI.

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

    Returns
    -------
    dict[str, t.Any]
        The response from the PokeAPI parsed as JSON.

    Raises
    ------
    HTTPException
        An error occurred while making the request.
    """
    if self._closing:
        raise RuntimeError("Cannot make a request while the HTTP client is closing.")
    self.connect()
    if self.session is not None:
        logger.debug(f"Sending {route.method} request to {route.url}")
        response = self.session.request(route.method, route.url, params=route.payload)
        return self._validate_response(response, route)
    raise HTTPException("No session was provided.", route, -1).create()

load_image ¶

load_image(url: str) -> bytes

Loads an image from the url synchronously.

Parameters:

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

Returns:

Type	Description
`bytes`	The raw image bytes.

Source code in pokelance/http/_sync.py

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

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

    Returns
    -------
    bytes
        The raw image bytes.
    """
    self.connect()
    if self.session is not None:
        logger.debug(f"Fetching image from {url}")
        response = self.session.get(url)
        return self._validate_image(response, url)
    return b""

load_audio ¶

load_audio(url: str) -> bytes

Loads an audio from the url synchronously.

Parameters:

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

Returns:

Type	Description
`bytes`	The raw audio bytes.

Source code in pokelance/http/_sync.py

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

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

    Returns
    -------
    bytes
        The raw audio bytes.
    """
    self.connect()
    if self.session is not None:
        logger.debug(f"Fetching audio from {url}")
        response = self.session.get(url)
        return self._validate_audio(response, url)
    return b""

ping ¶

ping() -> float

Pings the PokeAPI and returns the latency.

Source code in pokelance/http/_sync.py

def ping(self) -> float:
    """Pings the PokeAPI and returns the latency."""
    start = time.perf_counter()
    self.request(Route())
    latency = time.perf_counter() - start
    logger.debug(f"PokeAPI sync ping latency: {latency:.4f}s")
    return latency

SyncEndpointLoader ¶

SyncEndpointLoader(client: PokeLanceSyncClient)

Composition helper for scheduling and tracking sync endpoint pre-population tasks via thread pool.

Source code in pokelance/http/_sync.py

def __init__(self, client: PokeLanceSyncClient) -> None:
    self._client = client
    self._executor: concurrent.futures.ThreadPoolExecutor | None = None
    self._futures: set[concurrent.futures.Future[None]] = set()
    self._remaining: int = 0
    self._lock = threading.Lock()
    self._ready_event = threading.Event()
    self._ready_event.set()
    self._scheduled: bool = False

is_ready `property` ¶

is_ready: bool

Whether all background loading tasks have finished.

wait_until_ready ¶

wait_until_ready() -> None

Wait until all background endpoint tasks have completed.

Source code in pokelance/http/_sync.py

def wait_until_ready(self) -> None:
    """Wait until all background endpoint tasks have completed."""
    awaiting = not self._ready_event.is_set()
    if awaiting:
        logger.info("Waiting for endpoint loading to complete...")
    self._ready_event.wait()
    if awaiting:
        logger.info("Endpoint loading complete.")

schedule_tasks ¶

schedule_tasks() -> None

Schedules the background endpoint-loading tasks on a thread pool.

Source code in pokelance/http/_sync.py

def schedule_tasks(self) -> None:
    """Schedules the background endpoint-loading tasks on a thread pool."""
    with self._lock:
        if self._scheduled:
            return
        self._scheduled = True
        self._ready_event.clear()
        if not self._client.cache_endpoints:
            self._ready_event.set()
            self._client.ext_tasks.clear()
            return
        total = len(self._client.ext_tasks)
        self._remaining = total
        logger.info(f"Scheduling {total} endpoint pre-population task(s)...")
        if self._executor is None:
            self._executor = concurrent.futures.ThreadPoolExecutor(
                max_workers=min(32, max(4, total)),
                thread_name_prefix="PokeLance-EndpointLoader",
            )
        for num, (fn, name) in enumerate(self._client.ext_tasks):
            message = f"Extension {name} endpoints ({num + 1}/{total})"
            future = self._executor.submit(self._load_ext, fn, message)
            self._futures.add(future)
            future.add_done_callback(self._futures.discard)
        self._client.ext_tasks.clear()
        if self._remaining == 0:
            self._ready_event.set()

shutdown ¶

shutdown() -> None

Cancels and shuts down the loader thread pool.

Source code in pokelance/http/_sync.py

def shutdown(self) -> None:
    """Cancels and shuts down the loader thread pool."""
    with self._lock:
        count = sum(1 for f in self._futures if not f.done())
        if count > 0:
            logger.warning(f"Cancelling {count} in-flight endpoint loading tasks...")
        for future in list(self._futures):
            future.cancel()
        if self._executor is not None:
            logger.debug("Shutting down endpoint loader thread pool...")
            self._executor.shutdown(wait=False, cancel_futures=True)
            self._executor = None

Client

http ¶

BaseHttpClient ¶

is_closing property ¶

is_connected property ¶

cache_manager property ¶

AsyncHttpClient ¶

loader property ¶

close async ¶

connect async ¶

request async ¶

load_image async ¶

load_audio async ¶

ping async ¶

AsyncEndpointLoader ¶

is_ready property ¶

wait_until_ready async ¶

schedule_tasks async ¶

cancel_tasks async ¶

SyncHttpClient ¶

loader property ¶

close ¶

connect ¶

request ¶

load_image ¶

load_audio ¶

ping ¶

SyncEndpointLoader ¶

is_ready property ¶

wait_until_ready ¶

schedule_tasks ¶

shutdown ¶

Comments

is_closing `property` ¶

is_connected `property` ¶

cache_manager `property` ¶

loader `property` ¶

close `async` ¶

connect `async` ¶

request `async` ¶

load_image `async` ¶

load_audio `async` ¶

ping `async` ¶

is_ready `property` ¶

wait_until_ready `async` ¶

schedule_tasks `async` ¶

cancel_tasks `async` ¶

loader `property` ¶

is_ready `property` ¶