Skip to content

pop

NoValidRequestFound

Bases: HordeAPIObjectBaseModel

Base class for the number of jobs a worker skipped for, and why.

v2 API Model: NoValidRequestFound

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class NoValidRequestFound(HordeAPIObjectBaseModel):
    """Base class for the number of jobs a worker skipped for, and why.

    v2 API Model: `NoValidRequestFound`
    """

    blacklist: int | None = Field(default=None, ge=0)
    """How many waiting requests were skipped because they demanded a generation with a word that this worker does
    not accept."""
    bridge_version: int | None = Field(default=None, examples=[0], ge=0)
    """How many waiting requests were skipped because they require a higher version of the bridge than this worker
    is running (upgrade if you see this in your skipped list)."""
    kudos: int | None = Field(default=None)
    """How many waiting requests were skipped because the user didn't have enough kudos when this worker requires"""
    models: int | None = Field(default=None, examples=[0], ge=0)
    """How many waiting requests were skipped because they demanded a different model than what this worker
    provides."""
    nsfw: int | None = Field(default=None, ge=0)
    """How many waiting requests were skipped because they demanded a nsfw generation which this worker does not
    provide."""
    performance: int | None = Field(
        default=None,
        ge=0,
    )
    """How many waiting requests were skipped because they demanded a higher performance than this worker provides."""
    untrusted: int | None = Field(default=None, ge=0)
    """How many waiting requests were skipped because they demanded a trusted worker which this worker is not."""
    worker_id: int | None = Field(
        default=None,
        ge=0,
    )
    """How many waiting requests were skipped because they demanded a specific worker."""

    def is_empty(self) -> bool:
        """Whether or not this object has any non-zero values."""
        return len(self.model_fields_set) == 0

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

blacklist class-attribute instance-attribute 

blacklist: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a generation with a word that this worker does not accept.

bridge_version class-attribute instance-attribute 

bridge_version: int | None = Field(
    default=None, examples=[0], ge=0
)

How many waiting requests were skipped because they require a higher version of the bridge than this worker is running (upgrade if you see this in your skipped list).

kudos class-attribute instance-attribute 

kudos: int | None = Field(default=None)

How many waiting requests were skipped because the user didn't have enough kudos when this worker requires

models class-attribute instance-attribute 

models: int | None = Field(default=None, examples=[0], ge=0)

How many waiting requests were skipped because they demanded a different model than what this worker provides.

nsfw class-attribute instance-attribute 

nsfw: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a nsfw generation which this worker does not provide.

performance class-attribute instance-attribute 

performance: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a higher performance than this worker provides.

untrusted class-attribute instance-attribute 

untrusted: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a trusted worker which this worker is not.

worker_id class-attribute instance-attribute 

worker_id: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a specific worker.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

is_empty 

is_empty() -> bool

Whether or not this object has any non-zero values.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
70
71
72
def is_empty(self) -> bool:
    """Whether or not this object has any non-zero values."""
    return len(self.model_fields_set) == 0

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
74
75
76
77
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "NoValidRequestFound"

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 
53
54
55
56
@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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
    }

ImageGenerateJobPopSkippedStatus

Bases: NoValidRequestFound

The number of jobs a worker was skipped for, and why.

v2 API Model: NoValidRequestFoundStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class ImageGenerateJobPopSkippedStatus(NoValidRequestFound):
    """The number of jobs a worker was skipped for, and why.

    v2 API Model: `NoValidRequestFoundStable`
    """

    max_pixels: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they demanded a higher size than this worker provides."""
    unsafe_ip: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they came from an unsafe IP."""
    img2img: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they requested img2img."""
    painting: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they requested inpainting/outpainting."""
    post_processing: int = Field(default=0, ge=0, alias="post-processing")
    """How many waiting requests were skipped because they requested post-processing."""
    lora: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they requested loras."""
    controlnet: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they requested a controlnet."""
    step_count: int = Field(default=0, ge=0)
    """How many waiting requests were skipped because they requested more steps than this worker provides."""

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

max_pixels class-attribute instance-attribute 

max_pixels: int = Field(default=0, ge=0)

How many waiting requests were skipped because they demanded a higher size than this worker provides.

unsafe_ip class-attribute instance-attribute 

unsafe_ip: int = Field(default=0, ge=0)

How many waiting requests were skipped because they came from an unsafe IP.

img2img class-attribute instance-attribute 

img2img: int = Field(default=0, ge=0)

How many waiting requests were skipped because they requested img2img.

painting class-attribute instance-attribute 

painting: int = Field(default=0, ge=0)

How many waiting requests were skipped because they requested inpainting/outpainting.

post_processing class-attribute instance-attribute 

post_processing: int = Field(
    default=0, ge=0, alias="post-processing"
)

How many waiting requests were skipped because they requested post-processing.

lora class-attribute instance-attribute 

lora: int = Field(default=0, ge=0)

How many waiting requests were skipped because they requested loras.

controlnet class-attribute instance-attribute 

controlnet: int = Field(default=0, ge=0)

How many waiting requests were skipped because they requested a controlnet.

step_count class-attribute instance-attribute 

step_count: int = Field(default=0, ge=0)

How many waiting requests were skipped because they requested more steps than this worker provides.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

blacklist class-attribute instance-attribute 

blacklist: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a generation with a word that this worker does not accept.

bridge_version class-attribute instance-attribute 

bridge_version: int | None = Field(
    default=None, examples=[0], ge=0
)

How many waiting requests were skipped because they require a higher version of the bridge than this worker is running (upgrade if you see this in your skipped list).

kudos class-attribute instance-attribute 

kudos: int | None = Field(default=None)

How many waiting requests were skipped because the user didn't have enough kudos when this worker requires

models class-attribute instance-attribute 

models: int | None = Field(default=None, examples=[0], ge=0)

How many waiting requests were skipped because they demanded a different model than what this worker provides.

nsfw class-attribute instance-attribute 

nsfw: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a nsfw generation which this worker does not provide.

performance class-attribute instance-attribute 

performance: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a higher performance than this worker provides.

untrusted class-attribute instance-attribute 

untrusted: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a trusted worker which this worker is not.

worker_id class-attribute instance-attribute 

worker_id: int | None = Field(default=None, ge=0)

How many waiting requests were skipped because they demanded a specific worker.

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
103
104
105
106
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "NoValidRequestFoundStable"

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 
53
54
55
56
@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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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_empty 

is_empty() -> bool

Whether or not this object has any non-zero values.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
70
71
72
def is_empty(self) -> bool:
    """Whether or not this object has any non-zero values."""
    return len(self.model_fields_set) == 0

ImageGenerateJobPopPayload

Bases: ImageGenerateParamMixin

Mixin for the additional image generation parameters used in dispatching a job to a worker.

v2 API Model: ModelPayloadRootStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class ImageGenerateJobPopPayload(ImageGenerateParamMixin):
    """Mixin for the additional image generation parameters used in dispatching a job to a worker.

    v2 API Model: `ModelPayloadRootStable`
    """

    prompt: str | None = None
    """The prompt to use for this image generation."""

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

    n_iter: 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."""

prompt class-attribute instance-attribute 

prompt: str | None = None

The prompt to use for this image generation.

ddim_steps class-attribute instance-attribute 

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

The number of image generation steps to perform.

n_iter class-attribute instance-attribute 

n_iter: 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/base.py 
348
349
350
351
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "ModelPayloadRootStable"

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 
53
54
55
56
@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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
@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 
299
300
301
302
303
304
305
306
307
308
309
310
@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 
353
354
355
356
357
358
359
360
361
@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 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
@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

ExtraSourceImageMixin

Bases: ResponseRequiringDownloadMixin

Mixin for jobs which have extra source images.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class ExtraSourceImageMixin(ResponseRequiringDownloadMixin):
    """Mixin for jobs which have extra source images."""

    extra_source_images: list[ExtraSourceImageEntry] | None = None
    """Additional uploaded images (as base64) which can be used for further operations."""
    _downloaded_extra_source_images: list[ExtraSourceImageEntry] | None = None

    async def async_download_extra_source_images(
        self,
        client_session: aiohttp.ClientSession,
        *,
        max_retries: int = 5,
    ) -> list[ExtraSourceImageEntry] | None:
        """Download the extra source images concurrently.

        You can also use `get_downloaded_extra_source_images` to get the downloaded images later, if needed.

        Args:
            client_session: The aiohttp client session to use for downloading.
            max_retries: The maximum number of times to retry downloading an image.

        Returns:
            The downloaded extra source images.
        """
        if not self.extra_source_images:
            logger.info("No extra source images to download.")
            return None

        if self._downloaded_extra_source_images is not None:
            logger.warning("Extra source images already downloaded.")
            return self._downloaded_extra_source_images

        self._downloaded_extra_source_images = []

        download_tasks = []
        for extra_source_image in self.extra_source_images:
            download_tasks.append(
                asyncio.create_task(
                    self._download_image_if_needed(client_session, extra_source_image, max_retries),
                ),
            )

        await asyncio.gather(*download_tasks)

        self._sort_downloaded_images()
        return self._downloaded_extra_source_images.copy()

    def get_downloaded_extra_source_images(self) -> list[ExtraSourceImageEntry] | None:
        """Get the downloaded extra source images."""
        return (
            self._downloaded_extra_source_images.copy() if self._downloaded_extra_source_images is not None else None
        )

    async def _download_image_if_needed(
        self,
        client_session: aiohttp.ClientSession,
        extra_source_image: ExtraSourceImageEntry,
        max_retries: int,
    ) -> None:
        """Download an extra source image if it has not already been downloaded.

        Args:
            client_session: The aiohttp client session to use for downloading.
            extra_source_image: The extra source image to download.
            max_retries: The maximum number of times to retry downloading an image.
        """
        if self._downloaded_extra_source_images is None:
            self._downloaded_extra_source_images = []

        if extra_source_image.image in (entry.original_url for entry in self._downloaded_extra_source_images):
            logger.debug(f"Extra source image {extra_source_image.image} already downloaded.")
            return

        for attempt in range(max_retries):
            try:
                downloaded_image = await self.download_file_as_base64(client_session, extra_source_image.image)
                self._downloaded_extra_source_images.append(
                    ExtraSourceImageEntry(
                        image=downloaded_image,
                        strength=extra_source_image.strength,
                        original_url=extra_source_image.image,
                    ),
                )
                break
            except Exception as e:
                logger.error(f"Error downloading extra source image {extra_source_image.image}: {e}")
                if attempt == max_retries - 1:
                    logger.error(f"Failed to download image {extra_source_image.image} after {max_retries} attempts.")

    def _sort_downloaded_images(self) -> None:
        """Sort the downloaded extra source images in the order they were requested."""
        if self.extra_source_images is None or self._downloaded_extra_source_images is None:
            return

        _extra_source_images = self.extra_source_images.copy()
        self._downloaded_extra_source_images.sort(
            key=lambda entry: next(
                (i for i, image in enumerate(_extra_source_images) if image.image == entry.original_url),
                len(_extra_source_images),
            ),
        )

extra_source_images class-attribute instance-attribute 

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

Additional uploaded images (as base64) which can be used for further operations.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

async_download_extra_source_images async 

async_download_extra_source_images(
    client_session: ClientSession, *, max_retries: int = 5
) -> list[ExtraSourceImageEntry] | None

Download the extra source images concurrently.

You can also use get_downloaded_extra_source_images to get the downloaded images later, if needed.

Parameters:

  • client_session (ClientSession) –

    The aiohttp client session to use for downloading.

  • max_retries (int, default: 5 ) –

    The maximum number of times to retry downloading an image.

Returns:

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def async_download_extra_source_images(
    self,
    client_session: aiohttp.ClientSession,
    *,
    max_retries: int = 5,
) -> list[ExtraSourceImageEntry] | None:
    """Download the extra source images concurrently.

    You can also use `get_downloaded_extra_source_images` to get the downloaded images later, if needed.

    Args:
        client_session: The aiohttp client session to use for downloading.
        max_retries: The maximum number of times to retry downloading an image.

    Returns:
        The downloaded extra source images.
    """
    if not self.extra_source_images:
        logger.info("No extra source images to download.")
        return None

    if self._downloaded_extra_source_images is not None:
        logger.warning("Extra source images already downloaded.")
        return self._downloaded_extra_source_images

    self._downloaded_extra_source_images = []

    download_tasks = []
    for extra_source_image in self.extra_source_images:
        download_tasks.append(
            asyncio.create_task(
                self._download_image_if_needed(client_session, extra_source_image, max_retries),
            ),
        )

    await asyncio.gather(*download_tasks)

    self._sort_downloaded_images()
    return self._downloaded_extra_source_images.copy()

get_downloaded_extra_source_images 

get_downloaded_extra_source_images() -> (
    list[ExtraSourceImageEntry] | None
)

Get the downloaded extra source images.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
172
173
174
175
176
def get_downloaded_extra_source_images(self) -> list[ExtraSourceImageEntry] | None:
    """Get the downloaded extra source images."""
    return (
        self._downloaded_extra_source_images.copy() if self._downloaded_extra_source_images is not None else None
    )

download_file_as_base64 async 

download_file_as_base64(
    client_session: ClientSession, url: str
) -> str

Download a file and return the value as a base64 string.

Source code in horde_sdk/generic_api/apimodels.py 
304
305
306
307
308
async def download_file_as_base64(self, client_session: aiohttp.ClientSession, url: str) -> str:
    """Download a file and return the value as a base64 string."""
    async with client_session.get(url, ssl=_default_sslcontext) as response:
        response.raise_for_status()
        return base64.b64encode(await response.read()).decode("utf-8")

download_file_to_field_as_base64 async 

download_file_to_field_as_base64(
    client_session: ClientSession, url: str, field_name: str
) -> None

Download a file from a URL and save it to the field.

Parameters:

  • client_session (ClientSession) –

    The aiohttp client session to use for the download.

  • url (str) –

    The URL to download the file from.

  • field_name (str) –

    The name of the field to save the file to.

Source code in horde_sdk/generic_api/apimodels.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
async def download_file_to_field_as_base64(
    self,
    client_session: aiohttp.ClientSession,
    url: str,
    field_name: str,
) -> None:
    """Download a file from a URL and save it to the field.

    Args:
        client_session (aiohttp.ClientSession): The aiohttp client session to use for the download.
        url (str): The URL to download the file from.
        field_name (str): The name of the field to save the file to.
    """
    async with client_session.get(url, ssl=_default_sslcontext) as response:
        response.raise_for_status()
        setattr(self, field_name, base64.b64encode(await response.read()).decode("utf-8"))

async_download_additional_data abstractmethod async 

async_download_additional_data(
    client_session: ClientSession,
) -> None

Asynchronously download any additional data required for this response.

Source code in horde_sdk/generic_api/apimodels.py 
327
328
329
@abc.abstractmethod
async def async_download_additional_data(self, client_session: aiohttp.ClientSession) -> None:
    """Asynchronously download any additional data required for this response."""

download_additional_data abstractmethod 

download_additional_data() -> None

Download any additional data required for this response.

Source code in horde_sdk/generic_api/apimodels.py 
331
332
333
@abc.abstractmethod
def download_additional_data(self) -> None:
    """Download any additional data required for this response."""

PopResponseModelMessage

Bases: _ResponseModelMessageData

The message data which appears in a job pop response.

v2 API Model: _OVERLOADED_MODEL

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
class PopResponseModelMessage(_ResponseModelMessageData):
    """The message data which appears in a job pop response.

    v2 API Model: `_OVERLOADED_MODEL`
    """

    id_: str | None = Field(default=None, alias="id")
    """The ID of the message."""

    expiry: str | None = None
    """The time at which this message expires."""

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

id_ class-attribute instance-attribute 

id_: str | None = Field(default=None, alias='id')

The ID of the message.

expiry class-attribute instance-attribute 

expiry: str | None = None

The time at which this message expires.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

worker_id class-attribute instance-attribute 

worker_id: str | None = None

The ID of the worker that the message is for.

message instance-attribute 

message: str

The message.

origin class-attribute instance-attribute 

origin: str | None = None

The origin of the message.

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
240
241
242
243
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return _OVERLOADED_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 
53
54
55
56
@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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
    }

ImageGenerateJobPopResponse

Bases: HordeResponseBaseModel, ResponseRequiringFollowUpMixin, ExtraSourceImageMixin

Contains job data for workers, if any were available. Also contains data for jobs this worker was skipped for.

This is the key response type for all image workers as it contains all assignment data for the worker.

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

v2 API Model: GenerationPayloadStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
class ImageGenerateJobPopResponse(
    HordeResponseBaseModel,
    ResponseRequiringFollowUpMixin,
    ExtraSourceImageMixin,
):
    """Contains job data for workers, if any were available. Also contains data for jobs this worker was skipped for.

    This is the key response type for all image workers as it contains all assignment data for the worker.

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

    v2 API Model: `GenerationPayloadStable`
    """

    id_: GenerationID | None = Field(default=None, alias="id")
    """(Obsolete) The UUID for this image generation."""
    ids: list[GenerationID]
    """A list of UUIDs for image generation."""

    payload: ImageGenerateJobPopPayload
    """The parameters used to generate this image."""
    skipped: ImageGenerateJobPopSkippedStatus = Field(default_factory=ImageGenerateJobPopSkippedStatus)
    """The reasons this worker was not issued certain jobs, and the number of jobs for each reason."""
    model: str | None = None
    """Which of the available models to use for this request."""
    source_image: str | None = None
    """The URL or Base64-encoded webp to use for img2img."""
    _downloaded_source_image: str | None = None
    """The downloaded source image (as base64), if any. This is not part of the API response."""
    source_processing: str | 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 img_processing is set to 'inpainting' or 'outpainting', this parameter can be optionally provided as the
    mask of the areas to inpaint. If this arg is not passed, the inpainting/outpainting mask has to be embedded as
    alpha channel."""
    _downloaded_source_mask: str | None = None
    """The downloaded source mask (as base64), if any. This is not part of the API response."""
    r2_upload: str | None = None
    """(Obsolete) The r2 upload link to use to upload this image."""
    r2_uploads: list[str] | None = None
    """The r2 upload links for each this image. Each index matches the ID in self.ids"""
    ttl: int | None = None
    """The amount of seconds before this job is considered stale and aborted."""

    messages: list[PopResponseModelMessage] | None = None
    """The messages that have been sent to this worker."""

    @field_validator("source_processing")
    def source_processing_must_be_known(
        cls,
        v: str | KNOWN_IMAGE_SOURCE_PROCESSING,
    ) -> str | KNOWN_IMAGE_SOURCE_PROCESSING:
        """Ensure that the source processing is in this list of supported source processing."""
        if isinstance(v, KNOWN_IMAGE_SOURCE_PROCESSING):
            return v

        try:
            KNOWN_IMAGE_SOURCE_PROCESSING(v)
        except ValueError:
            logger.warning(f"Unknown source processing {v}. Is your SDK out of date or did the API change?")
        return v

    @field_validator("id_", mode="before")
    def validate_id(cls, v: str | GenerationID) -> GenerationID | str:
        """Validate the ID is not an empty string."""
        if isinstance(v, str) and v == "":
            logger.warning("Job ID is empty")
            return GenerationID(root=uuid.uuid4())

        return v

    _ids_present: bool = False

    @property
    def ids_present(self) -> bool:
        """Whether or not the IDs are present."""
        return self._ids_present

    @model_validator(mode="after")
    def validate_ids_present(self) -> ImageGenerateJobPopResponse:
        """Ensure that either id_ or ids is present."""
        if self.model is None:
            if self.skipped.is_empty():
                logger.debug("No model or skipped data found in response.")
            else:
                logger.debug("No model found in response.")
            return self

        if self.id_ is None and len(self.ids) == 0:
            raise ValueError("Neither id_ nor ids were present in the response.")

        self._ids_present = True

        return self

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

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

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

    @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
    def get_follow_up_failure_cleanup_params(self) -> dict[str, object]:
        return {
            "state": GENERATION_STATE.faulted,
            "seed": self.payload.seed,
            "generation": "Faulted",
        }  # TODO: One day, could I do away with the magic string?

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

    @override
    def ignore_failure(self) -> bool:
        if self.id_ is None:
            return True

        return super().ignore_failure()

    @property
    def has_upscaler(self) -> bool:
        """Whether or not this image generation has an upscaler."""
        if len(self.payload.post_processing) == 0:
            return False

        return any(
            post_processing in KNOWN_UPSCALERS.__members__ or post_processing in KNOWN_UPSCALERS.__members__.values()
            for post_processing in self.payload.post_processing
        )

    @property
    def has_facefixer(self) -> bool:
        """Whether or not this image generation has a facefixer."""
        if len(self.payload.post_processing) == 0:
            return False

        return any(post_processing in KNOWN_FACEFIXERS.__members__ for post_processing in self.payload.post_processing)

    def get_downloaded_source_image(self) -> str | None:
        """Get the downloaded source image."""
        return self._downloaded_source_image

    def get_downloaded_source_mask(self) -> str | None:
        """Get the downloaded source mask."""
        return self._downloaded_source_mask

    def async_download_source_image(self, client_session: aiohttp.ClientSession) -> asyncio.Task[None]:
        """Download the source image concurrently."""
        # If the source image is not set, there is nothing to download.
        if self.source_image is None:
            return asyncio.create_task(asyncio.sleep(0))

        # If the source image is not a URL, it is already a base64 string.
        if urlparse(self.source_image).scheme not in ["http", "https"]:
            self._downloaded_source_image = self.source_image
            return asyncio.create_task(asyncio.sleep(0))

        return asyncio.create_task(
            self.download_file_to_field_as_base64(client_session, self.source_image, "_downloaded_source_image"),
        )

    def async_download_source_mask(self, client_session: aiohttp.ClientSession) -> asyncio.Task[None]:
        """Download the source mask concurrently."""
        # If the source mask is not set, there is nothing to download.
        if self.source_mask is None:
            return asyncio.create_task(asyncio.sleep(0))

        # If the source mask is not a URL, it is already a base64 string.
        if urlparse(self.source_mask).scheme not in ["http", "https"]:
            self._downloaded_source_mask = self.source_mask
            return asyncio.create_task(asyncio.sleep(0))

        return asyncio.create_task(
            self.download_file_to_field_as_base64(client_session, self.source_mask, "_downloaded_source_mask"),
        )

    @override
    async def async_download_additional_data(self, client_session: aiohttp.ClientSession) -> None:
        """Download all additional images concurrently."""
        await asyncio.gather(
            self.async_download_source_image(client_session),
            self.async_download_source_mask(client_session),
            self.async_download_extra_source_images(client_session),
        )

    @override
    def download_additional_data(self) -> None:
        raise NotImplementedError("This method is not yet implemented. Use async_download_additional_data instead.")

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ImageGenerateJobPopResponse):
            return False

        if self.id_ is not None and other.id_ is not None:
            return self.id_ == other.id_

        if len(self.ids) > 0 and len(other.ids) > 0:
            if len(self.ids) != len(other.ids):
                return False

            return all(i in other.ids for i in self.ids)

        logger.warning("No ID or IDs found in response. This is a bug.")
        return False

    def __hash__(self) -> int:
        if self.id_ is not None:
            return hash(ImageGenerateJobPopResponse.__name__) + hash(self.id_)

        if len(self.ids) > 0:
            return hash(ImageGenerateJobPopResponse.__name__) + hash(tuple(sorted(self.ids)))

        logger.warning("No ID or IDs found in response. This is a bug.")
        return hash(0)

id_ class-attribute instance-attribute 

id_: GenerationID | None = Field(default=None, alias='id')

(Obsolete) The UUID for this image generation.

ids instance-attribute 

ids: list[GenerationID]

A list of UUIDs for image generation.

payload instance-attribute 

payload: ImageGenerateJobPopPayload

The parameters used to generate this image.

skipped class-attribute instance-attribute 

skipped: ImageGenerateJobPopSkippedStatus = Field(
    default_factory=ImageGenerateJobPopSkippedStatus
)

The reasons this worker was not issued certain jobs, and the number of jobs for each reason.

model class-attribute instance-attribute 

model: str | None = None

Which of the available models to use for this request.

source_image class-attribute instance-attribute 

source_image: str | None = None

The URL or Base64-encoded webp to use for img2img.

source_processing class-attribute instance-attribute 

source_processing: str | 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 img_processing is set to 'inpainting' or 'outpainting', this parameter can be optionally provided as the mask of the areas to inpaint. If this arg is not passed, the inpainting/outpainting mask has to be embedded as alpha channel.

r2_upload class-attribute instance-attribute 

r2_upload: str | None = None

(Obsolete) The r2 upload link to use to upload this image.

r2_uploads class-attribute instance-attribute 

r2_uploads: list[str] | None = None

The r2 upload links for each this image. Each index matches the ID in self.ids

ttl class-attribute instance-attribute 

ttl: int | None = None

The amount of seconds before this job is considered stale and aborted.

messages class-attribute instance-attribute 

messages: list[PopResponseModelMessage] | None = None

The messages that have been sent to this worker.

ids_present property 

ids_present: bool

Whether or not the IDs are present.

has_upscaler property 

has_upscaler: bool

Whether or not this image generation has an upscaler.

has_facefixer property 

has_facefixer: bool

Whether or not this image generation has a facefixer.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

extra_source_images class-attribute instance-attribute 

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

Additional uploaded images (as base64) which can be used for further operations.

time_constructed property 

time_constructed: float

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

source_processing_must_be_known 

source_processing_must_be_known(
    v: str | KNOWN_IMAGE_SOURCE_PROCESSING,
) -> str | KNOWN_IMAGE_SOURCE_PROCESSING

Ensure that the source processing is in this list of supported source processing.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
@field_validator("source_processing")
def source_processing_must_be_known(
    cls,
    v: str | KNOWN_IMAGE_SOURCE_PROCESSING,
) -> str | KNOWN_IMAGE_SOURCE_PROCESSING:
    """Ensure that the source processing is in this list of supported source processing."""
    if isinstance(v, KNOWN_IMAGE_SOURCE_PROCESSING):
        return v

    try:
        KNOWN_IMAGE_SOURCE_PROCESSING(v)
    except ValueError:
        logger.warning(f"Unknown source processing {v}. Is your SDK out of date or did the API change?")
    return v

validate_id 

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

Validate the ID is not an empty string.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
308
309
310
311
312
313
314
315
@field_validator("id_", mode="before")
def validate_id(cls, v: str | GenerationID) -> GenerationID | str:
    """Validate the ID is not an empty string."""
    if isinstance(v, str) and v == "":
        logger.warning("Job ID is empty")
        return GenerationID(root=uuid.uuid4())

    return v

validate_ids_present 

validate_ids_present() -> ImageGenerateJobPopResponse

Ensure that either id_ or ids is present.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
@model_validator(mode="after")
def validate_ids_present(self) -> ImageGenerateJobPopResponse:
    """Ensure that either id_ or ids is present."""
    if self.model is None:
        if self.skipped.is_empty():
            logger.debug("No model or skipped data found in response.")
        else:
            logger.debug("No model found in response.")
        return self

    if self.id_ is None and len(self.ids) == 0:
        raise ValueError("Neither id_ nor ids were present in the response.")

    self._ids_present = True

    return self

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
341
342
343
344
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "GenerationPayloadStable"

get_follow_up_default_request_type classmethod 

get_follow_up_default_request_type() -> (
    type[ImageGenerationJobSubmitRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
346
347
348
349
@override
@classmethod
def get_follow_up_default_request_type(cls) -> type[ImageGenerationJobSubmitRequest]:
    return ImageGenerationJobSubmitRequest

get_follow_up_failure_cleanup_request_type classmethod 

get_follow_up_failure_cleanup_request_type() -> (
    type[ImageGenerationJobSubmitRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
351
352
353
354
@override
@classmethod
def get_follow_up_failure_cleanup_request_type(cls) -> type[ImageGenerationJobSubmitRequest]:
    return ImageGenerationJobSubmitRequest

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/pop.py 
356
357
358
359
360
@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_failure_cleanup_params 

get_follow_up_failure_cleanup_params() -> dict[str, object]
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
362
363
364
365
366
367
368
@override
def get_follow_up_failure_cleanup_params(self) -> dict[str, object]:
    return {
        "state": GENERATION_STATE.faulted,
        "seed": self.payload.seed,
        "generation": "Faulted",
    }  # TODO: One day, could I do away with the magic string?

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/pop.py 
370
371
372
@override
def get_extra_fields_to_exclude_from_log(self) -> set[str]:
    return {"source_image", "source_mask", "extra_source_images"}

ignore_failure 

ignore_failure() -> bool
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
374
375
376
377
378
379
@override
def ignore_failure(self) -> bool:
    if self.id_ is None:
        return True

    return super().ignore_failure()

get_downloaded_source_image 

get_downloaded_source_image() -> str | None

Get the downloaded source image.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
400
401
402
def get_downloaded_source_image(self) -> str | None:
    """Get the downloaded source image."""
    return self._downloaded_source_image

get_downloaded_source_mask 

get_downloaded_source_mask() -> str | None

Get the downloaded source mask.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
404
405
406
def get_downloaded_source_mask(self) -> str | None:
    """Get the downloaded source mask."""
    return self._downloaded_source_mask

async_download_source_image 

async_download_source_image(
    client_session: ClientSession,
) -> asyncio.Task[None]

Download the source image concurrently.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
408
409
410
411
412
413
414
415
416
417
418
419
420
421
def async_download_source_image(self, client_session: aiohttp.ClientSession) -> asyncio.Task[None]:
    """Download the source image concurrently."""
    # If the source image is not set, there is nothing to download.
    if self.source_image is None:
        return asyncio.create_task(asyncio.sleep(0))

    # If the source image is not a URL, it is already a base64 string.
    if urlparse(self.source_image).scheme not in ["http", "https"]:
        self._downloaded_source_image = self.source_image
        return asyncio.create_task(asyncio.sleep(0))

    return asyncio.create_task(
        self.download_file_to_field_as_base64(client_session, self.source_image, "_downloaded_source_image"),
    )

async_download_source_mask 

async_download_source_mask(
    client_session: ClientSession,
) -> asyncio.Task[None]

Download the source mask concurrently.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def async_download_source_mask(self, client_session: aiohttp.ClientSession) -> asyncio.Task[None]:
    """Download the source mask concurrently."""
    # If the source mask is not set, there is nothing to download.
    if self.source_mask is None:
        return asyncio.create_task(asyncio.sleep(0))

    # If the source mask is not a URL, it is already a base64 string.
    if urlparse(self.source_mask).scheme not in ["http", "https"]:
        self._downloaded_source_mask = self.source_mask
        return asyncio.create_task(asyncio.sleep(0))

    return asyncio.create_task(
        self.download_file_to_field_as_base64(client_session, self.source_mask, "_downloaded_source_mask"),
    )

async_download_additional_data async 

async_download_additional_data(
    client_session: ClientSession,
) -> None

Download all additional images concurrently.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
438
439
440
441
442
443
444
445
@override
async def async_download_additional_data(self, client_session: aiohttp.ClientSession) -> None:
    """Download all additional images concurrently."""
    await asyncio.gather(
        self.async_download_source_image(client_session),
        self.async_download_source_mask(client_session),
        self.async_download_extra_source_images(client_session),
    )

download_additional_data 

download_additional_data() -> None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
447
448
449
@override
def download_additional_data(self) -> None:
    raise NotImplementedError("This method is not yet implemented. Use async_download_additional_data instead.")

__eq__ 

__eq__(other: object) -> bool
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def __eq__(self, other: object) -> bool:
    if not isinstance(other, ImageGenerateJobPopResponse):
        return False

    if self.id_ is not None and other.id_ is not None:
        return self.id_ == other.id_

    if len(self.ids) > 0 and len(other.ids) > 0:
        if len(self.ids) != len(other.ids):
            return False

        return all(i in other.ids for i in self.ids)

    logger.warning("No ID or IDs found in response. This is a bug.")
    return False

__hash__ 

__hash__() -> int
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
467
468
469
470
471
472
473
474
475
def __hash__(self) -> int:
    if self.id_ is not None:
        return hash(ImageGenerateJobPopResponse.__name__) + hash(self.id_)

    if len(self.ids) > 0:
        return hash(ImageGenerateJobPopResponse.__name__) + hash(tuple(sorted(self.ids)))

    logger.warning("No ID or IDs found in response. This is a bug.")
    return hash(0)

download_file_as_base64 async 

download_file_as_base64(
    client_session: ClientSession, url: str
) -> str

Download a file and return the value as a base64 string.

Source code in horde_sdk/generic_api/apimodels.py 
304
305
306
307
308
async def download_file_as_base64(self, client_session: aiohttp.ClientSession, url: str) -> str:
    """Download a file and return the value as a base64 string."""
    async with client_session.get(url, ssl=_default_sslcontext) as response:
        response.raise_for_status()
        return base64.b64encode(await response.read()).decode("utf-8")

download_file_to_field_as_base64 async 

download_file_to_field_as_base64(
    client_session: ClientSession, url: str, field_name: str
) -> None

Download a file from a URL and save it to the field.

Parameters:

  • client_session (ClientSession) –

    The aiohttp client session to use for the download.

  • url (str) –

    The URL to download the file from.

  • field_name (str) –

    The name of the field to save the file to.

Source code in horde_sdk/generic_api/apimodels.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
async def download_file_to_field_as_base64(
    self,
    client_session: aiohttp.ClientSession,
    url: str,
    field_name: str,
) -> None:
    """Download a file from a URL and save it to the field.

    Args:
        client_session (aiohttp.ClientSession): The aiohttp client session to use for the download.
        url (str): The URL to download the file from.
        field_name (str): The name of the field to save the file to.
    """
    async with client_session.get(url, ssl=_default_sslcontext) as response:
        response.raise_for_status()
        setattr(self, field_name, base64.b64encode(await response.read()).decode("utf-8"))

async_download_extra_source_images async 

async_download_extra_source_images(
    client_session: ClientSession, *, max_retries: int = 5
) -> list[ExtraSourceImageEntry] | None

Download the extra source images concurrently.

You can also use get_downloaded_extra_source_images to get the downloaded images later, if needed.

Parameters:

  • client_session (ClientSession) –

    The aiohttp client session to use for downloading.

  • max_retries (int, default: 5 ) –

    The maximum number of times to retry downloading an image.

Returns:

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def async_download_extra_source_images(
    self,
    client_session: aiohttp.ClientSession,
    *,
    max_retries: int = 5,
) -> list[ExtraSourceImageEntry] | None:
    """Download the extra source images concurrently.

    You can also use `get_downloaded_extra_source_images` to get the downloaded images later, if needed.

    Args:
        client_session: The aiohttp client session to use for downloading.
        max_retries: The maximum number of times to retry downloading an image.

    Returns:
        The downloaded extra source images.
    """
    if not self.extra_source_images:
        logger.info("No extra source images to download.")
        return None

    if self._downloaded_extra_source_images is not None:
        logger.warning("Extra source images already downloaded.")
        return self._downloaded_extra_source_images

    self._downloaded_extra_source_images = []

    download_tasks = []
    for extra_source_image in self.extra_source_images:
        download_tasks.append(
            asyncio.create_task(
                self._download_image_if_needed(client_session, extra_source_image, max_retries),
            ),
        )

    await asyncio.gather(*download_tasks)

    self._sort_downloaded_images()
    return self._downloaded_extra_source_images.copy()

get_downloaded_extra_source_images 

get_downloaded_extra_source_images() -> (
    list[ExtraSourceImageEntry] | None
)

Get the downloaded extra source images.

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
172
173
174
175
176
def get_downloaded_extra_source_images(self) -> list[ExtraSourceImageEntry] | None:
    """Get the downloaded extra source images."""
    return (
        self._downloaded_extra_source_images.copy() if self._downloaded_extra_source_images is not None else None
    )

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 
166
167
168
169
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 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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_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 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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

get_follow_up_request_types classmethod 

get_follow_up_request_types() -> list[type[HordeRequest]]

Return a list of all the possible follow up request types for this response.

Source code in horde_sdk/generic_api/apimodels.py 
235
236
237
238
@classmethod
def get_follow_up_request_types(cls) -> list[type[HordeRequest]]:
    """Return a list of all the possible follow up request types for this response."""
    return [cls.get_follow_up_default_request_type()]

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 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
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

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 
53
54
55
56
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    """Return a set of fields which should be redacted from logs."""
    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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
    }

PopInput

Bases: HordeAPIObjectBaseModel

The input data for a image worker requesting jobs.

v2 API Model: PopInput

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
class PopInput(HordeAPIObjectBaseModel):
    """The input data for a image worker requesting jobs.

    v2 API Model: `PopInput`
    """

    amount: int | None = Field(1, ge=1, le=20)
    """The number of jobs to pop at the same time."""
    bridge_agent: str | None = Field(
        "unknown:0:unknown",
        examples=["AI Horde Worker reGen:4.1.0:https://github.com/Haidra-Org/horde-worker-reGen"],
        max_length=1000,
    )
    """The worker name, version and website."""
    models: list[str]
    """The models this worker can generate."""
    name: str
    """The Name of the Worker."""
    nsfw: bool | None = Field(
        default=False,
    )
    """Whether this worker can generate NSFW requests or not."""
    priority_usernames: list[str] | None = None
    """The usernames that should be prioritized by this worker."""
    require_upfront_kudos: bool | None = Field(
        default=False,
        description=(
            "If True, this worker will only pick up requests where the owner has the required kudos to consume already"
            " available."
        ),
        examples=[
            False,
        ],
    )
    """If True, this worker will only pick up requests where the owner has the required kudos to consume already
    available."""
    threads: int | None = Field(
        default=1,
        description=(
            "How many threads this worker is running. This is used to accurately the current power available in the"
            " horde."
        ),
        ge=1,
        le=50,
    )
    """How many threads this worker is running. This is used to accurately the current power available in the horde."""

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

amount class-attribute instance-attribute 

amount: int | None = Field(1, ge=1, le=20)

The number of jobs to pop at the same time.

bridge_agent class-attribute instance-attribute 

bridge_agent: str | None = Field(
    "unknown:0:unknown",
    examples=[
        "AI Horde Worker reGen:4.1.0:https://github.com/Haidra-Org/horde-worker-reGen"
    ],
    max_length=1000,
)

The worker name, version and website.

models instance-attribute 

models: list[str]

The models this worker can generate.

name instance-attribute 

name: str

The Name of the Worker.

nsfw class-attribute instance-attribute 

nsfw: bool | None = Field(default=False)

Whether this worker can generate NSFW requests or not.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] | None = None

The usernames that should be prioritized by this worker.

require_upfront_kudos class-attribute instance-attribute 

require_upfront_kudos: bool | None = Field(
    default=False,
    description="If True, this worker will only pick up requests where the owner has the required kudos to consume already available.",
    examples=[False],
)

If True, this worker will only pick up requests where the owner has the required kudos to consume already available.

threads class-attribute instance-attribute 

threads: int | None = Field(
    default=1,
    description="How many threads this worker is running. This is used to accurately the current power available in the horde.",
    ge=1,
    le=50,
)

How many threads this worker is running. This is used to accurately the current power available in the horde.

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/pop.py 
525
526
527
528
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "PopInput"

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 
53
54
55
56
@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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
    }

ImageGenerateJobPopRequest

Bases: BaseAIHordeRequest, APIKeyAllowedInRequestMixin, PopInput

Request additional jobs, if any are available, for an image worker.

This is the key request type for all image workers as it contains all the parameters needed to request a job.

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

v2 API Model: PopInputStable

Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
@Unhashable
@Unequatable
class ImageGenerateJobPopRequest(BaseAIHordeRequest, APIKeyAllowedInRequestMixin, PopInput):
    """Request additional jobs, if any are available, for an image worker.

    This is the key request type for all image workers as it contains all the parameters needed to request a job.

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

    v2 API Model: `PopInputStable`
    """

    bridge_version: int | None = None
    """The version of the bridge this worker is running."""
    max_pixels: int = Field(examples=[262144])
    """The maximum number of pixels this worker can generate."""
    blacklist: list[str] = Field(default_factory=list)
    """The list of words this worker will not accept in a prompt."""
    allow_img2img: bool = True
    """Whether this worker can generate img2img."""
    allow_painting: bool = False
    """Whether this worker can generate inpainting/outpainting."""
    allow_unsafe_ipaddr: bool = True
    """Whether this worker will generate from unsafe/VPN IP addresses."""
    allow_post_processing: bool = True
    """Whether this worker can do post-processing."""
    allow_controlnet: bool = False
    """Whether this worker can generate using controlnets."""
    allow_sdxl_controlnet: bool = False
    """Whether this worker can generate using SDXL controlnets."""
    allow_lora: bool = False
    """Whether this worker can generate using Loras."""
    extra_slow_worker: bool = False
    """Marks the worker as extra slow."""
    limit_max_steps: bool = False
    """Prevents the worker picking up jobs with more steps than the model average."""

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

    @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_pop

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

bridge_version class-attribute instance-attribute 

bridge_version: int | None = None

The version of the bridge this worker is running.

max_pixels class-attribute instance-attribute 

max_pixels: int = Field(examples=[262144])

The maximum number of pixels this worker can generate.

blacklist class-attribute instance-attribute 

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

The list of words this worker will not accept in a prompt.

allow_img2img class-attribute instance-attribute 

allow_img2img: bool = True

Whether this worker can generate img2img.

allow_painting class-attribute instance-attribute 

allow_painting: bool = False

Whether this worker can generate inpainting/outpainting.

allow_unsafe_ipaddr class-attribute instance-attribute 

allow_unsafe_ipaddr: bool = True

Whether this worker will generate from unsafe/VPN IP addresses.

allow_post_processing class-attribute instance-attribute 

allow_post_processing: bool = True

Whether this worker can do post-processing.

allow_controlnet class-attribute instance-attribute 

allow_controlnet: bool = False

Whether this worker can generate using controlnets.

allow_sdxl_controlnet class-attribute instance-attribute 

allow_sdxl_controlnet: bool = False

Whether this worker can generate using SDXL controlnets.

allow_lora class-attribute instance-attribute 

allow_lora: bool = False

Whether this worker can generate using Loras.

extra_slow_worker class-attribute instance-attribute 

extra_slow_worker: bool = False

Marks the worker as extra slow.

limit_max_steps class-attribute instance-attribute 

limit_max_steps: bool = False

Prevents the worker picking up jobs with more steps than the model average.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

amount class-attribute instance-attribute 

amount: int | None = Field(1, ge=1, le=20)

The number of jobs to pop at the same time.

bridge_agent class-attribute instance-attribute 

bridge_agent: str | None = Field(
    "unknown:0:unknown",
    examples=[
        "AI Horde Worker reGen:4.1.0:https://github.com/Haidra-Org/horde-worker-reGen"
    ],
    max_length=1000,
)

The worker name, version and website.

models instance-attribute 

models: list[str]

The models this worker can generate.

name instance-attribute 

name: str

The Name of the Worker.

nsfw class-attribute instance-attribute 

nsfw: bool | None = Field(default=False)

Whether this worker can generate NSFW requests or not.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] | None = None

The usernames that should be prioritized by this worker.

require_upfront_kudos class-attribute instance-attribute 

require_upfront_kudos: bool | None = Field(
    default=False,
    description="If True, this worker will only pick up requests where the owner has the required kudos to consume already available.",
    examples=[False],
)

If True, this worker will only pick up requests where the owner has the required kudos to consume already available.

threads class-attribute instance-attribute 

threads: int | None = Field(
    default=1,
    description="How many threads this worker is running. This is used to accurately the current power available in the horde.",
    ge=1,
    le=50,
)

How many threads this worker is running. This is used to accurately the current power available in the horde.

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.

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
568
569
570
571
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "PopInputStable"

get_http_method classmethod 

get_http_method() -> HTTPMethod
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
573
574
575
576
@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/pop.py 
578
579
580
581
@override
@classmethod
def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
    return AI_HORDE_API_ENDPOINT_SUBPATH.v2_generate_pop

get_default_success_response_type classmethod 

get_default_success_response_type() -> (
    type[ImageGenerateJobPopResponse]
)
Source code in horde_sdk/ai_horde_api/apimodels/generate/pop.py 
583
584
585
586
@override
@classmethod
def get_default_success_response_type(cls) -> type[ImageGenerateJobPopResponse]:
    return ImageGenerateJobPopResponse

get_sensitive_fields classmethod 

get_sensitive_fields() -> set[str]
Source code in horde_sdk/generic_api/apimodels.py 
477
478
479
480
@override
@classmethod
def get_sensitive_fields(cls) -> set[str]:
    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 
58
59
60
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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 
489
490
491
492
@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 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
@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 
407
408
409
410
@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 
44
45
46
47
@override
@classmethod
def get_api_url(cls) -> str:
    return AI_HORDE_BASE_URL

get_success_status_response_pairs classmethod 

get_success_status_response_pairs() -> (
    dict[HTTPStatusCode, type[HordeResponseTypes]]
)

Return a dict of HTTP status codes and the expected HordeResponse.

Defaults to {HTTPStatusCode.OK: cls.get_expected_response_type()}, but may be overridden to support other status codes.

Source code in horde_sdk/generic_api/apimodels.py 
427
428
429
430
431
432
433
434
435
436
@classmethod
def get_success_status_response_pairs(
    cls,
) -> dict[HTTPStatusCode, type[HordeResponseTypes]]:
    """Return a dict of HTTP status codes and the expected `HordeResponse`.

    Defaults to `{HTTPStatusCode.OK: cls.get_expected_response_type()}`, but may be overridden to support other
    status codes.
    """
    return {HTTPStatusCode.OK: cls.get_default_success_response_type()}

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 
438
439
440
441
442
443
444
445
@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 
447
448
449
450
451
452
453
454
@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_number_of_results_expected 

get_number_of_results_expected() -> int

Return the number of (job) results expected from this request.

Defaults to 1, but may be overridden to dynamically change the number of results expected.

This is factored into context management; if the number of results expected is not met, the job is considered unhandled on an exception and followed up on to attempt to close it.

Source code in horde_sdk/generic_api/apimodels.py 
456
457
458
459
460
461
462
463
464
def get_number_of_results_expected(self) -> int:
    """Return the number of (job) results expected from this request.

    Defaults to `1`, but may be overridden to dynamically change the number of results expected.

    This is factored into context management; if the number of results expected is not met, the job is considered
    unhandled on an exception and followed up on to attempt to close it.
    """
    return 1

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 
466
467
468
469
470
471
472
473
474
475
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