Skip to content

pop

AlchemyFormPayloadStable

Bases: HordeAPIObjectBaseModel

Currently unsupported.

v2 API Model: ModelInterrogationFormPayloadStable

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class AlchemyFormPayloadStable(HordeAPIObjectBaseModel):
    """Currently unsupported.

    v2 API Model: `ModelInterrogationFormPayloadStable`
    """

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

    additionalProp1: str = Field(
        validation_alias="additionalProp1",
    )
    """Currently unsupported."""
    additionalProp2: str = Field(
        validation_alias="additionalProp2",
    )
    """Currently unsupported."""
    additionalProp3: str = Field(
        validation_alias="additionalProp3",
    )
    """Currently unsupported."""

additionalProp1 class-attribute instance-attribute 

additionalProp1: str = Field(
    validation_alias="additionalProp1"
)

Currently unsupported.

additionalProp2 class-attribute instance-attribute 

additionalProp2: str = Field(
    validation_alias="additionalProp2"
)

Currently unsupported.

additionalProp3 class-attribute instance-attribute 

additionalProp3: str = Field(
    validation_alias="additionalProp3"
)

Currently unsupported.

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/alchemy/pop.py 
33
34
35
36
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "ModelInterrogationFormPayloadStable"

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
    }

AlchemyPopFormPayload

Bases: HordeAPIObjectBaseModel, JobRequestMixin

Contains the data for a single alchemy generation form for workers (pop).

v2 API Model: InterrogationPopFormPayload

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
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
78
79
80
81
82
83
84
85
86
class AlchemyPopFormPayload(HordeAPIObjectBaseModel, JobRequestMixin):
    """Contains the data for a single alchemy generation form for workers (pop).

    v2 API Model: `InterrogationPopFormPayload`
    """

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

    form: KNOWN_ALCHEMY_TYPES | str = Field(
        examples=["caption"],
    )
    """The name of this interrogation form."""

    @field_validator("form", mode="before")
    def validate_form(cls, v: str | KNOWN_ALCHEMY_TYPES) -> KNOWN_ALCHEMY_TYPES | str:
        """Ensure that the form is a known alchemy type."""
        if isinstance(v, KNOWN_ALCHEMY_TYPES):
            return v
        if isinstance(v, str) and v not in KNOWN_ALCHEMY_TYPES.__members__:
            logger.warning(f"Unknown form type {v}")
        return v

    payload: AlchemyFormPayloadStable | None = None
    """The setting for this interrogation form."""
    r2_upload: str | None = Field(
        default=None,
    )
    """The URL in which the post-processed image can be uploaded."""
    source_image: str | None = Field(
        default=None,
    )
    """The URL From which the source image can be downloaded."""

form class-attribute instance-attribute 

form: KNOWN_ALCHEMY_TYPES | str = Field(
    examples=["caption"]
)

The name of this interrogation form.

payload class-attribute instance-attribute 

payload: AlchemyFormPayloadStable | None = None

The setting for this interrogation form.

r2_upload class-attribute instance-attribute 

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

The URL in which the post-processed image can be uploaded.

source_image class-attribute instance-attribute 

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

The URL From which the source image can be downloaded.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

id_ class-attribute instance-attribute 

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

The UUID for this job. Use this to post the results in the future.

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
58
59
60
61
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "InterrogationPopFormPayload"

validate_form 

validate_form(
    v: str | KNOWN_ALCHEMY_TYPES,
) -> KNOWN_ALCHEMY_TYPES | str

Ensure that the form is a known alchemy type.

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
68
69
70
71
72
73
74
75
@field_validator("form", mode="before")
def validate_form(cls, v: str | KNOWN_ALCHEMY_TYPES) -> KNOWN_ALCHEMY_TYPES | str:
    """Ensure that the form is a known alchemy type."""
    if isinstance(v, KNOWN_ALCHEMY_TYPES):
        return v
    if isinstance(v, str) and v not in KNOWN_ALCHEMY_TYPES.__members__:
        logger.warning(f"Unknown form type {v}")
    return v

validate_id 

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

Ensure that the job ID is not empty.

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

    return v

__eq__ 

__eq__(__value: object) -> bool
Source code in horde_sdk/ai_horde_api/apimodels/base.py 
65
66
67
68
def __eq__(self, __value: object) -> bool:
    if isinstance(__value, JobRequestMixin):
        return self.id_ == __value.id_
    return False

__hash__ 

__hash__() -> int
Source code in horde_sdk/ai_horde_api/apimodels/base.py 
70
71
def __hash__(self) -> int:
    return hash(JobRequestMixin.__name__) + hash(self.id_)

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
    }

NoValidAlchemyFound

Bases: HordeAPIObjectBaseModel

The number of jobs this worker was not eligible for, and why.

v2 API Model: NoValidInterrogationsFound

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class NoValidAlchemyFound(HordeAPIObjectBaseModel):
    """The number of jobs this worker was not eligible for, and why.

    v2 API Model: `NoValidInterrogationsFound`
    """

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

    bridge_version: int | None = Field(
        default=None,
        description=(
            "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)."
        ),
        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)."""
    untrusted: int | None = Field(
        default=None,
        description=(
            "How many waiting requests were skipped because they demanded a trusted worker which this worker is not."
        ),
        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 __eq__(self, other: object) -> bool:
        if not isinstance(other, NoValidAlchemyFound):
            return False

        return (
            self.bridge_version == other.bridge_version
            and self.untrusted == other.untrusted
            and self.worker_id == other.worker_id
        )

    def __hash__(self) -> int:
        return hash((self.bridge_version, self.untrusted, self.worker_id))

bridge_version class-attribute instance-attribute 

bridge_version: int | None = Field(
    default=None,
    description="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).",
    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).

untrusted class-attribute instance-attribute 

untrusted: int | None = Field(
    default=None,
    description="How many waiting requests were skipped because they demanded a trusted worker which this worker is not.",
    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()

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
95
96
97
98
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "NoValidInterrogationsFound"

__eq__ 

__eq__(other: object) -> bool
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
125
126
127
128
129
130
131
132
133
def __eq__(self, other: object) -> bool:
    if not isinstance(other, NoValidAlchemyFound):
        return False

    return (
        self.bridge_version == other.bridge_version
        and self.untrusted == other.untrusted
        and self.worker_id == other.worker_id
    )

__hash__ 

__hash__() -> int
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
135
136
def __hash__(self) -> int:
    return hash((self.bridge_version, self.untrusted, self.worker_id))

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
    }

AlchemyJobPopResponse

Bases: HordeResponseBaseModel, ResponseRequiringFollowUpMixin

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 alchemy workers as it contains all assignment data for the worker.

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

v2 API Model: InterrogationPopPayload

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
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
class AlchemyJobPopResponse(HordeResponseBaseModel, ResponseRequiringFollowUpMixin):
    """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 alchemy workers as it contains all assignment data for the worker.

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

    v2 API Model: `InterrogationPopPayload`
    """

    # and not actually specifying a schema
    forms: list[AlchemyPopFormPayload] | None = None
    """The forms that to be generated"""
    skipped: NoValidAlchemyFound | None = None
    """The requests that were skipped because this worker were not eligible for them."""

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

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

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

    @override
    def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, object]]:
        if not self.forms:
            return []
        all_ids: list[dict[str, object]] = []
        for form in self.forms:
            if not isinstance(form, AlchemyPopFormPayload):
                logger.warning(f"Skipping form {form} as it is not an AlchemyPopFormPayload")
                continue
            if form.id_:
                if as_python_field_name:
                    all_ids.append({"id_": form.id_})
                else:
                    all_ids.append({"id": form.id_})

        return all_ids

    @override
    @classmethod
    def get_follow_up_request_types(cls) -> list[type[AlchemyJobSubmitRequest]]:  # type: ignore[override]
        """Return a list of all the possible follow up request types for this response."""
        return [AlchemyJobSubmitRequest]

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

        forms_match = True
        skipped_match = True

        if self.forms is not None and other.forms is not None:
            forms_match = all(form in other.forms for form in self.forms)

        if self.skipped is not None:
            skipped_match = self.skipped == other.skipped

        return forms_match and skipped_match

    def __hash__(self) -> int:
        if self.forms is None:
            return hash(self.skipped)

        return hash((tuple(sorted([form.id_ for form in self.forms])), self.skipped))

    @property
    def ids(self) -> list[GenerationID]:
        """Return a list of all the ids in the response."""
        if self.forms is None:
            return []
        return [form.id_ for form in self.forms]

    @property
    def ids_present(self) -> bool:
        """Return whether the response has any ids."""
        return bool(self.ids)

forms class-attribute instance-attribute 

forms: list[AlchemyPopFormPayload] | None = None

The forms that to be generated

skipped class-attribute instance-attribute 

skipped: NoValidAlchemyFound | None = None

The requests that were skipped because this worker were not eligible for them.

ids property 

ids: list[GenerationID]

Return a list of all the ids in the response.

ids_present property 

ids_present: bool

Return whether the response has any ids.

time_constructed property 

time_constructed: float

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

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

get_api_model_name classmethod 

get_api_model_name() -> str | None
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
155
156
157
158
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "InterrogationPopPayload"

get_follow_up_default_request_type classmethod 

get_follow_up_default_request_type() -> (
    type[AlchemyJobSubmitRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
160
161
162
163
@override
@classmethod
def get_follow_up_default_request_type(cls) -> type[AlchemyJobSubmitRequest]:
    return AlchemyJobSubmitRequest

get_follow_up_failure_cleanup_request_type classmethod 

get_follow_up_failure_cleanup_request_type() -> (
    type[AlchemyJobSubmitRequest]
)
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
165
166
167
168
@override
@classmethod
def get_follow_up_failure_cleanup_request_type(cls) -> type[AlchemyJobSubmitRequest]:
    return AlchemyJobSubmitRequest

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/alchemy/pop.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@override
def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, object]]:
    if not self.forms:
        return []
    all_ids: list[dict[str, object]] = []
    for form in self.forms:
        if not isinstance(form, AlchemyPopFormPayload):
            logger.warning(f"Skipping form {form} as it is not an AlchemyPopFormPayload")
            continue
        if form.id_:
            if as_python_field_name:
                all_ids.append({"id_": form.id_})
            else:
                all_ids.append({"id": form.id_})

    return all_ids

get_follow_up_request_types classmethod 

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

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

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
187
188
189
190
191
@override
@classmethod
def get_follow_up_request_types(cls) -> list[type[AlchemyJobSubmitRequest]]:  # type: ignore[override]
    """Return a list of all the possible follow up request types for this response."""
    return [AlchemyJobSubmitRequest]

__eq__ 

__eq__(other: object) -> bool
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def __eq__(self, other: object) -> bool:
    if not isinstance(other, AlchemyJobPopResponse):
        return False

    forms_match = True
    skipped_match = True

    if self.forms is not None and other.forms is not None:
        forms_match = all(form in other.forms for form in self.forms)

    if self.skipped is not None:
        skipped_match = self.skipped == other.skipped

    return forms_match and skipped_match

__hash__ 

__hash__() -> int
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
208
209
210
211
212
def __hash__(self) -> int:
    if self.forms is None:
        return hash(self.skipped)

    return hash((tuple(sorted([form.id_ for form in self.forms])), self.skipped))

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_params 

get_follow_up_failure_cleanup_params() -> dict[str, object]

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

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

This is used for context management.

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

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

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

get_follow_up_failure_cleanup_request 

get_follow_up_failure_cleanup_request() -> (
    list[HordeRequest]
)

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

Source code in horde_sdk/generic_api/apimodels.py 
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

ignore_failure 

ignore_failure() -> bool

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

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

does_target_request_follow_up 

does_target_request_follow_up(
    target_request: HordeRequest,
) -> bool

Return whether the target_request would follow up on this request.

Parameters:

  • target_request (HordeRequest) –

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

Returns:

  • bool ( bool ) –

    Whether the target_request would follow up on this request.

Source code in horde_sdk/generic_api/apimodels.py 
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"}

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
    }

AlchemyPopRequest

Bases: BaseAIHordeRequest, APIKeyAllowedInRequestMixin

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

This is the key request type for all alchemy workers as it requests all available jobs for the worker.

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

v2 API Model: InterrogationPopInput

Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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
@Unhashable
@Unequatable
class AlchemyPopRequest(BaseAIHordeRequest, APIKeyAllowedInRequestMixin):
    """Request additional jobs, if any are available, for an alchemy worker.

    This is the key request type for all alchemy workers as it requests all available jobs for the worker.

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

    v2 API Model: `InterrogationPopInput`
    """

    bridge_agent: str = Field(
        default="unknown",
        examples=["AI Horde Worker reGen:4.1.0:https://github.com/Haidra-Org/horde-worker-reGen"],
        max_length=1000,
    )
    """The agent string for the bridge, in the format `name:version:url or contact info`."""

    name: str
    """The name of the request. This is used to identify the request in the logs."""
    priority_usernames: list[str]
    """The usernames that should be prioritized for this request."""
    forms: list[KNOWN_ALCHEMY_TYPES]
    """The types of alchemy that should be generated."""
    amount: int
    """The number of jobs to request."""
    threads: int = Field(
        default=1,
        ge=1,
    )
    """The number of threads to report that this bridge is using to the API."""
    max_tiles: int = Field(
        default=16,
        ge=1,
        le=256,
    )
    """The maximum number of 512x512 tiles that this worker can process."""

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

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

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

bridge_agent class-attribute instance-attribute 

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

The agent string for the bridge, in the format name:version:url or contact info.

name instance-attribute 

name: str

The name of the request. This is used to identify the request in the logs.

priority_usernames instance-attribute 

priority_usernames: list[str]

The usernames that should be prioritized for this request.

forms instance-attribute 

forms: list[KNOWN_ALCHEMY_TYPES]

The types of alchemy that should be generated.

amount instance-attribute 

amount: int

The number of jobs to request.

threads class-attribute instance-attribute 

threads: int = Field(default=1, ge=1)

The number of threads to report that this bridge is using to the API.

max_tiles class-attribute instance-attribute 

max_tiles: int = Field(default=16, ge=1, le=256)

The maximum number of 512x512 tiles that this worker can process.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

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/alchemy/pop.py 
266
267
268
269
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "InterrogationPopInput"

get_http_method classmethod 

get_http_method() -> HTTPMethod
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
271
272
273
274
@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/alchemy/pop.py 
276
277
278
279
@override
@classmethod
def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
    return AI_HORDE_API_ENDPOINT_SUBPATH.v2_interrogate_pop

get_default_success_response_type classmethod 

get_default_success_response_type() -> (
    type[AlchemyJobPopResponse]
)
Source code in horde_sdk/ai_horde_api/apimodels/alchemy/pop.py 
281
282
283
284
@override
@classmethod
def get_default_success_response_type(cls) -> type[AlchemyJobPopResponse]:
    return AlchemyJobPopResponse

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