ai_horde_clients

Definitions to help interact with the AI-Horde API.

BaseAIHordeClient

Base class for all AI-Horde API clients.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class BaseAIHordeClient:
    """Base class for all AI-Horde API clients."""

    _base_url: str = AI_HORDE_BASE_URL

    @property
    def base_url(self) -> str:
        """Get the base URL for the AI-Horde API."""
        return self.base_url

    @base_url.setter
    def base_url(self, value: str) -> None:
        """Set the base URL for the AI-Horde API."""
        if urllib.parse.urlparse(value).scheme not in ["http", "https"]:
            raise ValueError(f"Invalid scheme in URL: {value}")

        self.base_url = value

    def _handle_api_error(self, error_response: RequestErrorResponse, endpoint_url: str) -> None:
        """Handle an error response from the API.

        Args:
            error_response (RequestErrorResponse): The error response to handle.
            endpoint_url (str): The URL of the endpoint that was called.
        """
        logger.error("Error response received from the AI-Horde API.")
        logger.error(f"Endpoint: {endpoint_url}")
        logger.error(f"Message: {error_response.message}")

base_url property writable

base_url: str

Get the base URL for the AI-Horde API.

AIHordeAPIManualClient

Bases: GenericHordeAPIManualClient, BaseAIHordeClient

An API client specifically configured for the AI-Horde API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPIManualClient(GenericHordeAPIManualClient, BaseAIHordeClient):
    """An API client specifically configured for the AI-Horde API."""

    def __init__(self) -> None:
        """Create a new instance of the AIHordeAPIManualClient."""
        super().__init__(
            path_fields=AIHordePathData,
            query_fields=AIHordeQueryData,
        )

    def get_generate_check(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateCheckResponse:
        """Check if a pending image request has finished generating from the AI-Horde API.

        Not to be confused with `get_generate_status` which returns the images too.

        Args:
            gen_id (GenerationID | str): The ID of the request to check.

        Returns:
            ImageGenerateCheckResponse: The response from the API.
        """
        api_request = ImageGenerateCheckRequest(id=gen_id)

        api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

    def get_generate_status(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateStatusResponse:
        """Get the status and any generated images for a pending image request from the AI-Horde API.

        *Do not use this method more often than is necessary.* The AI-Horde API will rate limit you if you do.
        Use `get_generate_check` instead to check the status of a pending image request.

        Args:
            gen_id (GenerationID): The ID of the request to check.

        Returns:
            tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.
        """
        api_request = ImageGenerateStatusRequest(id=gen_id)

        api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

    def delete_pending_image(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateStatusResponse:
        """Delete a pending image request from the AI-Horde API.

        Args:
            gen_id (GenerationID): The ID of the request to delete.

        Returns:
            ImageGenerateStatusResponse: The response from the API.
        """
        api_request = DeleteImageGenerateRequest(id=gen_id)

        api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

base_url property writable

base_url: str

Get the base URL for the AI-Horde API.

retry_config instance-attribute

retry_config: RetryConfiguration = retry_config

__init__

__init__() -> None

Create a new instance of the AIHordeAPIManualClient.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def __init__(self) -> None:
    """Create a new instance of the AIHordeAPIManualClient."""
    super().__init__(
        path_fields=AIHordePathData,
        query_fields=AIHordeQueryData,
    )

get_generate_check

get_generate_check(
    gen_id: GenerationID,
) -> ImageGenerateCheckResponse

Check if a pending image request has finished generating from the AI-Horde API.

Not to be confused with get_generate_status which returns the images too.

Parameters:

• gen_id (GenerationID | str) –

  The ID of the request to check.

Returns:

• ImageGenerateCheckResponse ( ImageGenerateCheckResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def get_generate_check(
    self,
    gen_id: GenerationID,
) -> ImageGenerateCheckResponse:
    """Check if a pending image request has finished generating from the AI-Horde API.

    Not to be confused with `get_generate_status` which returns the images too.

    Args:
        gen_id (GenerationID | str): The ID of the request to check.

    Returns:
        ImageGenerateCheckResponse: The response from the API.
    """
    api_request = ImageGenerateCheckRequest(id=gen_id)

    api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

get_generate_status

get_generate_status(
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse

Get the status and any generated images for a pending image request from the AI-Horde API.

Do not use this method more often than is necessary. The AI-Horde API will rate limit you if you do. Use get_generate_check instead to check the status of a pending image request.

Parameters:

• gen_id (GenerationID) –

  The ID of the request to check.

Returns:

• ImageGenerateStatusResponse –

  tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def get_generate_status(
    self,
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse:
    """Get the status and any generated images for a pending image request from the AI-Horde API.

    *Do not use this method more often than is necessary.* The AI-Horde API will rate limit you if you do.
    Use `get_generate_check` instead to check the status of a pending image request.

    Args:
        gen_id (GenerationID): The ID of the request to check.

    Returns:
        tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.
    """
    api_request = ImageGenerateStatusRequest(id=gen_id)

    api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

delete_pending_image

delete_pending_image(
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse

Delete a pending image request from the AI-Horde API.

Parameters:

• gen_id (GenerationID) –

  The ID of the request to delete.

Returns:

• ImageGenerateStatusResponse ( ImageGenerateStatusResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def delete_pending_image(
    self,
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse:
    """Delete a pending image request from the AI-Horde API.

    Args:
        gen_id (GenerationID): The ID of the request to delete.

    Returns:
        ImageGenerateStatusResponse: The response from the API.
    """
    api_request = DeleteImageGenerateRequest(id=gen_id)

    api_response = self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

should_retry

should_retry(
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool

Determine if a request should be retried based on the status code and retry configuration.

Parameters:

• status_code (int) –

  The HTTP status code returned by the request.

• current_error_count (int) –

  The current number of errors encountered.

• retry_after (float) –

  The time to wait before retrying the request.

Returns:

• bool ( bool ) –

  True if the request should be retried, False otherwise.

Source code in horde_sdk/generic_api/generic_clients.py

def should_retry(
    self,
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool:
    """Determine if a request should be retried based on the status code and retry configuration.

    Args:
        status_code (int): The HTTP status code returned by the request.
        current_error_count (int): The current number of errors encountered.
        retry_after (float): The time to wait before retrying the request.

    Returns:
        bool: True if the request should be retried, False otherwise.
    """
    if not self._retry_by_default:
        return False

    if current_error_count >= self.retry_config.max_retries:
        return False

    if status_code not in self.retry_config.retry_status_codes:
        return False

    jitter = (self.retry_config.jitter_factor * retry_after) if self.retry_config.jitter_factor else 0

    retry_delay = (
        min(
            self.retry_config.initial_delay_seconds * (self.retry_config.backoff_factor**current_error_count),
            self.retry_config.max_delay_seconds,
        )
        + jitter
    )

    time.sleep(retry_delay)
    return True

submit_request

submit_request(
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse

Submit a request to the API and return the response.

If you are wondering why expected_response_type is a parameter, it is because the API may return different responses depending on the payload or other factors. It is up to you to determine which response type you expect, and pass it in here.

Parameters:

• api_request (HordeRequest) –

  The request to submit.

• expected_response_type (type[HordeResponse]) –

  The expected response type.

Returns:

• HordeResponseTypeVar | RequestErrorResponse –

  HordeResponseTypeVar | RequestErrorResponse: The response from the API.

Source code in horde_sdk/generic_api/generic_clients.py

def submit_request(
    self,
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse:
    """Submit a request to the API and return the response.

    If you are wondering why `expected_response_type` is a parameter, it is because the API may return different
    responses depending on the payload or other factors. It is up to you to determine which response type you
    expect, and pass it in here.

    Args:
        api_request (HordeRequest): The request to submit.
        expected_response_type (type[HordeResponse]): The expected response type.

    Returns:
        HordeResponseTypeVar | RequestErrorResponse: The response from the API.
    """
    http_method_name = api_request.get_http_method()

    if expected_response_type not in api_request.get_success_status_response_pairs().values():
        logger.warning(
            "The expected response type is not in the list of success status response pairs! This may result in "
            "unexpected behavior.",
        )
        logger.warning(f"Passed expected_response_type: {expected_response_type}")
        logger.warning(f"Allowable pairs defined in the SDK : {api_request.get_success_status_response_pairs()}")

    with logfire.span(
        self._msg_format_submit_request.format(
            sync_async="sync",
            http_method_name=http_method_name,
            api_request_type=type(api_request).__name__,
            expected_response_type=expected_response_type.__name__,
        ),
        sync_async="sync",
        http_method_name=http_method_name,
        api_request_type=type(api_request).__name__,
        expected_response_type=expected_response_type.__name__,
    ):
        parsed_request = self._validate_and_prepare_request(api_request)

        raw_response: requests.Response | None = None

        if http_method_name == HTTPMethod.GET:
            if parsed_request.request_body is not None:
                raise RuntimeError(
                    "GET requests cannot have a body! This may mean you forgot to override `get_header_fields()` "
                    "or perhaps you may need to define a `metadata.py` module or entry in it for your API.",
                )
            raw_response = requests.get(
                parsed_request.endpoint_no_query,
                headers=parsed_request.request_headers,
                params=parsed_request.request_queries,
                allow_redirects=True,
            )
        else:
            raw_response = requests.request(
                method=http_method_name,
                url=parsed_request.endpoint_no_query,
                headers=parsed_request.request_headers,
                params=parsed_request.request_queries,
                json=parsed_request.request_body,
                allow_redirects=True,
            )

        return self._after_request_handling(
            raw_response_json=raw_response.json(),
            returned_status_code=raw_response.status_code,
            expected_response_type=expected_response_type,
        )

AIHordeAPIAsyncManualClient

Bases: GenericAsyncHordeAPIManualClient, BaseAIHordeClient

An asyncio based API client specifically configured for the AI-Horde API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPIAsyncManualClient(GenericAsyncHordeAPIManualClient, BaseAIHordeClient):
    """An asyncio based API client specifically configured for the AI-Horde API."""

    def __init__(
        self,
        aiohttp_session: aiohttp.ClientSession,
        *,
        ssl_context: SSLContext = _default_sslcontext,
    ) -> None:
        """Create a new instance of the RatingsAPIClient."""
        super().__init__(
            aiohttp_session=aiohttp_session,
            path_fields=AIHordePathData,
            query_fields=AIHordeQueryData,
            ssl_context=ssl_context,
        )

    async def get_generate_check(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateCheckResponse:
        """Asynchronously check if a pending image request has finished generating and return the status of it.

        Not to be confused with `get_generate_status` which returns the images too.

        Args:
            gen_id (GenerationID | str): The ID of the request to check.

        Returns:
            ImageGenerateCheckResponse: The response from the API.
        """
        api_request = ImageGenerateCheckRequest(id=gen_id)

        api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

    async def get_generate_status(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateStatusResponse:
        """Asynchronously get the status and any generated images for a pending image request from the AI-Horde API.

        *Do not use this method more often than is necessary.* The AI-Horde API will rate limit you if you do.
        Use `get_generate_check` instead to check the status of a pending image request.

        Args:
            gen_id (GenerationID): The ID of the request to check.

        Returns:
            ImageGenerateStatusResponse: The response from the API.
        """
        api_request = ImageGenerateStatusRequest(id=gen_id)

        api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

    async def delete_pending_image(
        self,
        gen_id: GenerationID,
    ) -> ImageGenerateStatusResponse:
        """Asynchronously delete a pending image request from the AI-Horde API.

        Args:
            gen_id (GenerationID | str): The ID of the request to delete.

        Returns:
            ImageGenerateStatusResponse: The response from the API.
        """
        api_request = DeleteImageGenerateRequest(id=gen_id)

        api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
        if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
            self._handle_api_error(api_response, api_request.get_api_endpoint_url())
            raise AIHordeRequestError(api_response)

        return api_response

base_url property writable

base_url: str

Get the base URL for the AI-Horde API.

retry_config instance-attribute

retry_config: RetryConfiguration = retry_config

__init__

__init__(
    aiohttp_session: ClientSession,
    *,
    ssl_context: SSLContext = _default_sslcontext
) -> None

Create a new instance of the RatingsAPIClient.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def __init__(
    self,
    aiohttp_session: aiohttp.ClientSession,
    *,
    ssl_context: SSLContext = _default_sslcontext,
) -> None:
    """Create a new instance of the RatingsAPIClient."""
    super().__init__(
        aiohttp_session=aiohttp_session,
        path_fields=AIHordePathData,
        query_fields=AIHordeQueryData,
        ssl_context=ssl_context,
    )

get_generate_check async

get_generate_check(
    gen_id: GenerationID,
) -> ImageGenerateCheckResponse

Asynchronously check if a pending image request has finished generating and return the status of it.

Not to be confused with get_generate_status which returns the images too.

Parameters:

• gen_id (GenerationID | str) –

  The ID of the request to check.

Returns:

• ImageGenerateCheckResponse ( ImageGenerateCheckResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def get_generate_check(
    self,
    gen_id: GenerationID,
) -> ImageGenerateCheckResponse:
    """Asynchronously check if a pending image request has finished generating and return the status of it.

    Not to be confused with `get_generate_status` which returns the images too.

    Args:
        gen_id (GenerationID | str): The ID of the request to check.

    Returns:
        ImageGenerateCheckResponse: The response from the API.
    """
    api_request = ImageGenerateCheckRequest(id=gen_id)

    api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

get_generate_status async

get_generate_status(
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse

Asynchronously get the status and any generated images for a pending image request from the AI-Horde API.

Do not use this method more often than is necessary. The AI-Horde API will rate limit you if you do. Use get_generate_check instead to check the status of a pending image request.

Parameters:

• gen_id (GenerationID) –

  The ID of the request to check.

Returns:

• ImageGenerateStatusResponse ( ImageGenerateStatusResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def get_generate_status(
    self,
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse:
    """Asynchronously get the status and any generated images for a pending image request from the AI-Horde API.

    *Do not use this method more often than is necessary.* The AI-Horde API will rate limit you if you do.
    Use `get_generate_check` instead to check the status of a pending image request.

    Args:
        gen_id (GenerationID): The ID of the request to check.

    Returns:
        ImageGenerateStatusResponse: The response from the API.
    """
    api_request = ImageGenerateStatusRequest(id=gen_id)

    api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

delete_pending_image async

delete_pending_image(
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse

Asynchronously delete a pending image request from the AI-Horde API.

Parameters:

• gen_id (GenerationID | str) –

  The ID of the request to delete.

Returns:

• ImageGenerateStatusResponse ( ImageGenerateStatusResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def delete_pending_image(
    self,
    gen_id: GenerationID,
) -> ImageGenerateStatusResponse:
    """Asynchronously delete a pending image request from the AI-Horde API.

    Args:
        gen_id (GenerationID | str): The ID of the request to delete.

    Returns:
        ImageGenerateStatusResponse: The response from the API.
    """
    api_request = DeleteImageGenerateRequest(id=gen_id)

    api_response = await self.submit_request(api_request, api_request.get_default_success_response_type())
    if isinstance(api_response, RequestErrorResponse):  # pragma: no cover
        self._handle_api_error(api_response, api_request.get_api_endpoint_url())
        raise AIHordeRequestError(api_response)

    return api_response

should_retry

should_retry(
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool

Determine if a request should be retried based on the status code and retry configuration.

Parameters:

• status_code (int) –

  The HTTP status code returned by the request.

• current_error_count (int) –

  The current number of errors encountered.

• retry_after (float) –

  The time to wait before retrying the request.

Returns:

• bool ( bool ) –

  True if the request should be retried, False otherwise.

Source code in horde_sdk/generic_api/generic_clients.py

def should_retry(
    self,
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool:
    """Determine if a request should be retried based on the status code and retry configuration.

    Args:
        status_code (int): The HTTP status code returned by the request.
        current_error_count (int): The current number of errors encountered.
        retry_after (float): The time to wait before retrying the request.

    Returns:
        bool: True if the request should be retried, False otherwise.
    """
    if not self._retry_by_default:
        return False

    if current_error_count >= self.retry_config.max_retries:
        return False

    if status_code not in self.retry_config.retry_status_codes:
        return False

    jitter = (self.retry_config.jitter_factor * retry_after) if self.retry_config.jitter_factor else 0

    retry_delay = (
        min(
            self.retry_config.initial_delay_seconds * (self.retry_config.backoff_factor**current_error_count),
            self.retry_config.max_delay_seconds,
        )
        + jitter
    )

    time.sleep(retry_delay)
    return True

submit_request async

submit_request(
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse

Submit a request to the API asynchronously and return the response.

If you are wondering why expected_response_type is a parameter, it is because the API may return different responses depending on the payload or other factors. It is up to you to determine which response type you expect, and pass it in here.

Parameters:

• api_request (HordeRequest) –

  The request to submit.

• expected_response_type (type[HordeResponse]) –

  The expected response type.

Returns:

• HordeResponseTypeVar | RequestErrorResponse –

  HordeResponse | RequestErrorResponse: The response from the API.

Raises:

• ClientResponseError –

  If a network problem occurred.

Source code in horde_sdk/generic_api/generic_clients.py

async def submit_request(
    self,
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse:
    """Submit a request to the API asynchronously and return the response.

    If you are wondering why `expected_response_type` is a parameter, it is because the API may return different
    responses depending on the payload or other factors. It is up to you to determine which response type you
    expect, and pass it in here.

    Args:
        api_request (HordeRequest): The request to submit.
        expected_response_type (type[HordeResponse]): The expected response type.

    Returns:
        HordeResponse | RequestErrorResponse: The response from the API.

    Raises:
        ClientResponseError: If a network problem occurred.
    """
    http_method_name = api_request.get_http_method()

    parsed_request = self._validate_and_prepare_request(api_request)

    raw_response_json: dict[str, Any] = {}
    response_status: int = 599

    if not self._aiohttp_session:
        raise RuntimeError("No aiohttp session was provided but an async method was called!")

    with logfire.span(
        self._msg_format_submit_request.format(
            sync_async="async",
            http_method_name=http_method_name,
            api_request_type=type(api_request).__name__,
            expected_response_type=expected_response_type.__name__,
        ),
        sync_async="async",
        http_method_name=http_method_name,
        api_request_type=type(api_request).__name__,
        expected_response_type=expected_response_type.__name__,
    ):
        async with (
            self._aiohttp_session.request(
                http_method_name.value,
                parsed_request.endpoint_no_query,
                headers=parsed_request.request_headers,
                params=parsed_request.request_queries,
                json=parsed_request.request_body,
                allow_redirects=True,
                ssl=self._ssl_context,
            ) as response,
        ):
            raw_response_json = await response.json()
            response_status = response.status

        return self._after_request_handling(
            raw_response_json=raw_response_json,
            returned_status_code=response_status,
            expected_response_type=expected_response_type,
        )

AIHordeAPIClientSession

Bases: GenericHordeAPISession

Context handler representing an API session specifically configured for the AI-Horde API.

If you make a request which requires follow up (such as a request to generate an image), this will delete the generation in progress when the context manager exits. If you want to control this yourself, use AIHordeAPIManualClient instead.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPIClientSession(GenericHordeAPISession):
    """Context handler representing an API session specifically configured for the AI-Horde API.

    If you make a request which requires follow up (such as a request to generate an image), this will delete the
    generation in progress when the context manager exits. If you want to control this yourself, use
    `AIHordeAPIManualClient` instead.
    """

    def __init__(self) -> None:
        """Create a new instance of the RatingsAPIClient."""
        super().__init__(
            path_fields=AIHordePathData,
            query_fields=AIHordeQueryData,
        )

retry_config instance-attribute

retry_config: RetryConfiguration = retry_config

__init__

__init__() -> None

Create a new instance of the RatingsAPIClient.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def __init__(self) -> None:
    """Create a new instance of the RatingsAPIClient."""
    super().__init__(
        path_fields=AIHordePathData,
        query_fields=AIHordeQueryData,
    )

should_retry

should_retry(
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool

Determine if a request should be retried based on the status code and retry configuration.

Parameters:

• status_code (int) –

  The HTTP status code returned by the request.

• current_error_count (int) –

  The current number of errors encountered.

• retry_after (float) –

  The time to wait before retrying the request.

Returns:

• bool ( bool ) –

  True if the request should be retried, False otherwise.

Source code in horde_sdk/generic_api/generic_clients.py

def should_retry(
    self,
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool:
    """Determine if a request should be retried based on the status code and retry configuration.

    Args:
        status_code (int): The HTTP status code returned by the request.
        current_error_count (int): The current number of errors encountered.
        retry_after (float): The time to wait before retrying the request.

    Returns:
        bool: True if the request should be retried, False otherwise.
    """
    if not self._retry_by_default:
        return False

    if current_error_count >= self.retry_config.max_retries:
        return False

    if status_code not in self.retry_config.retry_status_codes:
        return False

    jitter = (self.retry_config.jitter_factor * retry_after) if self.retry_config.jitter_factor else 0

    retry_delay = (
        min(
            self.retry_config.initial_delay_seconds * (self.retry_config.backoff_factor**current_error_count),
            self.retry_config.max_delay_seconds,
        )
        + jitter
    )

    time.sleep(retry_delay)
    return True

submit_request

submit_request(
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse

Source code in horde_sdk/generic_api/generic_clients.py

@override
def submit_request(
    self,
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse:
    response = super().submit_request(api_request, expected_response_type)

    if isinstance(response, ResponseRequiringFollowUpMixin):
        self._pending_follow_ups.append(
            (api_request, response, response.get_follow_up_failure_cleanup_request()),
        )
    else:  # TODO: This whole else is duplicated in the asyncio version of this class. Refactor it out.
        # Check if this request is a cleanup or follow up request for a prior request
        # Loop through each item in self._pending_follow_ups list
        for index, (prior_request, prior_response, cleanup_request) in enumerate(self._pending_follow_ups):
            if cleanup_request is not None and api_request in cleanup_request:
                if not isinstance(response, RequestErrorResponse):
                    self._pending_follow_ups.pop(index)
                else:
                    logger.error(
                        "This api request would have followed up on an operation which requires it, but it "
                        "failed!",
                    )
                    logger.error(f"Request: {api_request.log_safe_model_dump()}")
                    logger.error(f"Response: {response}")
                break

            if not isinstance(prior_response, ResponseRequiringFollowUpMixin):
                continue

            # If the response isn't a final follow-up, we don't need to do anything else.
            if isinstance(response, ResponseWithProgressMixin):
                if not response.is_final_follow_up():
                    continue
                if not prior_request.get_requires_follow_up():
                    continue

                # See if the current api_request is a follow-up to the prior_request
                if not prior_response.does_target_request_follow_up(api_request):
                    continue

                # Check if the current response indicates that the job is complete
                if response.is_job_complete(prior_request.get_number_of_results_expected()):
                    # Remove the current item from the _pending_follow_ups list
                    # This is for the benefit of the __exit__ method (context management)
                    self._pending_follow_ups.pop(index)
                    break
            else:
                if not prior_response.does_target_request_follow_up(api_request):
                    continue

                self._pending_follow_ups.pop(index)
                break

    return response

__enter__

__enter__() -> GenericHordeAPISession

Enter the context manager.

Source code in horde_sdk/generic_api/generic_clients.py

def __enter__(self) -> GenericHordeAPISession:
    """Enter the context manager."""
    return self

__exit__

__exit__(
    exc_type: type[BaseException],
    exc_val: Exception,
    exc_tb: object,
) -> bool

Exit the context manager.

Source code in horde_sdk/generic_api/generic_clients.py

def __exit__(self, exc_type: type[BaseException], exc_val: Exception, exc_tb: object) -> bool:
    """Exit the context manager."""
    # If there was no exception, return True.
    if exc_type is None:
        return True

    # Log the error
    logger.error(f"Error: {exc_val}, Type: {exc_type}")

    # Show the traceback if there is one
    if exc_tb and hasattr(exc_tb, "print_exc"):
        exc_tb.print_exc()

    # If there are no pending follow-up requests, return True if the exception was a CancelledError.
    if not self._pending_follow_ups:
        return exc_type is asyncio.exceptions.CancelledError

    # Handle each pending follow-up request.
    all_handled = True
    for request_to_follow_up, response_to_follow_up, cleanup_request in self._pending_follow_ups:
        handled = self._handle_exit(request_to_follow_up, response_to_follow_up, cleanup_request)
        all_handled = all_handled and handled

    # Check if the exception was a CancelledError.
    is_cancelled = exc_type is asyncio.exceptions.CancelledError

    # If we cancelled the task and everything cleaned up ok, we don't want to raise an exception.
    return all_handled and is_cancelled  # Returns True if everything was handled and we cancelled the task.

AIHordeAPIAsyncClientSession

Bases: GenericAsyncHordeAPISession

Context handler representing an API session specifically configured for the AI-Horde API.

If you make a request which requires follow up (such as a request to generate an image), this will delete the generation in progress when the context manager exits. If you want to control this yourself, use AIHordeAPIManualClient instead.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPIAsyncClientSession(GenericAsyncHordeAPISession):
    """Context handler representing an API session specifically configured for the AI-Horde API.

    If you make a request which requires follow up (such as a request to generate an image), this will delete the
    generation in progress when the context manager exits. If you want to control this yourself, use
    `AIHordeAPIManualClient` instead.
    """

    def __init__(
        self,
        aiohttp_session: aiohttp.ClientSession,
        ssl_context: SSLContext = _default_sslcontext,
        apikey: str | None = None,
    ) -> None:
        """Create a new instance of the RatingsAPIClient."""
        super().__init__(
            aiohttp_session=aiohttp_session,
            apikey=apikey,
            path_fields=AIHordePathData,
            query_fields=AIHordeQueryData,
            ssl_context=ssl_context,
        )

retry_config instance-attribute

retry_config: RetryConfiguration = retry_config

__init__

__init__(
    aiohttp_session: ClientSession,
    ssl_context: SSLContext = _default_sslcontext,
    apikey: str | None = None,
) -> None

Create a new instance of the RatingsAPIClient.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def __init__(
    self,
    aiohttp_session: aiohttp.ClientSession,
    ssl_context: SSLContext = _default_sslcontext,
    apikey: str | None = None,
) -> None:
    """Create a new instance of the RatingsAPIClient."""
    super().__init__(
        aiohttp_session=aiohttp_session,
        apikey=apikey,
        path_fields=AIHordePathData,
        query_fields=AIHordeQueryData,
        ssl_context=ssl_context,
    )

should_retry

should_retry(
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool

Determine if a request should be retried based on the status code and retry configuration.

Parameters:

• status_code (int) –

  The HTTP status code returned by the request.

• current_error_count (int) –

  The current number of errors encountered.

• retry_after (float) –

  The time to wait before retrying the request.

Returns:

• bool ( bool ) –

  True if the request should be retried, False otherwise.

Source code in horde_sdk/generic_api/generic_clients.py

def should_retry(
    self,
    status_code: int,
    current_error_count: int,
    retry_after: float,
) -> bool:
    """Determine if a request should be retried based on the status code and retry configuration.

    Args:
        status_code (int): The HTTP status code returned by the request.
        current_error_count (int): The current number of errors encountered.
        retry_after (float): The time to wait before retrying the request.

    Returns:
        bool: True if the request should be retried, False otherwise.
    """
    if not self._retry_by_default:
        return False

    if current_error_count >= self.retry_config.max_retries:
        return False

    if status_code not in self.retry_config.retry_status_codes:
        return False

    jitter = (self.retry_config.jitter_factor * retry_after) if self.retry_config.jitter_factor else 0

    retry_delay = (
        min(
            self.retry_config.initial_delay_seconds * (self.retry_config.backoff_factor**current_error_count),
            self.retry_config.max_delay_seconds,
        )
        + jitter
    )

    time.sleep(retry_delay)
    return True

submit_request async

submit_request(
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse

Source code in horde_sdk/generic_api/generic_clients.py

@override
async def submit_request(
    self,
    api_request: HordeRequest,
    expected_response_type: type[HordeResponseTypeVar],
) -> HordeResponseTypeVar | RequestErrorResponse:
    # Add the request to the list of awaiting requests.

    async with self._awaiting_requests_lock:
        self._awaiting_requests.append(api_request)

    # Submit the request to the API and get the response.
    response = await super().submit_request(api_request, expected_response_type)

    # Remove the request from the list of awaiting requests.
    async with self._awaiting_requests_lock, self._pending_follow_ups_lock:
        self._awaiting_requests.remove(api_request)

        # Check if the response requires a follow-up request.
        if isinstance(response, ResponseRequiringFollowUpMixin):
            # Add the follow-up request to the list of pending follow-ups.
            if not response.ignore_failure():
                self._pending_follow_ups.append(
                    (api_request, response, response.get_follow_up_failure_cleanup_request()),
                )

        else:
            # Check if this request is a cleanup or follow up request for a prior request

            # Loop through each item in self._pending_follow_ups list
            for index, (prior_request, prior_response, cleanup_request) in enumerate(self._pending_follow_ups):
                if cleanup_request is not None and api_request in cleanup_request:
                    if not isinstance(response, RequestErrorResponse):
                        self._pending_follow_ups.pop(index)
                        break

                    logger.error(
                        "This api request would have followed up on an operation which requires it, but it "
                        "failed!",
                    )
                    logger.error(f"Request: {api_request.log_safe_model_dump()}")
                    logger.error(f"Response: {response.log_safe_model_dump()}")
                    break

                if not isinstance(prior_response, ResponseRequiringFollowUpMixin):
                    continue

                # If the response isn't a final follow-up, we don't need to do anything else.
                if isinstance(response, ResponseWithProgressMixin):
                    if not response.is_final_follow_up():
                        continue
                    if not prior_request.get_requires_follow_up():
                        continue

                    # See if the current api_request is a follow-up to the prior_request
                    if not prior_response.does_target_request_follow_up(api_request):
                        continue

                    # Check if the current response indicates that the job is complete
                    if response.is_job_complete(prior_request.get_number_of_results_expected()):
                        # Remove the current item from the _pending_follow_ups list
                        # This is for the benefit of the __exit__ method (context management)
                        self._pending_follow_ups.pop(index)
                        break
                else:
                    if not prior_response.does_target_request_follow_up(api_request):
                        continue

                    self._pending_follow_ups.pop(index)
                    break

    # Return the response from the API.
    return response

__aenter__ async

__aenter__() -> GenericAsyncHordeAPISession

Enter the context manager asynchronously.

Source code in horde_sdk/generic_api/generic_clients.py

async def __aenter__(self) -> GenericAsyncHordeAPISession:
    """Enter the context manager asynchronously."""
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException],
    exc_val: Exception,
    exc_tb: object,
) -> bool

Exit the context manager asynchronously.

Source code in horde_sdk/generic_api/generic_clients.py

async def __aexit__(self, exc_type: type[BaseException], exc_val: Exception, exc_tb: object) -> bool:
    """Exit the context manager asynchronously."""
    # If there are any requests that haven't been returned yet, log a warning.
    if self._awaiting_requests:
        logger.warning(
            "This session was used to submit asynchronous requests, but the context manager was exited "
            "before all requests were returned! This may result in requests not being handled properly.",
        )
        # Log each unhandled request.
        for request in self._awaiting_requests:
            logger.warning(f"Request Unhandled: {request.log_safe_model_dump()}")

    # Log the error if there was one.
    if exc_type:
        logger.error(f"Error: {exc_val}, Type: {exc_type}")

    # Show the traceback if there is one
    if exc_tb and hasattr(exc_tb, "print_exc"):
        exc_tb.print_exc()

    # If there are no pending follow-up requests, return True if the exception was a CancelledError.
    if not self._pending_follow_ups:
        return exc_type is asyncio.exceptions.CancelledError

    try:
        # Handle each pending follow-up request asynchronously.
        await asyncio.gather(
            *[
                self._handle_exit_async(request_to_follow_up, response_to_follow_up, cleanup_request)
                for request_to_follow_up, response_to_follow_up, cleanup_request in self._pending_follow_ups
            ],
        )

        # Return True if everything was handled and the task was cancelled deliberately,
        # False otherwise (which will reraise the exception)
        return exc_type is asyncio.exceptions.CancelledError
    except Exception as e:
        # If an exception occurred while handling the follow-up requests, log an error and return False.
        logger.exception(e)
        return False

BaseAIHordeSimpleClient

Bases: ABC

The base class for the most straightforward clients which interact with the AI-Horde API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class BaseAIHordeSimpleClient(ABC):
    """The base class for the most straightforward clients which interact with the AI-Horde API."""

    _msg_format_sleep = "sleeping for {seconds} seconds"

    reasonable_minimum_timeout = 20

    def validate_timeout(
        self,
        timeout: int,
        log_message: bool = False,
    ) -> int:
        """Check if a timeout is reasonable.

        Args:
            timeout (int): The timeout to check.
            log_message (bool, optional): Whether to log a message if the timeout is too short. Defaults to False.

        Returns:
            bool: True if the timeout is reasonable, False otherwise.
        """
        if timeout <= 0:  # pragma: no cover
            logger.debug(f"No timeout set. Using default timeout of {GENERATION_MAX_LIFE} seconds.")
            return GENERATION_MAX_LIFE

        if timeout > GENERATION_MAX_LIFE:  # pragma: no cover
            logger.warning(
                f"Timeout ({timeout}) is greater than the maximum possible of {GENERATION_MAX_LIFE} seconds."
                f"Using {GENERATION_MAX_LIFE} for the timeout.",
            )
            return GENERATION_MAX_LIFE

        if timeout < self.reasonable_minimum_timeout and log_message:  # pragma: no cover
            logger.warning(
                f"test_simple_client_async_image_generate_multiple than {self.reasonable_minimum_timeout} seconds, "
                "this is probably too short.",
            )

        return timeout

    @abstractmethod
    def download_image_from_generation(
        self,
        generation: ImageGeneration,
    ) -> PIL.Image.Image | Coroutine[None, None, tuple[PIL.Image.Image, GenerationID]]:
        """Convert from base64 or download an image from a response."""

    @abstractmethod
    def download_image_from_url(
        self,
        url: str,
    ) -> PIL.Image.Image | Coroutine[None, None, PIL.Image.Image]:
        """Download an image from a URL."""

    def _handle_initial_response(
        self,
        initial_response: HordeResponse | RequestErrorResponse,
    ) -> tuple[HordeRequest, GenerationID, list[dict[str, object]]]:
        # Check for error responses
        if isinstance(initial_response, RequestErrorResponse):  # pragma: no cover
            if "Image validation failed" in initial_response.message:  # TODO: No magic strings!
                raise AIHordeImageValidationError(initial_response)
            raise AIHordeRequestError(initial_response)

        if not isinstance(initial_response, ResponseRequiringFollowUpMixin):  # pragma: no cover
            raise RuntimeError("Response did not need follow up")

        # Get the follow up data from the response
        check_request_type = initial_response.get_follow_up_default_request_type()
        follow_up_data = initial_response.get_follow_up_returned_params()
        num_follow_up_data = len(follow_up_data)

        # If there is not exactly one follow up request, something has gone wrong (or the API has changed?)
        if num_follow_up_data != 1:  # FIXME?  # pragma: no cover
            raise RuntimeError(
                f"Expected exactly one check request should have been found, found {num_follow_up_data}",
            )

        # Create the check request from the follow up data
        check_request = check_request_type.model_validate(follow_up_data[0])

        if not isinstance(check_request, JobRequestMixin):  # pragma: no cover
            logger.error(f"Check request type is not a JobRequestMixin: {check_request.log_safe_model_dump()}")
            raise RuntimeError(
                f"Check request type is not a JobRequestMixin: {check_request.log_safe_model_dump()}",
            )

        gen_id: GenerationID = check_request.id_

        logger.log(PROGRESS_LOGGER_LABEL, f"Response received: {initial_response}")
        if isinstance(initial_response, ContainsMessageResponseMixin) and initial_response.message:
            if "warning" in initial_response.message.lower():
                logger.warning(f"{gen_id}: {initial_response.message}")
            else:
                logger.info(f"{gen_id}: {initial_response.message}")

        return check_request, gen_id, follow_up_data

    def _handle_progress_response(
        self,
        check_request: HordeRequest,
        check_response: HordeResponse | RequestErrorResponse,
        gen_id: GenerationID,
        *,
        check_count: int,
        number_of_responses: int,
        start_time: float,
        timeout: int,
        check_callback: Callable[[HordeResponse], None] | None = None,
        check_callback_type: type[ResponseWithProgressMixin | ResponseGenerationProgressCombinedMixin] | None = None,
    ) -> PROGRESS_STATE:
        """Handle a response from the API when checking the progress of a request.

        Typically, this is a response from a `check` or `status` request.
        """
        # Check for error responses
        if isinstance(check_response, RequestErrorResponse):
            raise AIHordeRequestError(check_response)

        # Check if the response has progress
        if not isinstance(check_response, ResponseWithProgressMixin):
            raise RuntimeError(f"Response did not have progress: {check_response}")

        # If there is a callback, call it with the response
        if check_callback is not None:
            if check_callback_type and not isinstance(check_response, check_callback_type):
                raise RuntimeError(f"Callback response type mismatch: {check_response}")
            if check_callback_type is None:
                logger.warning("Callback type not specified, skipping type check")
                logger.debug(f"Type of sent response: {type(check_response)}")
            check_callback(check_response)

        # Log a message indicating that the request has been checked
        log_message = f"Checked request: {gen_id}, is_possible: {check_response.is_job_possible()}"

        # Log the request if it's the first check or every 5th check
        if check_count == 1 or check_count % 5 == 0:
            logger.log(PROGRESS_LOGGER_LABEL, log_message)
            logger.log(PROGRESS_LOGGER_LABEL, f"{gen_id}: {check_response.log_safe_model_dump()}")
            if not check_response.is_job_possible():
                logger.warning(f"Job not possible: {gen_id}")
        # Otherwise, just log the message at the debug level
        else:
            logger.debug(log_message)

        # If the number of finished images is equal to the number of images requested, we're done
        if check_response.is_job_complete(number_of_responses):
            logger.log(PROGRESS_LOGGER_LABEL, f"Job finished and available on the server: {gen_id}")
            return PROGRESS_STATE.finished

        # If we've timed out, stop waiting, log a warning, and break out of the loop
        if timeout and timeout > 0 and time.time() - start_time > timeout:
            logger.warning(
                f"Timeout reached, cancelling generations still outstanding: {gen_id}: "
                f"{check_response.log_safe_model_dump()}:",
            )
            return PROGRESS_STATE.timed_out

        # If the job is not complete and the timeout has not been reached, continue waiting
        return PROGRESS_STATE.waiting

reasonable_minimum_timeout class-attribute instance-attribute

reasonable_minimum_timeout = 20

validate_timeout

validate_timeout(
    timeout: int, log_message: bool = False
) -> int

Check if a timeout is reasonable.

Parameters:

• timeout (int) –

  The timeout to check.

• log_message (bool, default: False ) –

  Whether to log a message if the timeout is too short. Defaults to False.

Returns:

• bool ( int ) –

  True if the timeout is reasonable, False otherwise.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def validate_timeout(
    self,
    timeout: int,
    log_message: bool = False,
) -> int:
    """Check if a timeout is reasonable.

    Args:
        timeout (int): The timeout to check.
        log_message (bool, optional): Whether to log a message if the timeout is too short. Defaults to False.

    Returns:
        bool: True if the timeout is reasonable, False otherwise.
    """
    if timeout <= 0:  # pragma: no cover
        logger.debug(f"No timeout set. Using default timeout of {GENERATION_MAX_LIFE} seconds.")
        return GENERATION_MAX_LIFE

    if timeout > GENERATION_MAX_LIFE:  # pragma: no cover
        logger.warning(
            f"Timeout ({timeout}) is greater than the maximum possible of {GENERATION_MAX_LIFE} seconds."
            f"Using {GENERATION_MAX_LIFE} for the timeout.",
        )
        return GENERATION_MAX_LIFE

    if timeout < self.reasonable_minimum_timeout and log_message:  # pragma: no cover
        logger.warning(
            f"test_simple_client_async_image_generate_multiple than {self.reasonable_minimum_timeout} seconds, "
            "this is probably too short.",
        )

    return timeout

download_image_from_generation abstractmethod

download_image_from_generation(
    generation: ImageGeneration,
) -> (
    PIL.Image.Image
    | Coroutine[
        None, None, tuple[PIL.Image.Image, GenerationID]
    ]
)

Convert from base64 or download an image from a response.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

@abstractmethod
def download_image_from_generation(
    self,
    generation: ImageGeneration,
) -> PIL.Image.Image | Coroutine[None, None, tuple[PIL.Image.Image, GenerationID]]:
    """Convert from base64 or download an image from a response."""

download_image_from_url abstractmethod

download_image_from_url(
    url: str,
) -> (
    PIL.Image.Image | Coroutine[None, None, PIL.Image.Image]
)

Download an image from a URL.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

@abstractmethod
def download_image_from_url(
    self,
    url: str,
) -> PIL.Image.Image | Coroutine[None, None, PIL.Image.Image]:
    """Download an image from a URL."""

AIHordeAPISimpleClient

Bases: BaseAIHordeSimpleClient

A simple client for the AI-Horde API. This is the easiest way to get started.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPISimpleClient(BaseAIHordeSimpleClient):
    """A simple client for the AI-Horde API. This is the easiest way to get started."""

    def download_image_from_generation(self, generation: ImageGeneration) -> PIL.Image.Image:
        """Convert from base64 or download an image from a response synchronously.

        Args:
            generation (ImageGeneration): The image generation to convert.

        Returns:
            PIL.Image.Image: The converted image.

        Raises:
            ClientResponseError: If the generation couldn't be downloaded.
            binascii.Error: If the image couldn't be parsed from base 64.
            RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

        """
        return download_image_from_generation(generation)

    def download_image_from_url(self, url: str) -> PIL.Image.Image:
        """Download an image from a URL synchronously.

        Args:
            url (str): The URL to download the image from.

        Returns:
            PIL.Image.Image: The downloaded image.

        Raises:
            ClientResponseError: If the image couldn't be downloaded.
            binascii.Error: If the image couldn't be parsed from base 64.
            RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

        """
        return download_image_from_url(url)

    @logfire.instrument()
    def _do_request_with_check(
        self,
        api_request: BaseAIHordeRequest,
        *,
        number_of_responses: int = 1,
        timeout: int = GENERATION_MAX_LIFE,
        check_callback: Callable[[HordeResponse], None] | None = None,
        check_callback_type: type[ResponseWithProgressMixin | ResponseGenerationProgressCombinedMixin] | None = None,
    ) -> tuple[HordeResponse, GenerationID]:
        """Submit a request which requires check/status polling to the AI-Horde API, and wait for it to complete.

        Args:
            api_request (BaseAIHordeRequest): The request to submit.
            number_of_responses (int, optional): The number of responses to expect. Defaults to 1.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to DEFAULT_GENERATION_TIMEOUT.
            check_callback (Callable[[HordeResponse], None], optional): A callback to call with the check response.
            check_callback_type (type[ResponseWithProgressMixin | ResponseGenerationProgressCombinedMixin], optional):
                The type of response expected by the callback.

        Returns:
            tuple[HordeResponse, GenerationID]: The final response and the corresponding job ID.
        """
        if check_callback is not None and len(inspect.getfullargspec(check_callback).args) == 0:
            raise ValueError("Callback must take at least one argument")

        logger.debug("Starting request with check")

        # This session class will cleanup incomplete requests in the event of an exception
        with AIHordeAPIClientSession() as horde_session:
            logger.debug(
                f"Submitting request: {api_request.log_safe_model_dump()} with timeout {timeout}",
            )
            initial_response = horde_session.submit_request(
                api_request=api_request,
                expected_response_type=api_request.get_default_success_response_type(),
            )

            # Handle the initial response to get the check request, job ID, and follow-up data
            check_request, gen_id, follow_up_data = self._handle_initial_response(initial_response)

            # There is a rate limit, so we start a clock to keep track of how long we've been waiting
            start_time = time.time()
            check_count = 0
            check_response: HordeResponse

            # Wait for the image generation to complete, checking every 4 seconds
            while True:
                check_count += 1

                # Submit the check request
                check_response = horde_session.submit_request(
                    api_request=check_request,
                    expected_response_type=check_request.get_default_success_response_type(),
                )

                # Handle the progress response to determine if the job is finished or timed out
                progress_state = self._handle_progress_response(
                    check_request,
                    check_response,
                    gen_id,
                    check_count=check_count,
                    number_of_responses=number_of_responses,
                    start_time=start_time,
                    timeout=timeout,
                    check_callback=check_callback,
                    check_callback_type=check_callback_type,
                )

                if progress_state == PROGRESS_STATE.finished or progress_state == PROGRESS_STATE.timed_out:
                    break

                # Wait for 4 seconds before checking again
                sleep_time = 4
                with logfire.span(self._msg_format_sleep.format(seconds=sleep_time), sleep_time=sleep_time):
                    time.sleep(sleep_time)

            # Check if the check response has progress
            if not isinstance(check_response, ResponseWithProgressMixin):
                raise RuntimeError(f"Response did not have progress: {check_response}")

            # Get the finalize request type from the check response
            finalize_request_type = check_response.get_finalize_success_request_type()

            # Set the final response to the check response by default
            final_response: HordeResponse = check_response

            # If there is a finalize request type, submit the finalize request
            if finalize_request_type:
                status_request = finalize_request_type.model_validate(follow_up_data[0])

                if not isinstance(status_request, JobRequestMixin):
                    logger.error(f"Finalize request type is not a JobRequestMixin: {finalize_request_type}")
                    raise RuntimeError(f"Finalize request type is not a JobRequestMixin: {finalize_request_type}")

                final_response = horde_session.submit_request(
                    api_request=status_request,
                    expected_response_type=status_request.get_default_success_response_type(),
                )

                if isinstance(final_response, RequestErrorResponse):
                    raise AIHordeRequestError(final_response)

            # Log a message indicating that the request is complete
            logger.log(COMPLETE_LOGGER_LABEL, f"Request complete: {gen_id}")

            # Return the final response and job ID
            return (final_response, gen_id)

        # If there is an exception, log an error and raise a RuntimeError
        logger.error("Something went wrong with the request:")
        logger.error(f"Request: {api_request.log_safe_model_dump()}")
        raise RuntimeError("Something went wrong with the request")

    def heartbeat_request(
        self,
    ) -> AIHordeHeartbeatResponse:
        """Submit a heartbeat request to the AI-Horde API.

        Returns:
            AIHordeHeartbeatResponse: The response from the API.
        """
        api_request = AIHordeHeartbeatRequest()

        with AIHordeAPIClientSession() as horde_session:
            api_response = horde_session.submit_request(api_request, api_request.get_default_success_response_type())

            if isinstance(api_response, RequestErrorResponse):
                raise AIHordeRequestError(api_response)

            return api_response

        raise RuntimeError("Something went wrong with the request")

    def image_generate_request(
        self,
        image_gen_request: ImageGenerateAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,
        check_callback: Callable[[ImageGenerateCheckResponse], None] | None = None,
    ) -> tuple[ImageGenerateStatusResponse, GenerationID]:
        """Submit an image generation request to the AI-Horde API, and wait for it to complete.

        Args:
            image_gen_request (ImageGenerateAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to -1.
            check_callback (Callable[[ImageGenerateCheckResponse], None], optional): A callback to call with the check
                response.

        Returns:
            list[ImageGeneration]: The completed images.

        Raises:
            ClientResponseError: If the generation couldn't be downloaded.
            binascii.Error: If the image couldn't be parsed from base 64.
            RuntimeError: If the image couldn't be downloaded or parsed for any other reason.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        timeout = self.validate_timeout(timeout, log_message=True)

        n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {n} images.")
        final_response, GenerationID = self._do_request_with_check(
            image_gen_request,
            number_of_responses=n,
            timeout=timeout,
            check_callback=generic_callback,
            check_callback_type=ImageGenerateCheckResponse,
        )

        if isinstance(final_response, RequestErrorResponse):  # pragma: no cover
            logger.error(f"Error response received: {final_response.message}")
            raise AIHordeRequestError(final_response)

        if not isinstance(final_response, ImageGenerateStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not an ImageGenerateStatusResponse")

        return (final_response, GenerationID)

    def image_generate_request_dry_run(
        self,
        image_gen_request: ImageGenerateAsyncRequest,
    ) -> ImageGenerateAsyncDryRunResponse:
        """Submit a dry run image generation, which will return the kudos cost without actually generating images.

        Args:
            image_gen_request (ImageGenerateAsyncRequest): The request to submit.

        Returns:
            ImageGenerateAsyncDryRunResponse: The response from the API.
        """
        if not image_gen_request.dry_run:
            raise RuntimeError("Dry run request must have dry_run set to True")

        n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting dry run for {n} images.")

        with AIHordeAPIClientSession() as horde_session:
            dry_run_response = horde_session.submit_request(image_gen_request, ImageGenerateAsyncDryRunResponse)

            if isinstance(dry_run_response, RequestErrorResponse):  # pragma: no cover
                logger.error(f"Error response received: {dry_run_response.message}")
                raise AIHordeRequestError(dry_run_response)

            return dry_run_response

        raise RuntimeError("Something went wrong with the request")

    def alchemy_request(
        self,
        alchemy_request: AlchemyAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,
        check_callback: Callable[[AlchemyStatusResponse], None] | None = None,
    ) -> tuple[AlchemyStatusResponse, GenerationID]:
        """Submit an alchemy request to the AI-Horde API, and wait for it to complete.

        Args:
            alchemy_request (AlchemyAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to -1.
            check_callback (Callable[[AlchemyStatusResponse], None], optional): A callback to call with the check
                response.

        Returns:
            AlchemyStatusResponse: The completed alchemy request(s).

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        timeout = self.validate_timeout(timeout, log_message=True)

        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {len(alchemy_request.forms)} alchemy requests.")
        for form in alchemy_request.forms:
            logger.debug(f"Request: {form}")

        response, gen_id = self._do_request_with_check(
            alchemy_request,
            number_of_responses=len(alchemy_request.forms),
            timeout=timeout,
            check_callback=generic_callback,
        )

        if isinstance(response, RequestErrorResponse):  # pragma: no cover
            raise AIHordeRequestError(response)

        if not isinstance(response, AlchemyStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not an AlchemyAsyncResponse")

        return (response, gen_id)

    def text_generate_request(
        self,
        text_gen_request: TextGenerateAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,
        check_callback: Callable[[TextGenerateStatusResponse], None] | None = None,
    ) -> tuple[TextGenerateStatusResponse, GenerationID]:
        """Submit a text generation request to the AI-Horde API, and wait for it to complete.

        Args:
            text_gen_request (TextGenerateAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to -1.
            check_callback (Callable[[TextGenerateStatusResponse], None], optional): A callback to call with the check
                response.

        Returns:
            TextGenerateStatusResponse: The completed text generation request.

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        timeout = self.validate_timeout(timeout, log_message=True)

        num_gens_requested = 1

        if text_gen_request.params and text_gen_request.params.n:
            num_gens_requested = text_gen_request.params.n

        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {num_gens_requested} text generation.")
        logger.debug(f"Request: {text_gen_request}")

        response, gen_id = self._do_request_with_check(
            text_gen_request,
            number_of_responses=1,
            timeout=timeout,
            check_callback=generic_callback,
            check_callback_type=TextGenerateStatusResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        if not isinstance(response, TextGenerateStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not a TextGenerateStatusResponse")

        return (response, gen_id)

    def text_generate_request_dry_run(
        self,
        text_gen_request: TextGenerateAsyncRequest,
    ) -> TextGenerateAsyncDryRunResponse:
        """Submit a dry run text generation, which will return the kudos cost without actually generating text.

        Args:
            text_gen_request (TextGenerateAsyncRequest): The request to submit.

        Returns:
            TextGenerateAsyncDryRunResponse: The response from the API.
        """
        if not text_gen_request.dry_run:
            raise RuntimeError("Dry run request must have dry_run set to True")

        logger.log(PROGRESS_LOGGER_LABEL, "Requesting dry run text generation.")
        logger.debug(f"Request: {text_gen_request}")

        with AIHordeAPIClientSession() as horde_session:
            dry_run_response = horde_session.submit_request(text_gen_request, TextGenerateAsyncDryRunResponse)

            if isinstance(dry_run_response, RequestErrorResponse):  # pragma: no cover
                logger.error(f"Error response received: {dry_run_response.message}")
                raise AIHordeRequestError(dry_run_response)

            return dry_run_response

        raise RuntimeError("Something went wrong with the request")

    def workers_all_details(
        self,
        worker_name: str | None = None,
        *,
        api_key: str | None = None,
    ) -> AllWorkersDetailsResponse:
        """Get all the details for all workers.

        Args:
            worker_name (str, optional): The name of the worker to get the details for.
            api_key (str, optional): The API key to use for the request.

        Returns:
            WorkersAllDetailsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                AllWorkersDetailsRequest(name=worker_name, apikey=api_key),
                AllWorkersDetailsResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def worker_details(
        self,
        worker_id: WorkerID | str,
        *,
        api_key: str | None = None,
    ) -> SingleWorkerDetailsResponse:
        """Get the details for a worker.

        Args:
            worker_id (WorkerID): The ID of the worker to get the details for.
            api_key (str, optional): The API key to use for the request.

        Returns:
            SingleWorkerDetailsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                SingleWorkerDetailsRequest(worker_id=worker_id, apikey=api_key),
                SingleWorkerDetailsResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def worker_details_by_name(
        self,
        worker_name: str,
        *,
        api_key: str | None = None,
    ) -> SingleWorkerDetailsResponse:
        """Get the details for a worker by worker name.

        Args:
            worker_name (str): The ID of the worker to get the details for.
            api_key (str, optional): The API key to use for the request.

        Returns:
            SingleWorkerDetailsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                SingleWorkerNameDetailsRequest(worker_name=worker_name, apikey=api_key),
                SingleWorkerDetailsResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def worker_modify(
        self,
        modify_worker_request: ModifyWorkerRequest,
    ) -> ModifyWorkerResponse:
        """Update a worker.

        Args:
            modify_worker_request (ModifyWorkerRequest): The request to update the worker.

        Returns:
            ModifyWorkerResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                modify_worker_request,
                ModifyWorkerResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def worker_delete(
        self,
        worker_id: WorkerID | str,
        *,
        api_key: str | None = None,
    ) -> DeleteWorkerResponse:
        """Delete a worker.

        Args:
            worker_id (WorkerID): The ID of the worker to delete.
            api_key (str, optional): The API key to use for the request.

        Returns:
            DeleteWorkerResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                DeleteWorkerRequest(worker_id=worker_id, apikey=api_key),
                DeleteWorkerResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def image_stats_totals(
        self,
    ) -> ImageStatsModelsTotalResponse:
        """Get the total stats for images.

        Returns:
            ImageStatsTotalsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(ImageStatsModelsTotalRequest(), ImageStatsModelsTotalResponse)

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def image_stats_models(
        self,
        model_state: str | MODEL_STATE = MODEL_STATE.known,
    ) -> ImageStatsModelsResponse:
        """Get the stats for images by model.

        Returns:
            ImageStatsModelsResponse: The response from the API.
        """
        model_state = MODEL_STATE(model_state)

        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                ImageStatsModelsRequest(model_state=model_state),
                ImageStatsModelsResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def text_stats_totals(
        self,
    ) -> TextStatsModelsTotalResponse:
        """Get the total stats for text.

        Returns:
            TextStatsTotalsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(TextStatsModelsTotalRequest(), TextStatsModelsTotalResponse)

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def text_stats_models(
        self,
    ) -> TextStatsModelResponse:
        """Get the stats for text by model.

        Returns:
            TextModelStatsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(TextStatsModelsRequest(), TextStatsModelResponse)

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def image_status_models_all(
        self,
    ) -> HordeStatusModelsAllResponse:
        """Get the status of all image models.

        Returns:
            ImageStatusModelsAllResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(HordeStatusModelsAllRequest(), HordeStatusModelsAllResponse)

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def image_status_models_single(
        self,
        model_name: str,
    ) -> HordeStatusModelsSingleResponse:
        """Get the status of a single image model.

        Args:
            model_name (str): The name of the model to get the status of.

        Returns:
            ImageStatusModelsSingleResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(
                HordeStatusModelsSingleRequest(model_name=model_name),
                HordeStatusModelsSingleResponse,
            )

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

    def get_news(
        self,
    ) -> NewsResponse:
        """Get the latest news from the AI-Horde API.

        Returns:
            NewsResponse: The response from the API.
        """
        with AIHordeAPIClientSession() as horde_session:
            response = horde_session.submit_request(NewsRequest(), NewsResponse)

            if isinstance(response, RequestErrorResponse):
                raise AIHordeRequestError(response)

            return response

        raise RuntimeError("Something went wrong with the request")

reasonable_minimum_timeout class-attribute instance-attribute

reasonable_minimum_timeout = 20

download_image_from_generation

download_image_from_generation(
    generation: ImageGeneration,
) -> PIL.Image.Image

Convert from base64 or download an image from a response synchronously.

Parameters:

• generation (ImageGeneration) –

  The image generation to convert.

Returns:

• Image –

  PIL.Image.Image: The converted image.

Raises:

• ClientResponseError –

  If the generation couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def download_image_from_generation(self, generation: ImageGeneration) -> PIL.Image.Image:
    """Convert from base64 or download an image from a response synchronously.

    Args:
        generation (ImageGeneration): The image generation to convert.

    Returns:
        PIL.Image.Image: The converted image.

    Raises:
        ClientResponseError: If the generation couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    return download_image_from_generation(generation)

download_image_from_url

download_image_from_url(url: str) -> PIL.Image.Image

Download an image from a URL synchronously.

Parameters:

• url (str) –

  The URL to download the image from.

Returns:

• Image –

  PIL.Image.Image: The downloaded image.

Raises:

• ClientResponseError –

  If the image couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def download_image_from_url(self, url: str) -> PIL.Image.Image:
    """Download an image from a URL synchronously.

    Args:
        url (str): The URL to download the image from.

    Returns:
        PIL.Image.Image: The downloaded image.

    Raises:
        ClientResponseError: If the image couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    return download_image_from_url(url)

heartbeat_request

heartbeat_request() -> AIHordeHeartbeatResponse

Submit a heartbeat request to the AI-Horde API.

Returns:

• AIHordeHeartbeatResponse ( AIHordeHeartbeatResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def heartbeat_request(
    self,
) -> AIHordeHeartbeatResponse:
    """Submit a heartbeat request to the AI-Horde API.

    Returns:
        AIHordeHeartbeatResponse: The response from the API.
    """
    api_request = AIHordeHeartbeatRequest()

    with AIHordeAPIClientSession() as horde_session:
        api_response = horde_session.submit_request(api_request, api_request.get_default_success_response_type())

        if isinstance(api_response, RequestErrorResponse):
            raise AIHordeRequestError(api_response)

        return api_response

    raise RuntimeError("Something went wrong with the request")

image_generate_request

image_generate_request(
    image_gen_request: ImageGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[ImageGenerateCheckResponse], None] | None
    ) = None,
) -> tuple[ImageGenerateStatusResponse, GenerationID]

Submit an image generation request to the AI-Horde API, and wait for it to complete.

Parameters:

• image_gen_request (ImageGenerateAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Defaults to -1.

• check_callback (Callable[[ImageGenerateCheckResponse], None], default: None ) –

  A callback to call with the check response.

Returns:

• tuple[ImageGenerateStatusResponse, GenerationID] –

  list[ImageGeneration]: The completed images.

Raises:

• ClientResponseError –

  If the generation couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_generate_request(
    self,
    image_gen_request: ImageGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: Callable[[ImageGenerateCheckResponse], None] | None = None,
) -> tuple[ImageGenerateStatusResponse, GenerationID]:
    """Submit an image generation request to the AI-Horde API, and wait for it to complete.

    Args:
        image_gen_request (ImageGenerateAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Defaults to -1.
        check_callback (Callable[[ImageGenerateCheckResponse], None], optional): A callback to call with the check
            response.

    Returns:
        list[ImageGeneration]: The completed images.

    Raises:
        ClientResponseError: If the generation couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    timeout = self.validate_timeout(timeout, log_message=True)

    n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {n} images.")
    final_response, GenerationID = self._do_request_with_check(
        image_gen_request,
        number_of_responses=n,
        timeout=timeout,
        check_callback=generic_callback,
        check_callback_type=ImageGenerateCheckResponse,
    )

    if isinstance(final_response, RequestErrorResponse):  # pragma: no cover
        logger.error(f"Error response received: {final_response.message}")
        raise AIHordeRequestError(final_response)

    if not isinstance(final_response, ImageGenerateStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not an ImageGenerateStatusResponse")

    return (final_response, GenerationID)

image_generate_request_dry_run

image_generate_request_dry_run(
    image_gen_request: ImageGenerateAsyncRequest,
) -> ImageGenerateAsyncDryRunResponse

Submit a dry run image generation, which will return the kudos cost without actually generating images.

Parameters:

• image_gen_request (ImageGenerateAsyncRequest) –

  The request to submit.

Returns:

• ImageGenerateAsyncDryRunResponse ( ImageGenerateAsyncDryRunResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_generate_request_dry_run(
    self,
    image_gen_request: ImageGenerateAsyncRequest,
) -> ImageGenerateAsyncDryRunResponse:
    """Submit a dry run image generation, which will return the kudos cost without actually generating images.

    Args:
        image_gen_request (ImageGenerateAsyncRequest): The request to submit.

    Returns:
        ImageGenerateAsyncDryRunResponse: The response from the API.
    """
    if not image_gen_request.dry_run:
        raise RuntimeError("Dry run request must have dry_run set to True")

    n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting dry run for {n} images.")

    with AIHordeAPIClientSession() as horde_session:
        dry_run_response = horde_session.submit_request(image_gen_request, ImageGenerateAsyncDryRunResponse)

        if isinstance(dry_run_response, RequestErrorResponse):  # pragma: no cover
            logger.error(f"Error response received: {dry_run_response.message}")
            raise AIHordeRequestError(dry_run_response)

        return dry_run_response

    raise RuntimeError("Something went wrong with the request")

alchemy_request

alchemy_request(
    alchemy_request: AlchemyAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[AlchemyStatusResponse], None] | None
    ) = None,
) -> tuple[AlchemyStatusResponse, GenerationID]

Submit an alchemy request to the AI-Horde API, and wait for it to complete.

Parameters:

• alchemy_request (AlchemyAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Defaults to -1.

• check_callback (Callable[[AlchemyStatusResponse], None], default: None ) –

  A callback to call with the check response.

Returns:

• AlchemyStatusResponse ( tuple[AlchemyStatusResponse, GenerationID] ) –

  The completed alchemy request(s).

Raises:

• AIHordeRequestError –

  If the request failed. The error response is included in the exception.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def alchemy_request(
    self,
    alchemy_request: AlchemyAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: Callable[[AlchemyStatusResponse], None] | None = None,
) -> tuple[AlchemyStatusResponse, GenerationID]:
    """Submit an alchemy request to the AI-Horde API, and wait for it to complete.

    Args:
        alchemy_request (AlchemyAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Defaults to -1.
        check_callback (Callable[[AlchemyStatusResponse], None], optional): A callback to call with the check
            response.

    Returns:
        AlchemyStatusResponse: The completed alchemy request(s).

    Raises:
        AIHordeRequestError: If the request failed. The error response is included in the exception.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    timeout = self.validate_timeout(timeout, log_message=True)

    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {len(alchemy_request.forms)} alchemy requests.")
    for form in alchemy_request.forms:
        logger.debug(f"Request: {form}")

    response, gen_id = self._do_request_with_check(
        alchemy_request,
        number_of_responses=len(alchemy_request.forms),
        timeout=timeout,
        check_callback=generic_callback,
    )

    if isinstance(response, RequestErrorResponse):  # pragma: no cover
        raise AIHordeRequestError(response)

    if not isinstance(response, AlchemyStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not an AlchemyAsyncResponse")

    return (response, gen_id)

text_generate_request

text_generate_request(
    text_gen_request: TextGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[TextGenerateStatusResponse], None] | None
    ) = None,
) -> tuple[TextGenerateStatusResponse, GenerationID]

Submit a text generation request to the AI-Horde API, and wait for it to complete.

Parameters:

• text_gen_request (TextGenerateAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Defaults to -1.

• check_callback (Callable[[TextGenerateStatusResponse], None], default: None ) –

  A callback to call with the check response.

Returns:

• TextGenerateStatusResponse ( tuple[TextGenerateStatusResponse, GenerationID] ) –

  The completed text generation request.

Raises:

• AIHordeRequestError –

  If the request failed. The error response is included in the exception.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def text_generate_request(
    self,
    text_gen_request: TextGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: Callable[[TextGenerateStatusResponse], None] | None = None,
) -> tuple[TextGenerateStatusResponse, GenerationID]:
    """Submit a text generation request to the AI-Horde API, and wait for it to complete.

    Args:
        text_gen_request (TextGenerateAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Defaults to -1.
        check_callback (Callable[[TextGenerateStatusResponse], None], optional): A callback to call with the check
            response.

    Returns:
        TextGenerateStatusResponse: The completed text generation request.

    Raises:
        AIHordeRequestError: If the request failed. The error response is included in the exception.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    timeout = self.validate_timeout(timeout, log_message=True)

    num_gens_requested = 1

    if text_gen_request.params and text_gen_request.params.n:
        num_gens_requested = text_gen_request.params.n

    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {num_gens_requested} text generation.")
    logger.debug(f"Request: {text_gen_request}")

    response, gen_id = self._do_request_with_check(
        text_gen_request,
        number_of_responses=1,
        timeout=timeout,
        check_callback=generic_callback,
        check_callback_type=TextGenerateStatusResponse,
    )

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    if not isinstance(response, TextGenerateStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not a TextGenerateStatusResponse")

    return (response, gen_id)

text_generate_request_dry_run

text_generate_request_dry_run(
    text_gen_request: TextGenerateAsyncRequest,
) -> TextGenerateAsyncDryRunResponse

Submit a dry run text generation, which will return the kudos cost without actually generating text.

Parameters:

• text_gen_request (TextGenerateAsyncRequest) –

  The request to submit.

Returns:

• TextGenerateAsyncDryRunResponse ( TextGenerateAsyncDryRunResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def text_generate_request_dry_run(
    self,
    text_gen_request: TextGenerateAsyncRequest,
) -> TextGenerateAsyncDryRunResponse:
    """Submit a dry run text generation, which will return the kudos cost without actually generating text.

    Args:
        text_gen_request (TextGenerateAsyncRequest): The request to submit.

    Returns:
        TextGenerateAsyncDryRunResponse: The response from the API.
    """
    if not text_gen_request.dry_run:
        raise RuntimeError("Dry run request must have dry_run set to True")

    logger.log(PROGRESS_LOGGER_LABEL, "Requesting dry run text generation.")
    logger.debug(f"Request: {text_gen_request}")

    with AIHordeAPIClientSession() as horde_session:
        dry_run_response = horde_session.submit_request(text_gen_request, TextGenerateAsyncDryRunResponse)

        if isinstance(dry_run_response, RequestErrorResponse):  # pragma: no cover
            logger.error(f"Error response received: {dry_run_response.message}")
            raise AIHordeRequestError(dry_run_response)

        return dry_run_response

    raise RuntimeError("Something went wrong with the request")

workers_all_details

workers_all_details(
    worker_name: str | None = None,
    *,
    api_key: str | None = None
) -> AllWorkersDetailsResponse

Get all the details for all workers.

Parameters:

• worker_name (str, default: None ) –

  The name of the worker to get the details for.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• WorkersAllDetailsResponse ( AllWorkersDetailsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def workers_all_details(
    self,
    worker_name: str | None = None,
    *,
    api_key: str | None = None,
) -> AllWorkersDetailsResponse:
    """Get all the details for all workers.

    Args:
        worker_name (str, optional): The name of the worker to get the details for.
        api_key (str, optional): The API key to use for the request.

    Returns:
        WorkersAllDetailsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            AllWorkersDetailsRequest(name=worker_name, apikey=api_key),
            AllWorkersDetailsResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

worker_details

worker_details(
    worker_id: WorkerID | str, *, api_key: str | None = None
) -> SingleWorkerDetailsResponse

Get the details for a worker.

Parameters:

• worker_id (WorkerID) –

  The ID of the worker to get the details for.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• SingleWorkerDetailsResponse ( SingleWorkerDetailsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def worker_details(
    self,
    worker_id: WorkerID | str,
    *,
    api_key: str | None = None,
) -> SingleWorkerDetailsResponse:
    """Get the details for a worker.

    Args:
        worker_id (WorkerID): The ID of the worker to get the details for.
        api_key (str, optional): The API key to use for the request.

    Returns:
        SingleWorkerDetailsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            SingleWorkerDetailsRequest(worker_id=worker_id, apikey=api_key),
            SingleWorkerDetailsResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

worker_details_by_name

worker_details_by_name(
    worker_name: str, *, api_key: str | None = None
) -> SingleWorkerDetailsResponse

Get the details for a worker by worker name.

Parameters:

• worker_name (str) –

  The ID of the worker to get the details for.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• SingleWorkerDetailsResponse ( SingleWorkerDetailsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def worker_details_by_name(
    self,
    worker_name: str,
    *,
    api_key: str | None = None,
) -> SingleWorkerDetailsResponse:
    """Get the details for a worker by worker name.

    Args:
        worker_name (str): The ID of the worker to get the details for.
        api_key (str, optional): The API key to use for the request.

    Returns:
        SingleWorkerDetailsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            SingleWorkerNameDetailsRequest(worker_name=worker_name, apikey=api_key),
            SingleWorkerDetailsResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

worker_modify

worker_modify(
    modify_worker_request: ModifyWorkerRequest,
) -> ModifyWorkerResponse

Update a worker.

Parameters:

• modify_worker_request (ModifyWorkerRequest) –

  The request to update the worker.

Returns:

• ModifyWorkerResponse ( ModifyWorkerResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def worker_modify(
    self,
    modify_worker_request: ModifyWorkerRequest,
) -> ModifyWorkerResponse:
    """Update a worker.

    Args:
        modify_worker_request (ModifyWorkerRequest): The request to update the worker.

    Returns:
        ModifyWorkerResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            modify_worker_request,
            ModifyWorkerResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

worker_delete

worker_delete(
    worker_id: WorkerID | str, *, api_key: str | None = None
) -> DeleteWorkerResponse

Delete a worker.

Parameters:

• worker_id (WorkerID) –

  The ID of the worker to delete.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• DeleteWorkerResponse ( DeleteWorkerResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def worker_delete(
    self,
    worker_id: WorkerID | str,
    *,
    api_key: str | None = None,
) -> DeleteWorkerResponse:
    """Delete a worker.

    Args:
        worker_id (WorkerID): The ID of the worker to delete.
        api_key (str, optional): The API key to use for the request.

    Returns:
        DeleteWorkerResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            DeleteWorkerRequest(worker_id=worker_id, apikey=api_key),
            DeleteWorkerResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

image_stats_totals

image_stats_totals() -> ImageStatsModelsTotalResponse

Get the total stats for images.

Returns:

• ImageStatsTotalsResponse ( ImageStatsModelsTotalResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_stats_totals(
    self,
) -> ImageStatsModelsTotalResponse:
    """Get the total stats for images.

    Returns:
        ImageStatsTotalsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(ImageStatsModelsTotalRequest(), ImageStatsModelsTotalResponse)

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

image_stats_models

image_stats_models(
    model_state: str | MODEL_STATE = MODEL_STATE.known,
) -> ImageStatsModelsResponse

Get the stats for images by model.

Returns:

• ImageStatsModelsResponse ( ImageStatsModelsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_stats_models(
    self,
    model_state: str | MODEL_STATE = MODEL_STATE.known,
) -> ImageStatsModelsResponse:
    """Get the stats for images by model.

    Returns:
        ImageStatsModelsResponse: The response from the API.
    """
    model_state = MODEL_STATE(model_state)

    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            ImageStatsModelsRequest(model_state=model_state),
            ImageStatsModelsResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

text_stats_totals

text_stats_totals() -> TextStatsModelsTotalResponse

Get the total stats for text.

Returns:

• TextStatsTotalsResponse ( TextStatsModelsTotalResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def text_stats_totals(
    self,
) -> TextStatsModelsTotalResponse:
    """Get the total stats for text.

    Returns:
        TextStatsTotalsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(TextStatsModelsTotalRequest(), TextStatsModelsTotalResponse)

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

text_stats_models

text_stats_models() -> TextStatsModelResponse

Get the stats for text by model.

Returns:

• TextModelStatsResponse ( TextStatsModelResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def text_stats_models(
    self,
) -> TextStatsModelResponse:
    """Get the stats for text by model.

    Returns:
        TextModelStatsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(TextStatsModelsRequest(), TextStatsModelResponse)

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

image_status_models_all

image_status_models_all() -> HordeStatusModelsAllResponse

Get the status of all image models.

Returns:

• ImageStatusModelsAllResponse ( HordeStatusModelsAllResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_status_models_all(
    self,
) -> HordeStatusModelsAllResponse:
    """Get the status of all image models.

    Returns:
        ImageStatusModelsAllResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(HordeStatusModelsAllRequest(), HordeStatusModelsAllResponse)

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

image_status_models_single

image_status_models_single(
    model_name: str,
) -> HordeStatusModelsSingleResponse

Get the status of a single image model.

Parameters:

• model_name (str) –

  The name of the model to get the status of.

Returns:

• ImageStatusModelsSingleResponse ( HordeStatusModelsSingleResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def image_status_models_single(
    self,
    model_name: str,
) -> HordeStatusModelsSingleResponse:
    """Get the status of a single image model.

    Args:
        model_name (str): The name of the model to get the status of.

    Returns:
        ImageStatusModelsSingleResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(
            HordeStatusModelsSingleRequest(model_name=model_name),
            HordeStatusModelsSingleResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

get_news

get_news() -> NewsResponse

Get the latest news from the AI-Horde API.

Returns:

• NewsResponse ( NewsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def get_news(
    self,
) -> NewsResponse:
    """Get the latest news from the AI-Horde API.

    Returns:
        NewsResponse: The response from the API.
    """
    with AIHordeAPIClientSession() as horde_session:
        response = horde_session.submit_request(NewsRequest(), NewsResponse)

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    raise RuntimeError("Something went wrong with the request")

validate_timeout

validate_timeout(
    timeout: int, log_message: bool = False
) -> int

Check if a timeout is reasonable.

Parameters:

• timeout (int) –

  The timeout to check.

• log_message (bool, default: False ) –

  Whether to log a message if the timeout is too short. Defaults to False.

Returns:

• bool ( int ) –

  True if the timeout is reasonable, False otherwise.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def validate_timeout(
    self,
    timeout: int,
    log_message: bool = False,
) -> int:
    """Check if a timeout is reasonable.

    Args:
        timeout (int): The timeout to check.
        log_message (bool, optional): Whether to log a message if the timeout is too short. Defaults to False.

    Returns:
        bool: True if the timeout is reasonable, False otherwise.
    """
    if timeout <= 0:  # pragma: no cover
        logger.debug(f"No timeout set. Using default timeout of {GENERATION_MAX_LIFE} seconds.")
        return GENERATION_MAX_LIFE

    if timeout > GENERATION_MAX_LIFE:  # pragma: no cover
        logger.warning(
            f"Timeout ({timeout}) is greater than the maximum possible of {GENERATION_MAX_LIFE} seconds."
            f"Using {GENERATION_MAX_LIFE} for the timeout.",
        )
        return GENERATION_MAX_LIFE

    if timeout < self.reasonable_minimum_timeout and log_message:  # pragma: no cover
        logger.warning(
            f"test_simple_client_async_image_generate_multiple than {self.reasonable_minimum_timeout} seconds, "
            "this is probably too short.",
        )

    return timeout

AIHordeAPIAsyncSimpleClient

Bases: BaseAIHordeSimpleClient

An asyncio based simple client for the AI-Horde API. Start with this class if you want asyncio capabilities..

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

class AIHordeAPIAsyncSimpleClient(BaseAIHordeSimpleClient):
    """An asyncio based simple client for the AI-Horde API. Start with this class if you want asyncio capabilities.."""

    _horde_client_session: AIHordeAPIAsyncClientSession

    def __init__(
        self,
        aiohttp_session: aiohttp.ClientSession | None = None,
        horde_client_session: AIHordeAPIAsyncClientSession | None = None,
        apikey: str | None = None,
    ) -> None:
        """Create a new instance of the AIHordeAPISimpleClient."""
        super().__init__()

        if aiohttp_session is None and horde_client_session is None:
            raise RuntimeError("No aiohttp session provided but an async request was made.")

        if (
            aiohttp_session is not None
            and horde_client_session is not None
            and horde_client_session._aiohttp_session != aiohttp_session
        ):
            raise RuntimeError("The aiohttp session provided does not match the session in the client session.")

        if aiohttp_session is not None and horde_client_session is None:
            logger.info("Creating a new AIHordeAPIAsyncClientSession with the provided aiohttp session.")
            self._aiohttp_session = aiohttp_session
            self._horde_client_session = AIHordeAPIAsyncClientSession(aiohttp_session, apikey=apikey)
        elif horde_client_session is not None:
            self._horde_client_session = horde_client_session
            self._aiohttp_session = horde_client_session._aiohttp_session

    async def download_image_from_generation(
        self,
        generation: ImageGeneration,
    ) -> tuple[PIL.Image.Image, GenerationID]:
        """Asynchronously convert from base64 or download an image from a response.

        Args:
            generation (ImageGeneration): The image generation to convert.

        Returns:
            PIL.Image.Image: The converted image.

        Raises:
            ClientResponseError: If the generation couldn't be downloaded.
            binascii.Error: If the image couldn't be parsed from base 64.
            RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

        """
        if generation.img is None:  # pragma: no cover
            raise ValueError("Generation has no image")

        if self._aiohttp_session is None:  # pragma: no cover
            raise RuntimeError("No aiohttp session provided but an async request was made.")

        image_bytes: bytes | None = None
        if urllib.parse.urlparse(generation.img).scheme in ["http", "https"]:
            async with self._aiohttp_session.get(generation.img, ssl=_default_sslcontext) as response:
                if response.status != 200:  # pragma: no cover
                    logger.error(f"Error downloading image: {response.status}")
                    response.raise_for_status()

                image_bytes = await response.read()
        else:
            try:
                image_bytes = base64.b64decode(generation.img)
            except Exception as e:
                logger.error(f"Error parsing image: {e}")
                raise e

        if image_bytes is None:  # pragma: no cover
            raise RuntimeError("Error downloading or parsing image")

        return (PIL.Image.open(io.BytesIO(image_bytes)), generation.id_)

    async def download_image_from_url(self, url: str) -> PIL.Image.Image:
        """Asynchronously download an image from a URL.

        Args:
            url (str): The URL to download the image from.

        Returns:
            PIL.Image.Image: The downloaded image.

        Raises:
            ClientResponseError: If the image couldn't be downloaded.
            binascii.Error: If the image couldn't be parsed from base 64.
            RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

        """
        if self._aiohttp_session is None:
            raise RuntimeError("No aiohttp session provided but an async request was made.")

        async with self._aiohttp_session.get(url, ssl=_default_sslcontext) as response:
            if response.status != 200:  # pragma: no cover
                logger.error(f"Error downloading image: {response.status}")
                response.raise_for_status()

            image_bytes = await response.read()

        if image_bytes is None:  # pragma: no cover
            raise RuntimeError("Error downloading or parsing image")

        return PIL.Image.open(io.BytesIO(image_bytes))

    @logfire.instrument()
    async def _do_request_with_check(
        self,
        api_request: BaseAIHordeRequest,
        *,
        number_of_responses: int = 1,
        timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
        check_callback: Callable[[HordeResponse], None] | None = None,
        check_callback_type: type[ResponseWithProgressMixin | ResponseGenerationProgressCombinedMixin] | None = None,
    ) -> tuple[HordeResponse, GenerationID]:
        """Submit a request which requires check/status polling to the AI-Horde API, and wait for it to complete.

        Args:
            api_request (BaseAIHordeRequest): The request to submit.
            number_of_responses (int, optional): The number of responses to expect. Defaults to 1.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to GENERATION_MAX_LIFE.
            check_callback (Callable[[HordeResponse], None], optional): A callback to call with the check response.
            check_callback_type (type[ResponseWithProgressMixin | ResponseGenerationProgressCombinedMixin], optional):
                The type of response expected by the callback.

        Returns:
            tuple[HordeResponse, GenerationID]: The final response and the corresponding job ID.

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        if check_callback is not None and len(inspect.getfullargspec(check_callback).args) == 0:
            raise ValueError("Callback must take at least one argument")

        logger.debug("Starting async request with check.")

        # Submit the initial request
        logger.debug(
            f"Submitting request: {api_request.log_safe_model_dump()} with timeout {timeout}",
        )
        initial_response = await self._horde_client_session.submit_request(
            api_request=api_request,
            expected_response_type=api_request.get_default_success_response_type(),
        )

        # Handle the initial response to get the check request, job ID, and follow-up data
        check_request, gen_id, follow_up_data = self._handle_initial_response(initial_response)

        # There is a rate limit, so we start a clock to keep track of how long we've been waiting
        start_time = time.time()
        check_count = 0
        check_response: HordeResponse

        # Wait for the image generation to complete, checking every 4 seconds
        while True:
            check_count += 1

            # Submit the check request
            check_response = await self._horde_client_session.submit_request(
                api_request=check_request,
                expected_response_type=check_request.get_default_success_response_type(),
            )

            # Handle the progress response to determine if the job is finished or timed out
            progress_state = self._handle_progress_response(
                check_request,
                check_response,
                gen_id,
                check_count=check_count,
                number_of_responses=number_of_responses,
                start_time=start_time,
                timeout=timeout,
                check_callback=check_callback,
                check_callback_type=check_callback_type,
            )

            if progress_state == PROGRESS_STATE.finished or progress_state == PROGRESS_STATE.timed_out:
                break

            # Wait for 4 seconds before checking again
            sleep_time = 4
            with logfire.span(self._msg_format_sleep.format(seconds=sleep_time), sleep_time=sleep_time):
                await asyncio.sleep(sleep_time)

        # This is for type safety, but should never happen in production
        if not isinstance(check_response, ResponseWithProgressMixin):  # pragma: no cover
            raise RuntimeError(f"Response did not have progress: {check_response}")

        # Get the finalize request type from the check response
        finalize_request_type = check_response.get_finalize_success_request_type()

        # Set the final response to the check response by default
        final_response: HordeResponse = check_response

        # If there is a finalize request type, submit the finalize request
        if finalize_request_type:
            finalize_request = finalize_request_type.model_validate(follow_up_data[0])

            # This is for type safety, but should never happen in production
            if not isinstance(finalize_request, JobRequestMixin):  # pragma: no cover
                logger.error(
                    f"Finalize request type is not a JobRequestMixin: {finalize_request.log_safe_model_dump()}",
                )
                raise RuntimeError(
                    f"Finalize request type is not a JobRequestMixin: {finalize_request.log_safe_model_dump()}",
                )

            final_response = await self._horde_client_session.submit_request(
                api_request=finalize_request,
                expected_response_type=finalize_request.get_default_success_response_type(),
            )

            if isinstance(final_response, RequestErrorResponse):
                raise AIHordeRequestError(final_response)

        # Log a message indicating that the request is complete
        logger.log(COMPLETE_LOGGER_LABEL, f"Request complete: {gen_id}")

        # Return the final response and job ID
        return (final_response, gen_id)

    async def heartbeat_request(
        self,
    ) -> AIHordeHeartbeatResponse:
        """Submit a heartbeat request to the AI-Horde API.

        Returns:
            AIHordeHeartbeatResponse: The response from the API.
        """
        api_request = AIHordeHeartbeatRequest()

        if self._horde_client_session is not None:
            api_response = await self._horde_client_session.submit_request(
                api_request,
                api_request.get_default_success_response_type(),
            )

            if isinstance(api_response, RequestErrorResponse):
                raise AIHordeRequestError(api_response)

            return api_response

        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    async def image_generate_request(
        self,
        image_gen_request: ImageGenerateAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
        check_callback: Callable[[ImageGenerateCheckResponse], None] | None = None,
        delay: float = 0.0,
    ) -> tuple[ImageGenerateStatusResponse, GenerationID]:
        """Submit an image generation request to the AI-Horde API, and wait for it to complete.

        *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
        Space concurrent requests apart slightly to allow them to be less than 10/second.

        Args:
            image_gen_request (ImageGenerateAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely.
                Defaults to -1.
            check_callback (Callable[[ImageGenerateCheckResponse], None], optional): A callback to call with the check
                response.
            delay (float, optional): The number of seconds to wait before checking the status. Defaults to 0.0.


        Returns:
            tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        await asyncio.sleep(delay)

        timeout = self.validate_timeout(timeout, log_message=True)

        n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
        final_response, gen_id = await self._do_request_with_check(
            image_gen_request,
            number_of_responses=n,
            timeout=timeout,
            check_callback=generic_callback,
            check_callback_type=ImageGenerateCheckResponse,
        )

        if isinstance(final_response, RequestErrorResponse):  # pragma: no cover
            raise AIHordeRequestError(final_response)

        if not isinstance(final_response, ImageGenerateStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not an ImageGenerateStatusResponse")

        return (final_response, gen_id)

    async def image_generate_request_dry_run(
        self,
        image_gen_request: ImageGenerateAsyncRequest,
    ) -> ImageGenerateAsyncDryRunResponse:
        """Submit a dry run image generation, which will return the kudos cost without actually generating images.

        Args:
            image_gen_request (ImageGenerateAsyncRequest): The request to submit.

        Returns:
            ImageGenerateAsyncDryRunResponse: The response from the API.
        """
        if not image_gen_request.dry_run:
            raise RuntimeError("Dry run request must have dry_run set to True")

        n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting dry run for {n} images.")

        if self._horde_client_session is not None:
            dry_run_response = await self._horde_client_session.submit_request(
                image_gen_request,
                ImageGenerateAsyncDryRunResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(dry_run_response, RequestErrorResponse):
            logger.error(f"Error response received: {dry_run_response.message}")
            raise AIHordeRequestError(dry_run_response)

        return dry_run_response

    async def alchemy_request(
        self,
        alchemy_request: AlchemyAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
        check_callback: Callable[[AlchemyStatusResponse], None] | None = None,
    ) -> tuple[AlchemyStatusResponse, GenerationID]:
        """Submit an alchemy request to the AI-Horde API, and wait for it to complete.

        *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
        Space concurrent requests apart slightly to allow them to be less than 10/second.

        Args:
            alchemy_request (AlchemyAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Defaults to -1.
            check_callback (Callable[[AlchemyStatusResponse], None], optional): A callback to call with the check
                response.

        Returns:
            tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        timeout = self.validate_timeout(timeout, log_message=True)

        response, gen_id = await self._do_request_with_check(
            alchemy_request,
            number_of_responses=len(alchemy_request.forms),
            timeout=timeout,
            check_callback=generic_callback,
            check_callback_type=AlchemyStatusResponse,
        )
        if isinstance(response, RequestErrorResponse):  # pragma: no cover
            raise AIHordeRequestError(response)

        if not isinstance(response, AlchemyStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not an AlchemyAsyncResponse")

        return (response, gen_id)

    async def text_generate_request(
        self,
        text_gen_request: TextGenerateAsyncRequest,
        timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
        check_callback: Callable[[TextGenerateStatusResponse], None] | None = None,
        delay: float = 0.0,
    ) -> tuple[TextGenerateStatusResponse, GenerationID]:
        """Submit a text generation request to the AI-Horde API, and wait for it to complete.

        *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
        Space concurrent requests apart slightly to allow them to be less than 10/second.

        Args:
            text_gen_request (TextGenerateAsyncRequest): The request to submit.
            timeout (int, optional): The number of seconds to wait before aborting.
                returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely.
                Defaults to -1.
            check_callback (Callable[[TextGenerateStatusResponse], None], optional): A callback to call with the check
                response.
            delay (float, optional): The number of seconds to wait before checking the status. Defaults to 0.0.

        Returns:
            tuple[TextGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

        Raises:
            AIHordeRequestError: If the request failed. The error response is included in the exception.
        """
        # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
        # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
        # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
        generic_callback = cast(Callable[[HordeResponse], None], check_callback)

        await asyncio.sleep(delay)

        timeout = self.validate_timeout(timeout, log_message=True)

        num_gens_requested = 1

        if text_gen_request.params and text_gen_request.params.n:
            num_gens_requested = text_gen_request.params.n

        logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {num_gens_requested} text generation.")
        logger.debug(f"Request: {text_gen_request}")

        response, gen_id = await self._do_request_with_check(
            text_gen_request,
            number_of_responses=1,
            timeout=timeout,
            check_callback=generic_callback,
            check_callback_type=TextGenerateStatusResponse,
        )

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        if not isinstance(response, TextGenerateStatusResponse):  # pragma: no cover
            raise RuntimeError("Response was not a TextGenerateStatusResponse")

        return (response, gen_id)

    async def text_generate_request_dry_run(
        self,
        text_gen_request: TextGenerateAsyncRequest,
    ) -> TextGenerateAsyncDryRunResponse:
        """Submit a dry run text generation, which will return the kudos cost without actually generating text.

        Args:
            text_gen_request (TextGenerateAsyncRequest): The request to submit.

        Returns:
            TextGenerateAsyncDryRunResponse: The response from the API.
        """
        if not text_gen_request.dry_run:
            raise RuntimeError("Dry run request must have dry_run set to True")

        logger.log(PROGRESS_LOGGER_LABEL, "Requesting dry run text generation.")
        logger.debug(f"Request: {text_gen_request}")

        if self._horde_client_session is not None:
            dry_run_response = await self._horde_client_session.submit_request(
                text_gen_request,
                TextGenerateAsyncDryRunResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(dry_run_response, RequestErrorResponse):
            logger.error(f"Error response received: {dry_run_response.message}")
            raise AIHordeRequestError(dry_run_response)

        return dry_run_response

    async def workers_all_details(
        self,
        worker_name: str | None = None,
        *,
        api_key: str | None = None,
    ) -> AllWorkersDetailsResponse:
        """Get all the details for all workers.

        Returns:
            WorkersAllDetailsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                AllWorkersDetailsRequest(name=worker_name, apikey=api_key),
                AllWorkersDetailsResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def worker_details(
        self,
        worker_id: WorkerID | str,
        *,
        api_key: str | None = None,
    ) -> SingleWorkerDetailsResponse:
        """Get the details for a worker.

        Args:
            worker_id (WorkerID): The ID of the worker to get the details for.
            api_key (str, optional): The API key to use for the request.

        Returns:
            SingleWorkerDetailsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                SingleWorkerDetailsRequest(worker_id=worker_id, apikey=api_key),
                SingleWorkerDetailsResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def worker_modify(
        self,
        modify_worker_request: ModifyWorkerRequest,
    ) -> ModifyWorkerResponse:
        """Update a worker.

        Args:
            modify_worker_request (ModifyWorkerRequest): The request to update the worker.

        Returns:
            ModifyWorkerResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                modify_worker_request,
                ModifyWorkerResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def worker_delete(
        self,
        worker_id: WorkerID | str,
        *,
        api_key: str | None = None,
    ) -> DeleteWorkerResponse:
        """Delete a worker.

        Args:
            worker_id (WorkerID): The ID of the worker to delete.
            api_key (str, optional): The API key to use for the request.

        Returns:
            DeleteWorkerResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                DeleteWorkerRequest(worker_id=worker_id, apikey=api_key),
                DeleteWorkerResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def image_stats_totals(
        self,
    ) -> ImageStatsModelsTotalResponse:
        """Get the total stats for images.

        Returns:
            ImageStatsTotalsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                ImageStatsModelsTotalRequest(),
                ImageStatsModelsTotalResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def image_stats_models(
        self,
        model_state: str | MODEL_STATE = MODEL_STATE.known,
    ) -> ImageStatsModelsResponse:
        """Get the stats for images by model.

        Returns:
            ImageStatsModelsResponse: The response from the API.
        """
        model_state = MODEL_STATE(model_state)

        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                ImageStatsModelsRequest(model_state=model_state),
                ImageStatsModelsResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def text_stats_totals(
        self,
    ) -> TextStatsModelsTotalResponse:
        """Get the total stats for text.

        Returns:
            TextStatsTotalsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                TextStatsModelsTotalRequest(),
                TextStatsModelsTotalResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def text_stats_models(
        self,
    ) -> TextStatsModelResponse:
        """Get the stats for text by model.

        Returns:
            TextModelStatsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                TextStatsModelsRequest(),
                TextStatsModelResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def image_status_models_all(
        self,
    ) -> HordeStatusModelsAllResponse:
        """Get the status of all image models.

        Returns:
            ImageStatusModelsAllResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                HordeStatusModelsAllRequest(),
                HordeStatusModelsAllResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def image_status_models_single(
        self,
        model_name: str,
    ) -> HordeStatusModelsSingleResponse:
        """Get the status of a single image model.

        Args:
            model_name (str): The name of the model to get the status of.

        Returns:
            ImageStatusModelsSingleResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                HordeStatusModelsSingleRequest(model_name=model_name),
                HordeStatusModelsSingleResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

    async def get_news(
        self,
    ) -> NewsResponse:
        """Get the latest news from the AI-Horde API.

        Returns:
            NewsResponse: The response from the API.
        """
        if self._horde_client_session is not None:
            response = await self._horde_client_session.submit_request(
                NewsRequest(),
                NewsResponse,
            )
        else:
            raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

        if isinstance(response, RequestErrorResponse):
            raise AIHordeRequestError(response)

        return response

reasonable_minimum_timeout class-attribute instance-attribute

reasonable_minimum_timeout = 20

__init__

__init__(
    aiohttp_session: ClientSession | None = None,
    horde_client_session: (
        AIHordeAPIAsyncClientSession | None
    ) = None,
    apikey: str | None = None,
) -> None

Create a new instance of the AIHordeAPISimpleClient.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def __init__(
    self,
    aiohttp_session: aiohttp.ClientSession | None = None,
    horde_client_session: AIHordeAPIAsyncClientSession | None = None,
    apikey: str | None = None,
) -> None:
    """Create a new instance of the AIHordeAPISimpleClient."""
    super().__init__()

    if aiohttp_session is None and horde_client_session is None:
        raise RuntimeError("No aiohttp session provided but an async request was made.")

    if (
        aiohttp_session is not None
        and horde_client_session is not None
        and horde_client_session._aiohttp_session != aiohttp_session
    ):
        raise RuntimeError("The aiohttp session provided does not match the session in the client session.")

    if aiohttp_session is not None and horde_client_session is None:
        logger.info("Creating a new AIHordeAPIAsyncClientSession with the provided aiohttp session.")
        self._aiohttp_session = aiohttp_session
        self._horde_client_session = AIHordeAPIAsyncClientSession(aiohttp_session, apikey=apikey)
    elif horde_client_session is not None:
        self._horde_client_session = horde_client_session
        self._aiohttp_session = horde_client_session._aiohttp_session

download_image_from_generation async

download_image_from_generation(
    generation: ImageGeneration,
) -> tuple[PIL.Image.Image, GenerationID]

Asynchronously convert from base64 or download an image from a response.

Parameters:

• generation (ImageGeneration) –

  The image generation to convert.

Returns:

• tuple[Image, GenerationID] –

  PIL.Image.Image: The converted image.

Raises:

• ClientResponseError –

  If the generation couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def download_image_from_generation(
    self,
    generation: ImageGeneration,
) -> tuple[PIL.Image.Image, GenerationID]:
    """Asynchronously convert from base64 or download an image from a response.

    Args:
        generation (ImageGeneration): The image generation to convert.

    Returns:
        PIL.Image.Image: The converted image.

    Raises:
        ClientResponseError: If the generation couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    if generation.img is None:  # pragma: no cover
        raise ValueError("Generation has no image")

    if self._aiohttp_session is None:  # pragma: no cover
        raise RuntimeError("No aiohttp session provided but an async request was made.")

    image_bytes: bytes | None = None
    if urllib.parse.urlparse(generation.img).scheme in ["http", "https"]:
        async with self._aiohttp_session.get(generation.img, ssl=_default_sslcontext) as response:
            if response.status != 200:  # pragma: no cover
                logger.error(f"Error downloading image: {response.status}")
                response.raise_for_status()

            image_bytes = await response.read()
    else:
        try:
            image_bytes = base64.b64decode(generation.img)
        except Exception as e:
            logger.error(f"Error parsing image: {e}")
            raise e

    if image_bytes is None:  # pragma: no cover
        raise RuntimeError("Error downloading or parsing image")

    return (PIL.Image.open(io.BytesIO(image_bytes)), generation.id_)

download_image_from_url async

download_image_from_url(url: str) -> PIL.Image.Image

Asynchronously download an image from a URL.

Parameters:

• url (str) –

  The URL to download the image from.

Returns:

• Image –

  PIL.Image.Image: The downloaded image.

Raises:

• ClientResponseError –

  If the image couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def download_image_from_url(self, url: str) -> PIL.Image.Image:
    """Asynchronously download an image from a URL.

    Args:
        url (str): The URL to download the image from.

    Returns:
        PIL.Image.Image: The downloaded image.

    Raises:
        ClientResponseError: If the image couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    if self._aiohttp_session is None:
        raise RuntimeError("No aiohttp session provided but an async request was made.")

    async with self._aiohttp_session.get(url, ssl=_default_sslcontext) as response:
        if response.status != 200:  # pragma: no cover
            logger.error(f"Error downloading image: {response.status}")
            response.raise_for_status()

        image_bytes = await response.read()

    if image_bytes is None:  # pragma: no cover
        raise RuntimeError("Error downloading or parsing image")

    return PIL.Image.open(io.BytesIO(image_bytes))

heartbeat_request async

heartbeat_request() -> AIHordeHeartbeatResponse

Submit a heartbeat request to the AI-Horde API.

Returns:

• AIHordeHeartbeatResponse ( AIHordeHeartbeatResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def heartbeat_request(
    self,
) -> AIHordeHeartbeatResponse:
    """Submit a heartbeat request to the AI-Horde API.

    Returns:
        AIHordeHeartbeatResponse: The response from the API.
    """
    api_request = AIHordeHeartbeatRequest()

    if self._horde_client_session is not None:
        api_response = await self._horde_client_session.submit_request(
            api_request,
            api_request.get_default_success_response_type(),
        )

        if isinstance(api_response, RequestErrorResponse):
            raise AIHordeRequestError(api_response)

        return api_response

    raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

image_generate_request async

image_generate_request(
    image_gen_request: ImageGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[ImageGenerateCheckResponse], None] | None
    ) = None,
    delay: float = 0.0,
) -> tuple[ImageGenerateStatusResponse, GenerationID]

Submit an image generation request to the AI-Horde API, and wait for it to complete.

Be warned that using this method too frequently could trigger a rate limit from the AI-Horde API. Space concurrent requests apart slightly to allow them to be less than 10/second.

Parameters:

• image_gen_request (ImageGenerateAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely. Defaults to -1.

• check_callback (Callable[[ImageGenerateCheckResponse], None], default: None ) –

  A callback to call with the check response.

• delay (float, default: 0.0 ) –

  The number of seconds to wait before checking the status. Defaults to 0.0.

Returns:

• tuple[ImageGenerateStatusResponse, GenerationID] –

  tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

Raises:

• AIHordeRequestError –

  If the request failed. The error response is included in the exception.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_generate_request(
    self,
    image_gen_request: ImageGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
    check_callback: Callable[[ImageGenerateCheckResponse], None] | None = None,
    delay: float = 0.0,
) -> tuple[ImageGenerateStatusResponse, GenerationID]:
    """Submit an image generation request to the AI-Horde API, and wait for it to complete.

    *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
    Space concurrent requests apart slightly to allow them to be less than 10/second.

    Args:
        image_gen_request (ImageGenerateAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely.
            Defaults to -1.
        check_callback (Callable[[ImageGenerateCheckResponse], None], optional): A callback to call with the check
            response.
        delay (float, optional): The number of seconds to wait before checking the status. Defaults to 0.0.


    Returns:
        tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

    Raises:
        AIHordeRequestError: If the request failed. The error response is included in the exception.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    await asyncio.sleep(delay)

    timeout = self.validate_timeout(timeout, log_message=True)

    n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
    final_response, gen_id = await self._do_request_with_check(
        image_gen_request,
        number_of_responses=n,
        timeout=timeout,
        check_callback=generic_callback,
        check_callback_type=ImageGenerateCheckResponse,
    )

    if isinstance(final_response, RequestErrorResponse):  # pragma: no cover
        raise AIHordeRequestError(final_response)

    if not isinstance(final_response, ImageGenerateStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not an ImageGenerateStatusResponse")

    return (final_response, gen_id)

image_generate_request_dry_run async

image_generate_request_dry_run(
    image_gen_request: ImageGenerateAsyncRequest,
) -> ImageGenerateAsyncDryRunResponse

Submit a dry run image generation, which will return the kudos cost without actually generating images.

Parameters:

• image_gen_request (ImageGenerateAsyncRequest) –

  The request to submit.

Returns:

• ImageGenerateAsyncDryRunResponse ( ImageGenerateAsyncDryRunResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_generate_request_dry_run(
    self,
    image_gen_request: ImageGenerateAsyncRequest,
) -> ImageGenerateAsyncDryRunResponse:
    """Submit a dry run image generation, which will return the kudos cost without actually generating images.

    Args:
        image_gen_request (ImageGenerateAsyncRequest): The request to submit.

    Returns:
        ImageGenerateAsyncDryRunResponse: The response from the API.
    """
    if not image_gen_request.dry_run:
        raise RuntimeError("Dry run request must have dry_run set to True")

    n = image_gen_request.params.n if image_gen_request.params and image_gen_request.params.n else 1
    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting dry run for {n} images.")

    if self._horde_client_session is not None:
        dry_run_response = await self._horde_client_session.submit_request(
            image_gen_request,
            ImageGenerateAsyncDryRunResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(dry_run_response, RequestErrorResponse):
        logger.error(f"Error response received: {dry_run_response.message}")
        raise AIHordeRequestError(dry_run_response)

    return dry_run_response

alchemy_request async

alchemy_request(
    alchemy_request: AlchemyAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[AlchemyStatusResponse], None] | None
    ) = None,
) -> tuple[AlchemyStatusResponse, GenerationID]

Submit an alchemy request to the AI-Horde API, and wait for it to complete.

Be warned that using this method too frequently could trigger a rate limit from the AI-Horde API. Space concurrent requests apart slightly to allow them to be less than 10/second.

Parameters:

• alchemy_request (AlchemyAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Defaults to -1.

• check_callback (Callable[[AlchemyStatusResponse], None], default: None ) –

  A callback to call with the check response.

Returns:

• tuple[AlchemyStatusResponse, GenerationID] –

  tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

Raises:

• AIHordeRequestError –

  If the request failed. The error response is included in the exception.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def alchemy_request(
    self,
    alchemy_request: AlchemyAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
    check_callback: Callable[[AlchemyStatusResponse], None] | None = None,
) -> tuple[AlchemyStatusResponse, GenerationID]:
    """Submit an alchemy request to the AI-Horde API, and wait for it to complete.

    *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
    Space concurrent requests apart slightly to allow them to be less than 10/second.

    Args:
        alchemy_request (AlchemyAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Defaults to -1.
        check_callback (Callable[[AlchemyStatusResponse], None], optional): A callback to call with the check
            response.

    Returns:
        tuple[ImageGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

    Raises:
        AIHordeRequestError: If the request failed. The error response is included in the exception.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    timeout = self.validate_timeout(timeout, log_message=True)

    response, gen_id = await self._do_request_with_check(
        alchemy_request,
        number_of_responses=len(alchemy_request.forms),
        timeout=timeout,
        check_callback=generic_callback,
        check_callback_type=AlchemyStatusResponse,
    )
    if isinstance(response, RequestErrorResponse):  # pragma: no cover
        raise AIHordeRequestError(response)

    if not isinstance(response, AlchemyStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not an AlchemyAsyncResponse")

    return (response, gen_id)

text_generate_request async

text_generate_request(
    text_gen_request: TextGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,
    check_callback: (
        Callable[[TextGenerateStatusResponse], None] | None
    ) = None,
    delay: float = 0.0,
) -> tuple[TextGenerateStatusResponse, GenerationID]

Submit a text generation request to the AI-Horde API, and wait for it to complete.

Be warned that using this method too frequently could trigger a rate limit from the AI-Horde API. Space concurrent requests apart slightly to allow them to be less than 10/second.

Parameters:

• text_gen_request (TextGenerateAsyncRequest) –

  The request to submit.

• timeout (int, default: GENERATION_MAX_LIFE ) –

  The number of seconds to wait before aborting. returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely. Defaults to -1.

• check_callback (Callable[[TextGenerateStatusResponse], None], default: None ) –

  A callback to call with the check response.

• delay (float, default: 0.0 ) –

  The number of seconds to wait before checking the status. Defaults to 0.0.

Returns:

• tuple[TextGenerateStatusResponse, GenerationID] –

  tuple[TextGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

Raises:

• AIHordeRequestError –

  If the request failed. The error response is included in the exception.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def text_generate_request(
    self,
    text_gen_request: TextGenerateAsyncRequest,
    timeout: int = GENERATION_MAX_LIFE,  # noqa: ASYNC109 # FIXME
    check_callback: Callable[[TextGenerateStatusResponse], None] | None = None,
    delay: float = 0.0,
) -> tuple[TextGenerateStatusResponse, GenerationID]:
    """Submit a text generation request to the AI-Horde API, and wait for it to complete.

    *Be warned* that using this method too frequently could trigger a rate limit from the AI-Horde API.
    Space concurrent requests apart slightly to allow them to be less than 10/second.

    Args:
        text_gen_request (TextGenerateAsyncRequest): The request to submit.
        timeout (int, optional): The number of seconds to wait before aborting.
            returns any completed images at the end of the timeout. Any value 0 or less will wait indefinitely.
            Defaults to -1.
        check_callback (Callable[[TextGenerateStatusResponse], None], optional): A callback to call with the check
            response.
        delay (float, optional): The number of seconds to wait before checking the status. Defaults to 0.0.

    Returns:
        tuple[TextGenerateStatusResponse, GenerationID]: The final status response and the corresponding job ID.

    Raises:
        AIHordeRequestError: If the request failed. The error response is included in the exception.
    """
    # `cast()` returns the value unchanged but tells coerces the type for mypy's benefit
    # Static type checkers can't see that `_do_request_with_check` is reliably passing an object of the correct
    # type, but we are guaranteed that it is due to the `ImageGenerateCheckResponse` type being passed as an arg.
    generic_callback = cast(Callable[[HordeResponse], None], check_callback)

    await asyncio.sleep(delay)

    timeout = self.validate_timeout(timeout, log_message=True)

    num_gens_requested = 1

    if text_gen_request.params and text_gen_request.params.n:
        num_gens_requested = text_gen_request.params.n

    logger.log(PROGRESS_LOGGER_LABEL, f"Requesting {num_gens_requested} text generation.")
    logger.debug(f"Request: {text_gen_request}")

    response, gen_id = await self._do_request_with_check(
        text_gen_request,
        number_of_responses=1,
        timeout=timeout,
        check_callback=generic_callback,
        check_callback_type=TextGenerateStatusResponse,
    )

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    if not isinstance(response, TextGenerateStatusResponse):  # pragma: no cover
        raise RuntimeError("Response was not a TextGenerateStatusResponse")

    return (response, gen_id)

text_generate_request_dry_run async

text_generate_request_dry_run(
    text_gen_request: TextGenerateAsyncRequest,
) -> TextGenerateAsyncDryRunResponse

Submit a dry run text generation, which will return the kudos cost without actually generating text.

Parameters:

• text_gen_request (TextGenerateAsyncRequest) –

  The request to submit.

Returns:

• TextGenerateAsyncDryRunResponse ( TextGenerateAsyncDryRunResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def text_generate_request_dry_run(
    self,
    text_gen_request: TextGenerateAsyncRequest,
) -> TextGenerateAsyncDryRunResponse:
    """Submit a dry run text generation, which will return the kudos cost without actually generating text.

    Args:
        text_gen_request (TextGenerateAsyncRequest): The request to submit.

    Returns:
        TextGenerateAsyncDryRunResponse: The response from the API.
    """
    if not text_gen_request.dry_run:
        raise RuntimeError("Dry run request must have dry_run set to True")

    logger.log(PROGRESS_LOGGER_LABEL, "Requesting dry run text generation.")
    logger.debug(f"Request: {text_gen_request}")

    if self._horde_client_session is not None:
        dry_run_response = await self._horde_client_session.submit_request(
            text_gen_request,
            TextGenerateAsyncDryRunResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(dry_run_response, RequestErrorResponse):
        logger.error(f"Error response received: {dry_run_response.message}")
        raise AIHordeRequestError(dry_run_response)

    return dry_run_response

workers_all_details async

workers_all_details(
    worker_name: str | None = None,
    *,
    api_key: str | None = None
) -> AllWorkersDetailsResponse

Get all the details for all workers.

Returns:

• WorkersAllDetailsResponse ( AllWorkersDetailsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def workers_all_details(
    self,
    worker_name: str | None = None,
    *,
    api_key: str | None = None,
) -> AllWorkersDetailsResponse:
    """Get all the details for all workers.

    Returns:
        WorkersAllDetailsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            AllWorkersDetailsRequest(name=worker_name, apikey=api_key),
            AllWorkersDetailsResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

worker_details async

worker_details(
    worker_id: WorkerID | str, *, api_key: str | None = None
) -> SingleWorkerDetailsResponse

Get the details for a worker.

Parameters:

• worker_id (WorkerID) –

  The ID of the worker to get the details for.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• SingleWorkerDetailsResponse ( SingleWorkerDetailsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def worker_details(
    self,
    worker_id: WorkerID | str,
    *,
    api_key: str | None = None,
) -> SingleWorkerDetailsResponse:
    """Get the details for a worker.

    Args:
        worker_id (WorkerID): The ID of the worker to get the details for.
        api_key (str, optional): The API key to use for the request.

    Returns:
        SingleWorkerDetailsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            SingleWorkerDetailsRequest(worker_id=worker_id, apikey=api_key),
            SingleWorkerDetailsResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

worker_modify async

worker_modify(
    modify_worker_request: ModifyWorkerRequest,
) -> ModifyWorkerResponse

Update a worker.

Parameters:

• modify_worker_request (ModifyWorkerRequest) –

  The request to update the worker.

Returns:

• ModifyWorkerResponse ( ModifyWorkerResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def worker_modify(
    self,
    modify_worker_request: ModifyWorkerRequest,
) -> ModifyWorkerResponse:
    """Update a worker.

    Args:
        modify_worker_request (ModifyWorkerRequest): The request to update the worker.

    Returns:
        ModifyWorkerResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            modify_worker_request,
            ModifyWorkerResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

worker_delete async

worker_delete(
    worker_id: WorkerID | str, *, api_key: str | None = None
) -> DeleteWorkerResponse

Delete a worker.

Parameters:

• worker_id (WorkerID) –

  The ID of the worker to delete.

• api_key (str, default: None ) –

  The API key to use for the request.

Returns:

• DeleteWorkerResponse ( DeleteWorkerResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def worker_delete(
    self,
    worker_id: WorkerID | str,
    *,
    api_key: str | None = None,
) -> DeleteWorkerResponse:
    """Delete a worker.

    Args:
        worker_id (WorkerID): The ID of the worker to delete.
        api_key (str, optional): The API key to use for the request.

    Returns:
        DeleteWorkerResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            DeleteWorkerRequest(worker_id=worker_id, apikey=api_key),
            DeleteWorkerResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

image_stats_totals async

image_stats_totals() -> ImageStatsModelsTotalResponse

Get the total stats for images.

Returns:

• ImageStatsTotalsResponse ( ImageStatsModelsTotalResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_stats_totals(
    self,
) -> ImageStatsModelsTotalResponse:
    """Get the total stats for images.

    Returns:
        ImageStatsTotalsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            ImageStatsModelsTotalRequest(),
            ImageStatsModelsTotalResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

image_stats_models async

image_stats_models(
    model_state: str | MODEL_STATE = MODEL_STATE.known,
) -> ImageStatsModelsResponse

Get the stats for images by model.

Returns:

• ImageStatsModelsResponse ( ImageStatsModelsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_stats_models(
    self,
    model_state: str | MODEL_STATE = MODEL_STATE.known,
) -> ImageStatsModelsResponse:
    """Get the stats for images by model.

    Returns:
        ImageStatsModelsResponse: The response from the API.
    """
    model_state = MODEL_STATE(model_state)

    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            ImageStatsModelsRequest(model_state=model_state),
            ImageStatsModelsResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

text_stats_totals async

text_stats_totals() -> TextStatsModelsTotalResponse

Get the total stats for text.

Returns:

• TextStatsTotalsResponse ( TextStatsModelsTotalResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def text_stats_totals(
    self,
) -> TextStatsModelsTotalResponse:
    """Get the total stats for text.

    Returns:
        TextStatsTotalsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            TextStatsModelsTotalRequest(),
            TextStatsModelsTotalResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

text_stats_models async

text_stats_models() -> TextStatsModelResponse

Get the stats for text by model.

Returns:

• TextModelStatsResponse ( TextStatsModelResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def text_stats_models(
    self,
) -> TextStatsModelResponse:
    """Get the stats for text by model.

    Returns:
        TextModelStatsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            TextStatsModelsRequest(),
            TextStatsModelResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

image_status_models_all async

image_status_models_all() -> HordeStatusModelsAllResponse

Get the status of all image models.

Returns:

• ImageStatusModelsAllResponse ( HordeStatusModelsAllResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_status_models_all(
    self,
) -> HordeStatusModelsAllResponse:
    """Get the status of all image models.

    Returns:
        ImageStatusModelsAllResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            HordeStatusModelsAllRequest(),
            HordeStatusModelsAllResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

image_status_models_single async

image_status_models_single(
    model_name: str,
) -> HordeStatusModelsSingleResponse

Get the status of a single image model.

Parameters:

• model_name (str) –

  The name of the model to get the status of.

Returns:

• ImageStatusModelsSingleResponse ( HordeStatusModelsSingleResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def image_status_models_single(
    self,
    model_name: str,
) -> HordeStatusModelsSingleResponse:
    """Get the status of a single image model.

    Args:
        model_name (str): The name of the model to get the status of.

    Returns:
        ImageStatusModelsSingleResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            HordeStatusModelsSingleRequest(model_name=model_name),
            HordeStatusModelsSingleResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

get_news async

get_news() -> NewsResponse

Get the latest news from the AI-Horde API.

Returns:

• NewsResponse ( NewsResponse ) –

  The response from the API.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

async def get_news(
    self,
) -> NewsResponse:
    """Get the latest news from the AI-Horde API.

    Returns:
        NewsResponse: The response from the API.
    """
    if self._horde_client_session is not None:
        response = await self._horde_client_session.submit_request(
            NewsRequest(),
            NewsResponse,
        )
    else:
        raise RuntimeError("No AIHordeAPIAsyncClientSession provided")

    if isinstance(response, RequestErrorResponse):
        raise AIHordeRequestError(response)

    return response

validate_timeout

validate_timeout(
    timeout: int, log_message: bool = False
) -> int

Check if a timeout is reasonable.

Parameters:

• timeout (int) –

  The timeout to check.

• log_message (bool, default: False ) –

  Whether to log a message if the timeout is too short. Defaults to False.

Returns:

• bool ( int ) –

  True if the timeout is reasonable, False otherwise.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def validate_timeout(
    self,
    timeout: int,
    log_message: bool = False,
) -> int:
    """Check if a timeout is reasonable.

    Args:
        timeout (int): The timeout to check.
        log_message (bool, optional): Whether to log a message if the timeout is too short. Defaults to False.

    Returns:
        bool: True if the timeout is reasonable, False otherwise.
    """
    if timeout <= 0:  # pragma: no cover
        logger.debug(f"No timeout set. Using default timeout of {GENERATION_MAX_LIFE} seconds.")
        return GENERATION_MAX_LIFE

    if timeout > GENERATION_MAX_LIFE:  # pragma: no cover
        logger.warning(
            f"Timeout ({timeout}) is greater than the maximum possible of {GENERATION_MAX_LIFE} seconds."
            f"Using {GENERATION_MAX_LIFE} for the timeout.",
        )
        return GENERATION_MAX_LIFE

    if timeout < self.reasonable_minimum_timeout and log_message:  # pragma: no cover
        logger.warning(
            f"test_simple_client_async_image_generate_multiple than {self.reasonable_minimum_timeout} seconds, "
            "this is probably too short.",
        )

    return timeout

download_image_bytes

download_image_bytes(url: str) -> io.BytesIO

Download an image from a URL.

Parameters:

• url (str) –

  The URL to download the image from.

Returns:

• BytesIO –

  PIL.Image.Image: The downloaded image.

Raises:

• ClientResponseError –

  If the image couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def download_image_bytes(url: str) -> io.BytesIO:
    """Download an image from a URL.

    Args:
        url (str): The URL to download the image from.

    Returns:
        PIL.Image.Image: The downloaded image.

    Raises:
        ClientResponseError: If the image couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    response = requests.get(url)
    if response.status_code != 200:
        logger.error(f"Error downloading image: {response.status_code}")
        response.raise_for_status()

    logger.debug(f"Downloaded image: {url}")
    return io.BytesIO(response.content)

download_image_from_url

download_image_from_url(url: str) -> PIL.Image.Image

Download an image from a URL.

Parameters:

• url (str) –

  The URL to download the image from.

Returns:

• Image –

  PIL.Image.Image: The downloaded image.

Raises:

• ClientResponseError –

  If the image couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def download_image_from_url(url: str) -> PIL.Image.Image:
    """Download an image from a URL.

    Args:
        url (str): The URL to download the image from.

    Returns:
        PIL.Image.Image: The downloaded image.

    Raises:
        ClientResponseError: If the image couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    return PIL.Image.open(download_image_bytes(url))

download_image_from_generation

download_image_from_generation(
    generation: ImageGeneration,
) -> PIL.Image.Image

Fetch and parse an image from a response.

Parameters:

• generation (ImageGeneration) –

  The image generation to convert.

Returns:

• Image –

  PIL.Image.Image: The converted image.

Raises:

• ClientResponseError –

  If the generation couldn't be downloaded.

• Error –

  If the image couldn't be parsed from base 64.

• RuntimeError –

  If the image couldn't be downloaded or parsed for any other reason.

Source code in horde_sdk/ai_horde_api/ai_horde_clients.py

def download_image_from_generation(generation: ImageGeneration) -> PIL.Image.Image:
    """Fetch and parse an image from a response.

    Args:
        generation (ImageGeneration): The image generation to convert.

    Returns:
        PIL.Image.Image: The converted image.

    Raises:
        ClientResponseError: If the generation couldn't be downloaded.
        binascii.Error: If the image couldn't be parsed from base 64.
        RuntimeError: If the image couldn't be downloaded or parsed for any other reason.

    """
    if generation.img is None:
        raise ValueError("Generation has no image")

    image_bytes: bytes | None = None
    if urllib.parse.urlparse(generation.img).scheme in ["http", "https"]:
        image_bytes = download_image_bytes(generation.img).read()
    else:
        image_bytes = base64.b64decode(generation.img)

    if image_bytes is None:
        raise RuntimeError("Error downloading or parsing image")

    return PIL.Image.open(io.BytesIO(image_bytes))