Skip to content

apimodels

API data model bases applicable across all (or many) horde APIs.

T module-attribute

T = TypeVar('T')

HordeResponseTypes module-attribute

HordeResponseTypes = (
    HordeResponseRootModel[Any] | HordeResponseBaseModel
)

A type hint for any type of the valid horde response models.

__all__ module-attribute

__all__ = [
    "APIKeyAllowedInRequestMixin",
    "ContainsMessageResponseMixin",
    "ContainsWarningsResponseMixin",
    "HordeAPIData",
    "HordeAPIMessage",
    "HordeAPIObject",
    "HordeAPIObjectBaseModel",
    "HordeRequest",
    "HordeResponse",
    "HordeResponseBaseModel",
    "HordeResponseRootModel",
    "HordeResponseTypes",
    "MessageSpecifiesUserIDMixin",
    "RequestErrorResponse",
    "RequestSingleWarning",
    "RequestUsesWorkerMixin",
    "ResponseRequiringDownloadMixin",
    "ResponseRequiringFollowUpMixin",
    "ResponseWithProgressMixin",
]

HordeAPIObject

Bases: ABC

Base class for all Horde API data models, requests, or responses.

This is an abstract class that you probably shouldn't inherit from directly. Instead, inherit from one of the subclasses defined in this module.

Requests generally would inherit from HordeRequest, responses from HordeResponse, and data models from HordeAPIObjectBaseModel (if it appears on the API as a published model) or HordeAPIDataObject (if it is a data object that is not specifically defined by the API docs, such as an intermediate class or an anonymous model).

Source code in horde_sdk/generic_api/apimodels.py
class HordeAPIObject(abc.ABC):
    """Base class for all Horde API data models, requests, or responses.

    This is an abstract class that you probably shouldn't inherit from directly. Instead, inherit from one of the
    subclasses defined in this module.

    Requests generally would inherit from `HordeRequest`, responses from `HordeResponse`, and data models from
    `HordeAPIObjectBaseModel` (if it appears on the API as a published model) or `HordeAPIDataObject` (if it is a
    data object that is not specifically defined by the API docs, such as an intermediate class or an anonymous model).

    """

    @classmethod
    @abc.abstractmethod
    def get_api_model_name(cls) -> str | None:
        """Return the name of the model as seen in the published swagger doc.

        If none, there is no payload, such as for a GET request.
        """

    @classmethod
    def get_sensitive_fields(cls) -> set[str]:
        """Return a set of fields which should be redacted from logs."""
        return {"apikey"}

    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()

    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
        }

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeAPIObjectBaseModel

Bases: HordeAPIObject, BaseModel

Base class for all Horde API data models (leveraging pydantic).

Source code in horde_sdk/generic_api/apimodels.py
class HordeAPIObjectBaseModel(HordeAPIObject, BaseModel):
    """Base class for all Horde API data models (leveraging pydantic)."""

    model_config = get_default_frozen_model_config_dict()

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeAPIData

Bases: BaseModel

Base class for all Horde API data models which appear as objects within other data models.

These are objects which are not specifically defined by the API docs, but (logically or otherwise) are returned by the API. Occasionally, objects derived from this class may also be used as a mixin to compose other models.

Source code in horde_sdk/generic_api/apimodels.py
class HordeAPIData(BaseModel):
    """Base class for all Horde API data models which appear as objects within other data models.

    These are objects which are not specifically defined by the API docs, but (logically or otherwise) are
    returned by the API. Occasionally, objects derived from this class may also be used as a mixin to compose other
    models.
    """

    model_config = get_default_frozen_model_config_dict()

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

HordeAPIMessage

Bases: HordeAPIObject

Represents any request or response from any Horde API.

Source code in horde_sdk/generic_api/apimodels.py
class HordeAPIMessage(HordeAPIObject):
    """Represents any request or response from any Horde API."""

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeResponse

Bases: HordeAPIMessage

Represents any response from any Horde API.

Source code in horde_sdk/generic_api/apimodels.py
class HordeResponse(HordeAPIMessage):
    """Represents any response from any Horde API."""

    _time_constructed: float = PrivateAttr(default_factory=lambda: time.time())

    @property
    def time_constructed(self) -> float:
        """The time the model was constructed (in epoch time)."""
        return self._time_constructed

time_constructed property

time_constructed: float

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

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeResponseRootModel

Bases: RootModel[T], HordeResponse

Base class for all Horde API response data models which model another data type (leveraging pydantic).

A typical example for using this class would be responses which are a list of another data type. Define subclasses of this class with the type of the data model as the type argument.

For example:

class MyDataRootModel(HordeResponseRootModel[MyData]):
    pass

Source code in horde_sdk/generic_api/apimodels.py
class HordeResponseRootModel(RootModel[T], HordeResponse):
    """Base class for all Horde API response data models which model another data type (leveraging pydantic).

    A typical example for using this class would be responses which *are* a list of another data type.
    Define subclasses of this class with the type of the data model as the type argument.

    For example:
    ```python
    class MyDataRootModel(HordeResponseRootModel[MyData]):
        pass
    ```

    """

    model_config = ConfigDict(
        frozen=True,
        use_attribute_docstrings=True,
        # `extra` is not allowed with RootModel
    )

model_config class-attribute instance-attribute

model_config = ConfigDict(
    frozen=True, use_attribute_docstrings=True
)

time_constructed property

time_constructed: float

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

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeResponseBaseModel

Bases: HordeResponse, BaseModel

Base class for all Horde API response data models (leveraging pydantic).

Source code in horde_sdk/generic_api/apimodels.py
class HordeResponseBaseModel(HordeResponse, BaseModel):
    """Base class for all Horde API response data models (leveraging pydantic)."""

    model_config = get_default_frozen_model_config_dict()

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

time_constructed property

time_constructed: float

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

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

ResponseRequiringFollowUpMixin

Bases: ABC

Represents any response from any Horde API which requires a follow up request of some kind.

Source code in horde_sdk/generic_api/apimodels.py
class ResponseRequiringFollowUpMixin(abc.ABC):
    """Represents any response from any Horde API which requires a follow up request of some kind."""

    @abc.abstractmethod
    def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, Any]]:
        """Return the information required from this response to submit a follow up request.

        Note that this dict uses the alias field names (as seen on the API), not the python field names.
        GenerationIDs will be returned as `{"id": "00000000-0000-0000-0000-000000000000"}` instead of
        `{"id_": "00000000-0000-0000-0000-000000000000"}`.

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

    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?

    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()}]

    @classmethod
    @abc.abstractmethod
    def get_follow_up_default_request_type(cls) -> type[HordeRequest]:
        """Return the default request type for this response."""

    @classmethod
    @abc.abstractmethod
    def get_follow_up_failure_cleanup_request_type(cls) -> type[HordeRequest]:
        """Return the request type for this response to clean up after a failed follow up request.

        Defaults to `None`, meaning no cleanup request is needed.
        """

    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 {}

    _cleanup_requests: list[HordeRequest] | None

    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

    @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()]

    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

    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_follow_up_returned_params abstractmethod

get_follow_up_returned_params(
    *, as_python_field_name: bool = False
) -> list[dict[str, Any]]

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

Note that this dict uses the alias field names (as seen on the API), not the python field names. GenerationIDs will be returned as {"id": "00000000-0000-0000-0000-000000000000"} instead of {"id_": "00000000-0000-0000-0000-000000000000"}.

Returns:

  • list[dict[str, Any]] –

    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
@abc.abstractmethod
def get_follow_up_returned_params(self, *, as_python_field_name: bool = False) -> list[dict[str, Any]]:
    """Return the information required from this response to submit a follow up request.

    Note that this dict uses the alias field names (as seen on the API), not the python field names.
    GenerationIDs will be returned as `{"id": "00000000-0000-0000-0000-000000000000"}` instead of
    `{"id_": "00000000-0000-0000-0000-000000000000"}`.

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

get_follow_up_extra_params

get_follow_up_extra_params() -> dict[str, object]

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

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

get_follow_up_all_params

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

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

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

get_follow_up_failure_cleanup_params is not included.

This is used for context management.

Returns:

  • list[dict[str, object]] –

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

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

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

    `get_follow_up_failure_cleanup_params` is **not** included.

    This is used for context management.

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

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

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

get_follow_up_default_request_type abstractmethod classmethod

get_follow_up_default_request_type() -> type[HordeRequest]

Return the default request type for this response.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_follow_up_default_request_type(cls) -> type[HordeRequest]:
    """Return the default request type for this response."""

get_follow_up_failure_cleanup_request_type abstractmethod classmethod

get_follow_up_failure_cleanup_request_type() -> (
    type[HordeRequest]
)

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

Defaults to None, meaning no cleanup request is needed.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_follow_up_failure_cleanup_request_type(cls) -> type[HordeRequest]:
    """Return the request type for this response to clean up after a failed follow up request.

    Defaults to `None`, meaning no cleanup request is needed.
    """

get_follow_up_failure_cleanup_params

get_follow_up_failure_cleanup_params() -> dict[str, object]

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

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

This is used for context management.

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

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

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

get_follow_up_failure_cleanup_request

get_follow_up_failure_cleanup_request() -> (
    list[HordeRequest]
)

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

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

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

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

    self._cleanup_requests = []

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

    return self._cleanup_requests

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
@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()]

ignore_failure

ignore_failure() -> bool

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

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

does_target_request_follow_up

does_target_request_follow_up(
    target_request: HordeRequest,
) -> bool

Return whether the target_request would follow up on this request.

Parameters:

  • target_request (HordeRequest) –

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

Returns:

  • bool ( bool ) –

    Whether the target_request would follow up on this request.

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

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

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

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

ResponseWithProgressMixin

Bases: HordeAPIData

Represents any response from any Horde API which contains progress information.

Source code in horde_sdk/generic_api/apimodels.py
class ResponseWithProgressMixin(HordeAPIData):
    """Represents any response from any Horde API which contains progress information."""

    @abc.abstractmethod
    def is_job_complete(self, number_of_result_expected: int) -> bool:
        """Return whether the job is complete.

        Args:
            number_of_result_expected (int): The number of results expected from the job.

        Returns:
            bool: Whether the job is complete.
        """

    @abc.abstractmethod
    def is_job_possible(self) -> bool:
        """Return whether the job is possible.

        Returns:
            bool: Whether the job is possible.
        """

    @classmethod
    def is_final_follow_up(cls) -> bool:
        """Return whether this response is the final follow up response."""
        return False

    @classmethod
    @abc.abstractmethod
    def get_finalize_success_request_type(cls) -> type[HordeRequest] | None:
        """Return the request type for this response to finalize the job on success, or `None` if not needed."""

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

is_job_complete abstractmethod

is_job_complete(number_of_result_expected: int) -> bool

Return whether the job is complete.

Parameters:

  • number_of_result_expected (int) –

    The number of results expected from the job.

Returns:

  • bool ( bool ) –

    Whether the job is complete.

Source code in horde_sdk/generic_api/apimodels.py
@abc.abstractmethod
def is_job_complete(self, number_of_result_expected: int) -> bool:
    """Return whether the job is complete.

    Args:
        number_of_result_expected (int): The number of results expected from the job.

    Returns:
        bool: Whether the job is complete.
    """

is_job_possible abstractmethod

is_job_possible() -> bool

Return whether the job is possible.

Returns:

  • bool ( bool ) –

    Whether the job is possible.

Source code in horde_sdk/generic_api/apimodels.py
@abc.abstractmethod
def is_job_possible(self) -> bool:
    """Return whether the job is possible.

    Returns:
        bool: Whether the job is possible.
    """

is_final_follow_up classmethod

is_final_follow_up() -> bool

Return whether this response is the final follow up response.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
def is_final_follow_up(cls) -> bool:
    """Return whether this response is the final follow up response."""
    return False

get_finalize_success_request_type abstractmethod classmethod

get_finalize_success_request_type() -> (
    type[HordeRequest] | None
)

Return the request type for this response to finalize the job on success, or None if not needed.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_finalize_success_request_type(cls) -> type[HordeRequest] | None:
    """Return the request type for this response to finalize the job on success, or `None` if not needed."""

ResponseRequiringDownloadMixin

Bases: HordeAPIData

Represents any response which may require downloading additional data.

Source code in horde_sdk/generic_api/apimodels.py
class ResponseRequiringDownloadMixin(HordeAPIData):
    """Represents any response which may require downloading additional data."""

    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")

    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"))

    @abc.abstractmethod
    async def async_download_additional_data(self, client_session: aiohttp.ClientSession) -> None:
        """Asynchronously download any additional data required for this response."""

    @abc.abstractmethod
    def download_additional_data(self) -> None:
        """Download any additional data required for this response."""

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

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
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
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
@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
@abc.abstractmethod
def download_additional_data(self) -> None:
    """Download any additional data required for this response."""

ContainsMessageResponseMixin

Bases: HordeAPIData

Represents any response from any Horde API which contains a message.

Source code in horde_sdk/generic_api/apimodels.py
class ContainsMessageResponseMixin(HordeAPIData):
    """Represents any response from any Horde API which contains a message."""

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

message class-attribute instance-attribute

message: str = ''

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

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

RequestSingleWarning

Bases: HordeAPIObjectBaseModel

Represents a single warning from the API.

Source code in horde_sdk/generic_api/apimodels.py
class RequestSingleWarning(HordeAPIObjectBaseModel):
    """Represents a single warning from the API."""

    message: str | None = None
    """A message from the API. This is typically an error or warning message, but may also be informational."""

    code: str | None = None
    """The code associated with this warning."""

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

message class-attribute instance-attribute

message: str | None = None

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

code class-attribute instance-attribute

code: str | None = None

The code associated with this warning.

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/generic_api/apimodels.py
@override
@classmethod
def get_api_model_name(cls) -> str | None:
    return "RequestSingleWarning"

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

ContainsWarningsResponseMixin

Bases: HordeAPIData

Represents any response from any Horde API which contains warnings.

Source code in horde_sdk/generic_api/apimodels.py
@Unhashable
@Unequatable
class ContainsWarningsResponseMixin(HordeAPIData):
    """Represents any response from any Horde API which contains warnings."""

    warnings: list[RequestSingleWarning] | None = None
    """A list of warnings from the API. This is typically an error or warning message,
    but may also be informational."""

warnings class-attribute instance-attribute

warnings: list[RequestSingleWarning] | None = None

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

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

RequestErrorResponse

Bases: HordeResponseBaseModel, ContainsMessageResponseMixin

The catch all error response for any request to any Horde API.

v2 API Model: RequestError

Source code in horde_sdk/generic_api/apimodels.py
class RequestErrorResponse(HordeResponseBaseModel, ContainsMessageResponseMixin):
    """The catch all error response for any request to any Horde API.

    v2 API Model: `RequestError`
    """

    object_data: object = None
    """This is a catch all for any additional data that may be returned by the API relevant to the error."""

    rc: str = "RC_MISSING"
    """The return code from the API which maps to a reason for the error."""

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

object_data class-attribute instance-attribute

object_data: object = None

This is a catch all for any additional data that may be returned by the API relevant to the error.

rc class-attribute instance-attribute

rc: str = 'RC_MISSING'

The return code from the API which maps to a reason for the error.

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

message class-attribute instance-attribute

message: str = ''

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

time_constructed property

time_constructed: float

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

get_api_model_name classmethod

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

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

HordeRequest

Bases: HordeAPIMessage, BaseModel

Represents any request to any Horde API.

Source code in horde_sdk/generic_api/apimodels.py
class HordeRequest(HordeAPIMessage, BaseModel):
    """Represents any request to any Horde API."""

    model_config = get_default_frozen_model_config_dict()

    @classmethod
    @abc.abstractmethod
    def get_http_method(cls) -> HTTPMethod:
        """Return the HTTP method (verb) this request uses."""

    accept: GenericAcceptTypes = GenericAcceptTypes.json
    """The 'accept' header field."""
    # X_Fields # TODO

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

    @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())

    @classmethod
    @abc.abstractmethod
    def get_api_url(cls) -> str:
        """Return the base URL for the API this request is for."""

    @classmethod
    @abc.abstractmethod
    def get_api_endpoint_subpath(cls) -> GENERIC_API_ENDPOINT_SUBPATH:
        """Return the subpath to the specific API action defined by this object."""

    @classmethod
    @abc.abstractmethod
    def get_default_success_response_type(cls) -> type[HordeResponseTypes]:
        """Return the `type` of the response expected in the ordinary case of success."""

    @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()}

    @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 []

    @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 []

    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

    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

    @override
    @classmethod
    def get_sensitive_fields(cls) -> set[str]:
        return {"apikey"}

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

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_http_method abstractmethod classmethod

get_http_method() -> HTTPMethod

Return the HTTP method (verb) this request uses.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_http_method(cls) -> HTTPMethod:
    """Return the HTTP method (verb) this request uses."""

get_api_endpoint_url classmethod

get_api_endpoint_url() -> str

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

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

get_api_url abstractmethod classmethod

get_api_url() -> str

Return the base URL for the API this request is for.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_url(cls) -> str:
    """Return the base URL for the API this request is for."""

get_api_endpoint_subpath abstractmethod classmethod

get_api_endpoint_subpath() -> GENERIC_API_ENDPOINT_SUBPATH

Return the subpath to the specific API action defined by this object.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_endpoint_subpath(cls) -> GENERIC_API_ENDPOINT_SUBPATH:
    """Return the subpath to the specific API action defined by this object."""

get_default_success_response_type abstractmethod classmethod

get_default_success_response_type() -> (
    type[HordeResponseTypes]
)

Return the type of the response expected in the ordinary case of success.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_default_success_response_type(cls) -> type[HordeResponseTypes]:
    """Return the `type` of the response expected in the ordinary case of success."""

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
@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
@classmethod
def get_header_fields(cls) -> list[str]:
    """Return a list of field names from this request object that should be sent as header fields.

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

get_query_fields classmethod

get_query_fields() -> list[str]

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

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

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

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

get_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
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
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

get_sensitive_fields classmethod

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

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

APIKeyAllowedInRequestMixin

Bases: HordeAPIObjectBaseModel

Mix-in class to describe an endpoint which may require authentication.

Source code in horde_sdk/generic_api/apimodels.py
class APIKeyAllowedInRequestMixin(HordeAPIObjectBaseModel):
    """Mix-in class to describe an endpoint which may require authentication."""

    apikey: str | None = None
    """Defaults to `ANON_API_KEY`. See also `.is_api_key_required()`"""

    @classmethod
    def is_api_key_required(cls) -> bool:
        """Return whether this endpoint requires an API key."""
        return True

    @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

apikey class-attribute instance-attribute

apikey: str | None = None

Defaults to ANON_API_KEY. See also .is_api_key_required()

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

is_api_key_required classmethod

is_api_key_required() -> bool

Return whether this endpoint requires an API key.

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

validate_api_key_length

validate_api_key_length(v: str) -> str

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

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

get_api_model_name abstractmethod classmethod

get_api_model_name() -> str | None

Return the name of the model as seen in the published swagger doc.

If none, there is no payload, such as for a GET request.

Source code in horde_sdk/generic_api/apimodels.py
@classmethod
@abc.abstractmethod
def get_api_model_name(cls) -> str | None:
    """Return the name of the model as seen in the published swagger doc.

    If none, there is no payload, such as for a GET request.
    """

get_sensitive_fields classmethod

get_sensitive_fields() -> set[str]

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

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

get_extra_fields_to_exclude_from_log

get_extra_fields_to_exclude_from_log() -> set[str]

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

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

log_safe_model_dump

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

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

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

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

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

MessageSpecifiesUserIDMixin

Bases: HordeAPIData

Mix-in class to describe an endpoint for which you can specify a user.

Source code in horde_sdk/generic_api/apimodels.py
class MessageSpecifiesUserIDMixin(HordeAPIData):
    """Mix-in class to describe an endpoint for which you can specify a user."""

    user_id: str
    """The user's ID, as a `str`, but only containing numeric values."""

    @field_validator("user_id")
    def user_id_is_numeric(cls, value: str) -> str:
        """Check if the ID is a numeric string.

        The API endpoint expects a string, but the only valid values would be numbers only.
        """
        if not value.isnumeric():
            raise ValueError("user_id must only contain numeric values")
        return value

user_id instance-attribute

user_id: str

The user's ID, as a str, but only containing numeric values.

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()

user_id_is_numeric

user_id_is_numeric(value: str) -> str

Check if the ID is a numeric string.

The API endpoint expects a string, but the only valid values would be numbers only.

Source code in horde_sdk/generic_api/apimodels.py
@field_validator("user_id")
def user_id_is_numeric(cls, value: str) -> str:
    """Check if the ID is a numeric string.

    The API endpoint expects a string, but the only valid values would be numbers only.
    """
    if not value.isnumeric():
        raise ValueError("user_id must only contain numeric values")
    return value

RequestUsesWorkerMixin

Bases: HordeAPIData

Mix-in class to describe an endpoint for which you can specify workers.

Source code in horde_sdk/generic_api/apimodels.py
class RequestUsesWorkerMixin(HordeAPIData):
    """Mix-in class to describe an endpoint for which you can specify workers."""

    trusted_workers: bool = False
    """When true, only trusted workers will serve this request. When False, Evaluating workers will also be used
     which can increase speed but adds more risk!"""
    slow_workers: bool = True
    """When True, allows slower workers to pick up this request. Disabling this incurs an extra kudos cost."""
    extra_slow_workers: bool = False
    """When True, allows extra slow workers to pick up this request."""
    workers: list[str] = Field(default_factory=list)
    """A list of worker IDs to use for this request. If empty, any worker can pick up the request. Using this incurs
    and extra kudos cost."""
    worker_blacklist: bool = False
    """If true, the worker list will be treated as a blacklist instead of a whitelist."""
    models: list[str]
    """The generative models to use for this request."""

    validated_backends: bool | None = Field(default=None)
    """When true, only inference backends that are validated by the AI Horde devs will serve this request. When False,
    non-validated backends will also be used which can increase speed but you may end up with unexpected results."""
    style: str | None = Field(default=None, examples=["00000000-0000-0000-0000-000000000000"])
    """The style ID to use for the generation."""

    dry_run: bool = False
    """If true, the request will not be processed, but will return a response with the estimated kudos cost."""
    webhook: str | None = Field(default=None)
    """Provide a URL where the AI Horde will send a POST call after each delivered generation.
    The request will include the details of the job as well as the request ID."""
    allow_downgrade: bool = False
    """If true, the request can be modified to lesser parameters if the original request is too expensive."""
    proxied_account: str | None = Field(default=None)
    """The account this request is being proxied for. This requires the sending API to be a service account."""
    disable_batching: bool | None = Field(False)
    """When true, This request will not use batching. This will allow you to retrieve accurate seeds.
    Feature is restricted to Trusted users and Patreons."""

trusted_workers class-attribute instance-attribute

trusted_workers: bool = False

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

slow_workers class-attribute instance-attribute

slow_workers: bool = True

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

extra_slow_workers class-attribute instance-attribute

extra_slow_workers: bool = False

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

workers class-attribute instance-attribute

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

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

worker_blacklist class-attribute instance-attribute

worker_blacklist: bool = False

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

models instance-attribute

models: list[str]

The generative models to use for this request.

validated_backends class-attribute instance-attribute

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

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

style class-attribute instance-attribute

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

The style ID to use for the generation.

dry_run class-attribute instance-attribute

dry_run: bool = False

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

webhook class-attribute instance-attribute

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

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

allow_downgrade class-attribute instance-attribute

allow_downgrade: bool = False

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

proxied_account class-attribute instance-attribute

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

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

disable_batching class-attribute instance-attribute

disable_batching: bool | None = Field(False)

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

model_config class-attribute instance-attribute

model_config = get_default_frozen_model_config_dict()