Skip to content

async_

ImageGenerateAsyncResponse

Bases: HordeResponseBaseModel, JobResponseMixin, ResponseRequiringFollowUpMixin, ContainsMessageResponseMixin

The expected cost of the requested job and any warnings generated by the server.

A typical warning is that the request is not currently possible. You may want to handle certain warnings in your application. See :class:horde_sdk.ai_horde_api.apimodels.base.SingleWarningEntry for more information.

Represents the data returned from the /v2/generate/async endpoint with http status code 202.

v2 API Model: RequestAsync

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
class ImageGenerateAsyncResponse(
    HordeResponseBaseModel,
    JobResponseMixin,
    ResponseRequiringFollowUpMixin,
    ContainsMessageResponseMixin,
):
    """The expected cost of the requested job and any warnings generated by the server.

    A typical warning is that the request is not currently possible. You may want to handle certain
    warnings in your application. See :class:`horde_sdk.ai_horde_api.apimodels.base.SingleWarningEntry` for more
    information.

    Represents the data returned from the /v2/generate/async endpoint with http status code 202.

    v2 API Model: `RequestAsync`
    """

    """The UUID for this image generation."""
    kudos: float
    """The expected kudos consumption for this request."""
    warnings: list[SingleWarningEntry] | None = None
    """Any warnings that were generated by the server or a serving worker."""

    @model_validator(mode="after")
    def validate_warnings(self) -> ImageGenerateAsyncResponse:
        """Log any warnings that were generated by the server or a serving worker."""
        if self.warnings is None:
            return self

        for warning in self.warnings:
            logger.warning(f"Warning from server ({warning.code}): {warning.message}")

        return self

    @override
    def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, object]]:
        if as_python_field_name:
            return [{"id_": self.id_}]
        return [{"id": self.id_}]

    @override
    @classmethod
    def get_follow_up_default_request_type(cls) -> type[ImageGenerateCheckRequest]:
        return ImageGenerateCheckRequest

    @override
    @classmethod
    def get_follow_up_request_types(  # type: ignore[override]
        cls,
    ) -> list[type[ImageGenerateCheckRequest | ImageGenerateStatusRequest]]:
        return [ImageGenerateCheckRequest, ImageGenerateStatusRequest]

    @override
    @classmethod
    def get_follow_up_failure_cleanup_request_type(cls) -> type[DeleteImageGenerateRequest]:
        return DeleteImageGenerateRequest

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "RequestAsync"

    def __hash__(self) -> int:
        return hash(ImageGenerateAsyncResponse.__name__) + hash(self.id_)

    def __eq__(self, __value: object) -> bool:
        return isinstance(__value, ImageGenerateAsyncResponse) and self.id_ == __value.id_

kudos instance-attribute

kudos: float

The expected kudos consumption for this request.

warnings class-attribute instance-attribute

warnings: list[SingleWarningEntry] | None = None

Any warnings that were generated by the server or a serving worker.

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

message class-attribute instance-attribute

message: str = ''

A message from the API. This is typically an error or warning message, but may also be informational.

id_ class-attribute instance-attribute

id_: GenerationID = Field(alias='id')

The UUID for this job.

time_constructed property

time_constructed: float

The time the model was constructed (in epoch time).

validate_warnings

validate_warnings() -> ImageGenerateAsyncResponse

Log any warnings that were generated by the server or a serving worker.

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@model_validator(mode="after")
def validate_warnings(self) -> ImageGenerateAsyncResponse:
    """Log any warnings that were generated by the server or a serving worker."""
    if self.warnings is None:
        return self

    for warning in self.warnings:
        logger.warning(f"Warning from server ({warning.code}): {warning.message}")

    return self

get_follow_up_returned_params

get_follow_up_returned_params(
    *, as_python_field_name: bool = False
) -> list[dict[str, object]]
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, object]]:
    if as_python_field_name:
        return [{"id_": self.id_}]
    return [{"id": self.id_}]

get_follow_up_default_request_type classmethod

get_follow_up_default_request_type() -> (
    type[ImageGenerateCheckRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_follow_up_default_request_type(cls) -> type[ImageGenerateCheckRequest]:
    return ImageGenerateCheckRequest

get_follow_up_request_types classmethod

get_follow_up_request_types() -> (
    list[
        type[
            ImageGenerateCheckRequest
            | ImageGenerateStatusRequest
        ]
    ]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_follow_up_request_types(  # type: ignore[override]
    cls,
) -> list[type[ImageGenerateCheckRequest | ImageGenerateStatusRequest]]:
    return [ImageGenerateCheckRequest, ImageGenerateStatusRequest]

get_follow_up_failure_cleanup_request_type classmethod

get_follow_up_failure_cleanup_request_type() -> (
    type[DeleteImageGenerateRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_follow_up_failure_cleanup_request_type(cls) -> type[DeleteImageGenerateRequest]:
    return DeleteImageGenerateRequest

get_api_model_name classmethod

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "RequestAsync"

__hash__

__hash__() -> int
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
def __hash__(self) -> int:
    return hash(ImageGenerateAsyncResponse.__name__) + hash(self.id_)

__eq__

__eq__(__value: object) -> bool
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
def __eq__(self, __value: object) -> bool:
    return isinstance(__value, ImageGenerateAsyncResponse) and self.id_ == __value.id_

get_follow_up_extra_params

get_follow_up_extra_params() -> dict[str, object]

Return any additional information required from this response to submit a follow up request.

Source code in horde_sdk/generic_api/apimodels.py
def get_follow_up_extra_params(self) -> dict[str, object]:
    """Return any additional information required from this response to submit a follow up request."""
    logger.warning("This method may be deprecated in the future.")
    return {}  # TODO: Would extra params need to come into play for a list of follow up requests?

get_follow_up_all_params

get_follow_up_all_params() -> list[dict[str, object]]

Return the required inf from this response to submit any follow up requests warranted from this response.

Note that this dict uses the alias field names (as seen on the API), not the python field names.

get_follow_up_failure_cleanup_params is not included.

This is used for context management.

Returns:

  • list[dict[str, object]]

    list[dict[str, object]]: A list of dicts of parameter names and values for each follow up request.

Source code in horde_sdk/generic_api/apimodels.py
def get_follow_up_all_params(self) -> list[dict[str, object]]:
    """Return the required inf from this response to submit any follow up requests warranted from this response.

    Note that this dict uses the alias field names (as seen on the API), not the python field names.

    `get_follow_up_failure_cleanup_params` is **not** included.

    This is used for context management.

    Returns:
        list[dict[str, object]]: A list of dicts of parameter names and values for each follow up request.
    """
    follow_up_params = self.get_follow_up_returned_params()

    if isinstance(follow_up_params, list):
        return follow_up_params  # FIXME: Would extra params need to come into play?

    return [{**follow_up_params, **self.get_follow_up_extra_params()}]

get_follow_up_failure_cleanup_params

get_follow_up_failure_cleanup_params() -> dict[str, object]

Return any extra information required from this response to clean up after a failed follow up request.

Note that this dict uses the alias field names (as seen on the API), not the python field names.

This is used for context management.

Source code in horde_sdk/generic_api/apimodels.py
def get_follow_up_failure_cleanup_params(self) -> dict[str, object]:
    """Return any extra information required from this response to clean up after a failed follow up request.

    Note that this dict uses the alias field names (as seen on the API), not the python field names.

    This is used for context management.
    """
    return {}

get_follow_up_failure_cleanup_request

get_follow_up_failure_cleanup_request() -> (
    list[HordeRequest]
)

Return the request for this response to clean up after a failed follow up request.

Source code in horde_sdk/generic_api/apimodels.py
def get_follow_up_failure_cleanup_request(self) -> list[HordeRequest]:
    """Return the request for this response to clean up after a failed follow up request."""
    if self.ignore_failure():
        return []

    if hasattr(self, "_cleanup_requests") and self._cleanup_requests is not None:
        return self._cleanup_requests

    cleanup_request_type = self.get_follow_up_failure_cleanup_request_type()
    if not cleanup_request_type:
        raise ValueError("No cleanup request type defined")

    self._cleanup_requests = []

    all_cleanup_params: list[dict[str, object]] = self.get_follow_up_all_params()
    for cleanup_params in all_cleanup_params:
        cleanup_params.update(self.get_follow_up_failure_cleanup_params())
        self._cleanup_requests.append(cleanup_request_type.model_validate(cleanup_params))

    return self._cleanup_requests

ignore_failure

ignore_failure() -> bool

Return if the object is in a state which doesn't require failure follow up.

Source code in horde_sdk/generic_api/apimodels.py
def ignore_failure(self) -> bool:
    """Return if the object is in a state which doesn't require failure follow up."""
    # ImageGenerateJobPopResponse was the use case at the time of writing
    return False

does_target_request_follow_up

does_target_request_follow_up(
    target_request: HordeRequest,
) -> bool

Return whether the target_request would follow up on this request.

Parameters:

  • target_request (HordeRequest) –

    The request to check if it would follow up on this request.

Returns:

  • bool ( bool ) –

    Whether the target_request would follow up on this request.

Source code in horde_sdk/generic_api/apimodels.py
def does_target_request_follow_up(self, target_request: HordeRequest) -> bool:
    """Return whether the `target_request` would follow up on this request.

    Args:
        target_request (HordeRequest): The request to check if it would follow up on this request.

    Returns:
        bool: Whether the `target_request` would follow up on this request.
    """
    follow_up_returned_params = self.get_follow_up_returned_params(as_python_field_name=True)

    if len(follow_up_returned_params) == 0:
        logger.warning("No follow up returned params defined for this request")
        return False
    all_match = True
    for param_set in follow_up_returned_params:
        for key, value in param_set.items():
            if hasattr(target_request, key) and getattr(target_request, key) != value:
                all_match = False
                break
    return all_match

validate_id

validate_id(v: str | GenerationID) -> GenerationID | str

Ensure that the job ID is not empty.

Source code in horde_sdk/ai_horde_api/apimodels/base.py
@field_validator("id_", mode="before")
def validate_id(cls, v: str | GenerationID) -> GenerationID | str:
    """Ensure that the job ID is not empty."""
    if isinstance(v, str) and v == "":
        logger.warning("Job ID is empty")
        return GenerationID(root=uuid.uuid4())

    return v

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

Return a set of fields which should be redacted from logs.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    """Return a set of fields which should be redacted from logs."""
    return {"apikey"}

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

Return an additional set of fields to exclude from the log_safe_model_dump method.

Source code in horde_sdk/generic_api/apimodels.py
def get_extra_fields_to_exclude_from_log(self) -> set[str]:
    """Return an additional set of fields to exclude from the log_safe_model_dump method."""
    return set()

log_safe_model_dump

log_safe_model_dump(
    extra_exclude: set[str] | None = None,
) -> dict[Any, Any]

Return a dict of the model's fields, with any sensitive fields redacted.

Source code in horde_sdk/generic_api/apimodels.py
def log_safe_model_dump(self, extra_exclude: set[str] | None = None) -> dict[Any, Any]:
    """Return a dict of the model's fields, with any sensitive fields redacted."""
    if extra_exclude is None:
        extra_exclude = set()

    if hasattr(self, "model_dump"):
        return self.model_dump(  # type: ignore
            exclude=self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude,
        )

    logger.warning("Model does not have a model_dump method. Using python native class compatible method.")
    logger.debug(
        "Generally this should not be relied upon. If you're seeing this and you're a developer for the SDK, "
        "consider using pydantic models instead.",
    )
    # Its not a pydantic model, use python native class compatible method
    return {
        key: getattr(self, key)
        for key in self.__dict__
        if key not in self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude
    }

ImageGenerateAsyncDryRunResponse

Bases: HordeResponseBaseModel

Request the expected cost of the job without actually starting the job.

Represents the data returned from the /v2/generate/async endpoint with http status code 200.

v2 API Model: _ANONYMOUS_MODEL

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
class ImageGenerateAsyncDryRunResponse(HordeResponseBaseModel):
    """Request the expected cost of the job without actually starting the job.

    Represents the data returned from the /v2/generate/async endpoint with http status code 200.

    v2 API Model: `_ANONYMOUS_MODEL`
    """

    kudos: float
    """The expected kudos consumption for this request."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return _ANONYMOUS_MODEL

kudos instance-attribute

kudos: float

The expected kudos consumption for this request.

time_constructed property

time_constructed: float

The time the model was constructed (in epoch time).

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

get_api_model_name classmethod

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return _ANONYMOUS_MODEL

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

Return a set of fields which should be redacted from logs.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    """Return a set of fields which should be redacted from logs."""
    return {"apikey"}

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

Return an additional set of fields to exclude from the log_safe_model_dump method.

Source code in horde_sdk/generic_api/apimodels.py
def get_extra_fields_to_exclude_from_log(self) -> set[str]:
    """Return an additional set of fields to exclude from the log_safe_model_dump method."""
    return set()

log_safe_model_dump

log_safe_model_dump(
    extra_exclude: set[str] | None = None,
) -> dict[Any, Any]

Return a dict of the model's fields, with any sensitive fields redacted.

Source code in horde_sdk/generic_api/apimodels.py
def log_safe_model_dump(self, extra_exclude: set[str] | None = None) -> dict[Any, Any]:
    """Return a dict of the model's fields, with any sensitive fields redacted."""
    if extra_exclude is None:
        extra_exclude = set()

    if hasattr(self, "model_dump"):
        return self.model_dump(  # type: ignore
            exclude=self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude,
        )

    logger.warning("Model does not have a model_dump method. Using python native class compatible method.")
    logger.debug(
        "Generally this should not be relied upon. If you're seeing this and you're a developer for the SDK, "
        "consider using pydantic models instead.",
    )
    # Its not a pydantic model, use python native class compatible method
    return {
        key: getattr(self, key)
        for key in self.__dict__
        if key not in self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude
    }

ImageGenerationInputPayload

Bases: ImageGenerateParamMixin

Represents the 'params' field in the /v2/generate/async endpoint.

v2 API Model: ModelGenerationInputStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
class ImageGenerationInputPayload(ImageGenerateParamMixin):
    """Represents the 'params' field in the `/v2/generate/async` endpoint.

    v2 API Model: `ModelGenerationInputStable`
    """

    steps: int = Field(default=25, ge=1, validation_alias=AliasChoices("steps", "ddim_steps"))
    """The number of image generation steps to perform."""

    n: int = Field(default=1, ge=1, le=20, validation_alias=AliasChoices("n", "n_iter"))
    """The number of images to generate. Defaults to 1, maximum is 20."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ModelGenerationInputStable"

    @field_validator("n", mode="before")
    def validate_n(cls, value: int) -> int:
        """Ensure that n is at least 1."""
        if value == 0:
            logger.debug("n (number of images to generate) is not set; defaulting to 1")
            return 1

        return value

steps class-attribute instance-attribute

steps: int = Field(
    default=25,
    ge=1,
    validation_alias=AliasChoices("steps", "ddim_steps"),
)

The number of image generation steps to perform.

n class-attribute instance-attribute

n: int = Field(
    default=1,
    ge=1,
    le=20,
    validation_alias=AliasChoices("n", "n_iter"),
)

The number of images to generate. Defaults to 1, maximum is 20.

model_config class-attribute instance-attribute

model_config = (
    ConfigDict(frozen=True, extra="allow")
    if not getenv("TESTS_ONGOING")
    else ConfigDict(frozen=True, extra="forbid")
)

height class-attribute instance-attribute

height: int = Field(
    default=512, ge=64, le=3072, multiple_of=64
)

The desired output image height.

width class-attribute instance-attribute

width: int = Field(
    default=512, ge=64, le=3072, multiple_of=64
)

The desired output image width.

sampler_name class-attribute instance-attribute

sampler_name: KNOWN_IMAGE_SAMPLERS | str = k_euler

The sampler to use for this generation. Defaults to KNOWN_IMAGE_SAMPLERS.k_lms.

karras class-attribute instance-attribute

karras: bool = True

Set to True if you want to use the Karras scheduling.

cfg_scale class-attribute instance-attribute

cfg_scale: float = Field(default=7.5, ge=0, le=10)

The cfg_scale to use for this generation. Defaults to 7.5.

denoising_strength class-attribute instance-attribute

denoising_strength: float | None = Field(
    default=1, ge=0, le=1
)

The denoising strength to use for this generation. Defaults to 1.

clip_skip class-attribute instance-attribute

clip_skip: int = Field(default=1, ge=1, le=12)

The number of clip layers to skip.

post_processing class-attribute instance-attribute

post_processing: list[
    str
    | KNOWN_UPSCALERS
    | KNOWN_FACEFIXERS
    | KNOWN_MISC_POST_PROCESSORS
] = Field(default_factory=list)

A list of post-processing models to use.

post_processing_order class-attribute instance-attribute

post_processing_order: POST_PROCESSOR_ORDER_TYPE = (
    facefixers_first
)

The order in which to apply post-processing models. Applying upscalers or removing backgrounds before facefixers costs less kudos.

facefixer_strength class-attribute instance-attribute

facefixer_strength: float | None = Field(
    default=None, ge=0, le=1
)

The strength of the facefixer model.

hires_fix class-attribute instance-attribute

hires_fix: bool = False

Set to True if you want to use the hires fix.

hires_fix_denoising_strength class-attribute instance-attribute

hires_fix_denoising_strength: float | None = Field(
    default=None, ge=0, le=1
)

The strength of the denoising for the hires fix second pass.

loras class-attribute instance-attribute

loras: list[LorasPayloadEntry] | None = None

A list of lora parameters to use.

tis class-attribute instance-attribute

tis: list[TIPayloadEntry] | None = None

A list of textual inversion (embedding) parameters to use.

workflow class-attribute instance-attribute

workflow: str | KNOWN_IMAGE_WORKFLOWS | None = None

The specific comfyUI workflow to use.

transparent class-attribute instance-attribute

transparent: bool | None = None

When true, will generate an image with a transparent background

tiling class-attribute instance-attribute

tiling: bool = False

Set to True if you want to use seamless tiling.

special class-attribute instance-attribute

special: dict[Any, Any] | None = None

Reserved for future use.

seed class-attribute instance-attribute

seed: str | None = None

The seed to use for this generation. If not provided, a random seed will be used.

seed_variation class-attribute instance-attribute

seed_variation: int | None = Field(
    default=None, ge=1, le=1000
)

Deprecated.

control_type class-attribute instance-attribute

control_type: str | KNOWN_IMAGE_CONTROLNETS | None = None

The type of control net type to use.

image_is_control class-attribute instance-attribute

image_is_control: bool | None = None

Set to True if the image is a control image.

return_control_map class-attribute instance-attribute

return_control_map: bool | None = None

Set to True if you want the ControlNet map returned instead of a generated image.

extra_texts class-attribute instance-attribute

extra_texts: list[ExtraTextEntry] | None = None

A list of extra texts and prompts to use in the comfyUI workflow.

use_nsfw_censor class-attribute instance-attribute

use_nsfw_censor: bool = False

If the request is SFW, and the worker accidentally generates NSFW, it will send back a censored image.

get_api_model_name classmethod

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "ModelGenerationInputStable"

validate_n

validate_n(value: int) -> int

Ensure that n is at least 1.

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@field_validator("n", mode="before")
def validate_n(cls, value: int) -> int:
    """Ensure that n is at least 1."""
    if value == 0:
        logger.debug("n (number of images to generate) is not set; defaulting to 1")
        return 1

    return value

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

Return a set of fields which should be redacted from logs.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    """Return a set of fields which should be redacted from logs."""
    return {"apikey"}

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

Return an additional set of fields to exclude from the log_safe_model_dump method.

Source code in horde_sdk/generic_api/apimodels.py
def get_extra_fields_to_exclude_from_log(self) -> set[str]:
    """Return an additional set of fields to exclude from the log_safe_model_dump method."""
    return set()

log_safe_model_dump

log_safe_model_dump(
    extra_exclude: set[str] | None = None,
) -> dict[Any, Any]

Return a dict of the model's fields, with any sensitive fields redacted.

Source code in horde_sdk/generic_api/apimodels.py
def log_safe_model_dump(self, extra_exclude: set[str] | None = None) -> dict[Any, Any]:
    """Return a dict of the model's fields, with any sensitive fields redacted."""
    if extra_exclude is None:
        extra_exclude = set()

    if hasattr(self, "model_dump"):
        return self.model_dump(  # type: ignore
            exclude=self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude,
        )

    logger.warning("Model does not have a model_dump method. Using python native class compatible method.")
    logger.debug(
        "Generally this should not be relied upon. If you're seeing this and you're a developer for the SDK, "
        "consider using pydantic models instead.",
    )
    # Its not a pydantic model, use python native class compatible method
    return {
        key: getattr(self, key)
        for key in self.__dict__
        if key not in self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude
    }

post_processors_must_be_known

post_processors_must_be_known(
    v: list[
        str
        | KNOWN_UPSCALERS
        | KNOWN_FACEFIXERS
        | KNOWN_MISC_POST_PROCESSORS
    ],
) -> list[
    str
    | KNOWN_UPSCALERS
    | KNOWN_FACEFIXERS
    | KNOWN_MISC_POST_PROCESSORS
]

Ensure that the post processors are in this list of supported post processors.

Source code in horde_sdk/ai_horde_api/apimodels/base.py
@field_validator("post_processing")
def post_processors_must_be_known(
    cls,
    v: list[str | KNOWN_UPSCALERS | KNOWN_FACEFIXERS | KNOWN_MISC_POST_PROCESSORS],
) -> list[str | KNOWN_UPSCALERS | KNOWN_FACEFIXERS | KNOWN_MISC_POST_PROCESSORS]:
    """Ensure that the post processors are in this list of supported post processors."""
    _valid_types: list[type] = [str, KNOWN_UPSCALERS, KNOWN_FACEFIXERS, KNOWN_MISC_POST_PROCESSORS]
    for post_processor in v:
        if post_processor not in _all_valid_post_processors_names_and_values or (
            type(post_processor) not in _valid_types
        ):
            logger.warning(
                f"Unknown post processor {post_processor}. Is your SDK out of date or did the API change?",
            )
    return v

sampler_name_must_be_known

sampler_name_must_be_known(
    v: str | KNOWN_IMAGE_SAMPLERS,
) -> str | KNOWN_IMAGE_SAMPLERS

Ensure that the sampler name is in this list of supported samplers.

Source code in horde_sdk/ai_horde_api/apimodels/base.py
@field_validator("sampler_name")
def sampler_name_must_be_known(cls, v: str | KNOWN_IMAGE_SAMPLERS) -> str | KNOWN_IMAGE_SAMPLERS:
    """Ensure that the sampler name is in this list of supported samplers."""
    if isinstance(v, KNOWN_IMAGE_SAMPLERS):
        return v

    try:
        KNOWN_IMAGE_SAMPLERS(v)
    except ValueError:
        logger.warning(f"Unknown sampler name {v}. Is your SDK out of date or did the API change?")

    return v

random_seed_if_none

random_seed_if_none(v: str | None) -> str | None

If the seed is None, generate a random seed.

Source code in horde_sdk/ai_horde_api/apimodels/base.py
@field_validator("seed")
def random_seed_if_none(cls, v: str | None) -> str | None:
    """If the seed is None, generate a random seed."""
    if v is None:
        random_seed = str(random.randint(1, 1000000000))
        logger.debug(f"Using random seed ({random_seed})")
        return random_seed

    return v

control_type_must_be_known

control_type_must_be_known(
    v: str | KNOWN_IMAGE_CONTROLNETS | None,
) -> str | KNOWN_IMAGE_CONTROLNETS | None

Ensure that the control type is in this list of supported control types.

Source code in horde_sdk/ai_horde_api/apimodels/base.py
@field_validator("control_type")
def control_type_must_be_known(
    cls,
    v: str | KNOWN_IMAGE_CONTROLNETS | None,
) -> str | KNOWN_IMAGE_CONTROLNETS | None:
    """Ensure that the control type is in this list of supported control types."""
    if v is None:
        return None
    if isinstance(v, KNOWN_IMAGE_CONTROLNETS):
        return v

    try:
        KNOWN_IMAGE_CONTROLNETS(v)
    except ValueError:
        logger.warning(f"Unknown control type {v}. Is your SDK out of date or did the API change?")

    return v

ImageGenerateAsyncRequest

Bases: BaseAIHordeRequest, APIKeyAllowedInRequestMixin, RequestUsesWorkerMixin

Request to start an image generation job.

Upon a successful response, you should poll the status of the job using the check endpoint. See :class:horde_sdk.ai_horde_api.apimodels.generate.check.ImageGenerateCheckRequest for more information.

Represents a POST request to the /v2/generate/async endpoint.

v2 API Model: GenerationInputStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@Unhashable
@Unequatable
class ImageGenerateAsyncRequest(
    BaseAIHordeRequest,
    APIKeyAllowedInRequestMixin,
    RequestUsesWorkerMixin,
):
    """Request to start an image generation job.

    Upon a successful response, you should poll the status of the job using the
    check endpoint. See :class:`horde_sdk.ai_horde_api.apimodels.generate.check.ImageGenerateCheckRequest` for more
    information.

    Represents a POST request to the /v2/generate/async endpoint.

    v2 API Model: `GenerationInputStable`
    """

    prompt: str
    """The prompt which will be sent to Stable Diffusion to generate an image."""

    params: ImageGenerationInputPayload | None = None
    """The parameters for the image generation."""

    nsfw: bool | None = True
    """Set to true if this request is NSFW. This will skip workers which censor images."""
    censor_nsfw: bool = False
    """If the request is SFW, and the worker accidentally generates NSFW, it will send back a censored image."""

    r2: bool = True
    """If True, the image will be sent via cloudflare r2 download link."""

    shared: bool = False
    """If True, The image will be shared with LAION for improving their dataset. This will also reduce your
    kudos consumption by 2. For anonymous users, this is always True."""

    replacement_filter: bool = True
    """If enabled, suspicious prompts are sanitized through a string replacement filter instead."""

    source_image: str | None = None
    """The public URL of the source image or a base64 string to use."""
    source_processing: KNOWN_IMAGE_SOURCE_PROCESSING = KNOWN_IMAGE_SOURCE_PROCESSING.txt2img
    """If source_image is provided, specifies how to process it."""
    source_mask: str | None = None
    """If source_processing is set to 'inpainting' or 'outpainting', this parameter can be optionally provided as the
    Base64-encoded webp mask of the areas to inpaint. If this arg is not passed, the inpainting/outpainting mask has to
    be embedded as alpha channel."""
    extra_source_images: list[ExtraSourceImageEntry] | None = None
    """Additional uploaded images which can be used for further operations."""

    @model_validator(mode="after")
    def validate_censor_nsfw(self) -> ImageGenerateAsyncRequest:
        """Ensure that censor_nsfw is not set when nsfw is enabled."""
        if self.nsfw and self.censor_nsfw:
            raise ValueError("Cannot censor NSFW content when NSFW detection is enabled.")
        return self

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "GenerationInputStable"

    @override
    @classmethod
    def get_http_method(cls) -> HTTPMethod:
        return HTTPMethod.POST

    @override
    @classmethod
    def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
        return AI_HORDE_API_ENDPOINT_SUBPATH.v2_generate_async

    @override
    @classmethod
    def get_default_success_response_type(cls) -> type[ImageGenerateAsyncResponse]:
        return ImageGenerateAsyncResponse

    @override
    @classmethod
    def get_success_status_response_pairs(cls) -> dict[HTTPStatusCode, type[HordeResponseTypes]]:
        return {
            HTTPStatusCode.OK: ImageGenerateAsyncDryRunResponse,
            HTTPStatusCode.ACCEPTED: cls.get_default_success_response_type(),
        }

    @override
    def get_number_of_results_expected(self) -> int:
        if not self.params:
            return 1
        return self.params.n

    @override
    def get_extra_fields_to_exclude_from_log(self) -> set[str]:
        return {"source_image", "source_mask", "extra_source_images"}

prompt instance-attribute

prompt: str

The prompt which will be sent to Stable Diffusion to generate an image.

params class-attribute instance-attribute

params: ImageGenerationInputPayload | None = None

The parameters for the image generation.

nsfw class-attribute instance-attribute

nsfw: bool | None = True

Set to true if this request is NSFW. This will skip workers which censor images.

censor_nsfw class-attribute instance-attribute

censor_nsfw: bool = False

If the request is SFW, and the worker accidentally generates NSFW, it will send back a censored image.

r2 class-attribute instance-attribute

r2: bool = True

If True, the image will be sent via cloudflare r2 download link.

shared class-attribute instance-attribute

shared: bool = False

If True, The image will be shared with LAION for improving their dataset. This will also reduce your kudos consumption by 2. For anonymous users, this is always True.

replacement_filter class-attribute instance-attribute

replacement_filter: bool = True

If enabled, suspicious prompts are sanitized through a string replacement filter instead.

source_image class-attribute instance-attribute

source_image: str | None = None

The public URL of the source image or a base64 string to use.

source_processing class-attribute instance-attribute

source_processing: KNOWN_IMAGE_SOURCE_PROCESSING = txt2img

If source_image is provided, specifies how to process it.

source_mask class-attribute instance-attribute

source_mask: str | None = None

If source_processing is set to 'inpainting' or 'outpainting', this parameter can be optionally provided as the Base64-encoded webp mask of the areas to inpaint. If this arg is not passed, the inpainting/outpainting mask has to be embedded as alpha channel.

extra_source_images class-attribute instance-attribute

extra_source_images: list[ExtraSourceImageEntry] | None = (
    None
)

Additional uploaded images which can be used for further operations.

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

trusted_workers class-attribute instance-attribute

trusted_workers: bool = False

When true, only trusted workers will serve this request. When False, Evaluating workers will also be used which can increase speed but adds more risk!

slow_workers class-attribute instance-attribute

slow_workers: bool = True

When True, allows slower workers to pick up this request. Disabling this incurs an extra kudos cost.

extra_slow_workers class-attribute instance-attribute

extra_slow_workers: bool = False

When True, allows extra slow workers to pick up this request.

workers class-attribute instance-attribute

workers: list[str] = Field(default_factory=list)

A list of worker IDs to use for this request. If empty, any worker can pick up the request. Using this incurs and extra kudos cost.

worker_blacklist class-attribute instance-attribute

worker_blacklist: bool = False

If true, the worker list will be treated as a blacklist instead of a whitelist.

models instance-attribute

models: list[str]

The generative models to use for this request.

validated_backends class-attribute instance-attribute

validated_backends: bool | None = Field(default=None)

When true, only inference backends that are validated by the AI Horde devs will serve this request. When False, non-validated backends will also be used which can increase speed but you may end up with unexpected results.

style class-attribute instance-attribute

style: str | None = Field(
    default=None,
    examples=["00000000-0000-0000-0000-000000000000"],
)

The style ID to use for the generation.

dry_run class-attribute instance-attribute

dry_run: bool = False

If true, the request will not be processed, but will return a response with the estimated kudos cost.

webhook class-attribute instance-attribute

webhook: str | None = Field(default=None)

Provide a URL where the AI Horde will send a POST call after each delivered generation. The request will include the details of the job as well as the request ID.

allow_downgrade class-attribute instance-attribute

allow_downgrade: bool = False

If true, the request can be modified to lesser parameters if the original request is too expensive.

proxied_account class-attribute instance-attribute

proxied_account: str | None = Field(default=None)

The account this request is being proxied for. This requires the sending API to be a service account.

disable_batching class-attribute instance-attribute

disable_batching: bool | None = Field(False)

When true, This request will not use batching. This will allow you to retrieve accurate seeds. Feature is restricted to Trusted users and Patreons.

apikey class-attribute instance-attribute

apikey: str | None = None

Defaults to ANON_API_KEY. See also .is_api_key_required()

accept class-attribute instance-attribute

accept: GenericAcceptTypes = json

The 'accept' header field.

client_agent class-attribute instance-attribute

client_agent: str = Field(
    default=default_bridge_agent_string,
    alias="Client-Agent",
)

The requesting client's agent. You should set this to reflect the name, version and contact information for your client.

validate_censor_nsfw

validate_censor_nsfw() -> ImageGenerateAsyncRequest

Ensure that censor_nsfw is not set when nsfw is enabled.

Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@model_validator(mode="after")
def validate_censor_nsfw(self) -> ImageGenerateAsyncRequest:
    """Ensure that censor_nsfw is not set when nsfw is enabled."""
    if self.nsfw and self.censor_nsfw:
        raise ValueError("Cannot censor NSFW content when NSFW detection is enabled.")
    return self

get_api_model_name classmethod

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "GenerationInputStable"

get_http_method classmethod

get_http_method() -> HTTPMethod
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_http_method(cls) -> HTTPMethod:
    return HTTPMethod.POST

get_api_endpoint_subpath classmethod

get_api_endpoint_subpath() -> AI_HORDE_API_ENDPOINT_SUBPATH
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
    return AI_HORDE_API_ENDPOINT_SUBPATH.v2_generate_async

get_default_success_response_type classmethod

get_default_success_response_type() -> (
    type[ImageGenerateAsyncResponse]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_default_success_response_type(cls) -> type[ImageGenerateAsyncResponse]:
    return ImageGenerateAsyncResponse

get_success_status_response_pairs classmethod

get_success_status_response_pairs() -> (
    dict[HTTPStatusCode, type[HordeResponseTypes]]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
@classmethod
def get_success_status_response_pairs(cls) -> dict[HTTPStatusCode, type[HordeResponseTypes]]:
    return {
        HTTPStatusCode.OK: ImageGenerateAsyncDryRunResponse,
        HTTPStatusCode.ACCEPTED: cls.get_default_success_response_type(),
    }

get_number_of_results_expected

get_number_of_results_expected() -> int
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
def get_number_of_results_expected(self) -> int:
    if not self.params:
        return 1
    return self.params.n

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]
Source code in horde_sdk/ai_horde_api/apimodels/generate/async_.py
@override
def get_extra_fields_to_exclude_from_log(self) -> set[str]:
    return {"source_image", "source_mask", "extra_source_images"}

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]
Source code in horde_sdk/generic_api/apimodels.py
@override
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    return {"apikey"}

log_safe_model_dump

log_safe_model_dump(
    extra_exclude: set[str] | None = None,
) -> dict[Any, Any]

Return a dict of the model's fields, with any sensitive fields redacted.

Source code in horde_sdk/generic_api/apimodels.py
def log_safe_model_dump(self, extra_exclude: set[str] | None = None) -> dict[Any, Any]:
    """Return a dict of the model's fields, with any sensitive fields redacted."""
    if extra_exclude is None:
        extra_exclude = set()

    if hasattr(self, "model_dump"):
        return self.model_dump(  # type: ignore
            exclude=self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude,
        )

    logger.warning("Model does not have a model_dump method. Using python native class compatible method.")
    logger.debug(
        "Generally this should not be relied upon. If you're seeing this and you're a developer for the SDK, "
        "consider using pydantic models instead.",
    )
    # Its not a pydantic model, use python native class compatible method
    return {
        key: getattr(self, key)
        for key in self.__dict__
        if key not in self.get_sensitive_fields() | self.get_extra_fields_to_exclude_from_log() | extra_exclude
    }

is_api_key_required classmethod

is_api_key_required() -> bool

Return whether this endpoint requires an API key.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def is_api_key_required(cls) -> bool:
    """Return whether this endpoint requires an API key."""
    return True

validate_api_key_length

validate_api_key_length(v: str) -> str

Validate that the API key is the correct length, or is the special ANON_API_KEY.

Source code in horde_sdk/generic_api/apimodels.py
@field_validator("apikey", mode="before")
def validate_api_key_length(cls, v: str) -> str:
    """Validate that the API key is the correct length, or is the special ANON_API_KEY."""
    if v is None:
        return ANON_API_KEY
    if v == ANON_API_KEY:
        return v
    if len(v) == 36:
        try:
            uuid.UUID(v)
            return v
        except ValueError as e:
            raise ValueError("API key must be a valid UUID") from e
        return v
    if len(v) != 22:
        raise ValueError("API key must be 22 characters long")
    return v

get_api_endpoint_url classmethod

get_api_endpoint_url() -> str

Return the endpoint URL, including the path to the specific API action defined by this object.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_api_endpoint_url(cls) -> str:
    """Return the endpoint URL, including the path to the specific API action defined by this object."""
    return url_with_path(base_url=cls.get_api_url(), path=cls.get_api_endpoint_subpath())

get_api_url classmethod

get_api_url() -> str
Source code in horde_sdk/ai_horde_api/apimodels/base.py
@override
@classmethod
def get_api_url(cls) -> str:
    return AI_HORDE_BASE_URL

get_header_fields classmethod

get_header_fields() -> list[str]

Return a list of field names from this request object that should be sent as header fields.

This is in addition to GenericHeaderFields's values, and possibly the API specific class which inherits from GenericHeaderFields, typically found in the horde_sdk.<api_name>_api.metadata module.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_header_fields(cls) -> list[str]:
    """Return a list of field names from this request object that should be sent as header fields.

    This is in addition to `GenericHeaderFields`'s values, and possibly the API specific class
    which inherits from `GenericHeaderFields`, typically found in the `horde_sdk.<api_name>_api.metadata` module.
    """
    return []

get_query_fields classmethod

get_query_fields() -> list[str]

Return a list of field names from this request object that should be sent as query parameters.

This is in addition to GenericQueryFields's values, and possibly the API specific class which inherits from GenericQueryFields, typically found in the horde_sdk.<api_name>_api.metadata module.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def get_query_fields(cls) -> list[str]:
    """Return a list of field names from this request object that should be sent as query parameters.

    This is in addition to `GenericQueryFields`'s values, and possibly the API specific class
    which inherits from `GenericQueryFields`, typically found in the `horde_sdk.<api_name>_api.metadata` module.
    """
    return []

get_requires_follow_up

get_requires_follow_up() -> bool

Return whether this request requires a follow up request(s).

Returns:

  • bool ( bool ) –

    Whether this request requires a follow up request to close the job on the server.

Source code in horde_sdk/generic_api/apimodels.py
def get_requires_follow_up(self) -> bool:
    """Return whether this request requires a follow up request(s).

    Returns:
        bool: Whether this request requires a follow up request to close the job on the server.
    """
    for response_type in self.get_success_status_response_pairs().values():
        if issubclass(response_type, ResponseRequiringFollowUpMixin):
            return True
    return False