Skip to content

jobs

DEFAULT_UPLOAD_METHOD module-attribute 

DEFAULT_UPLOAD_METHOD = PUT

ImageWorkerJob

Bases: HordeWorkerJob[ImageSingleGeneration, ImageGenerationParameters]

A job containing only image generations.

Source code in horde_sdk/worker/jobs.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class ImageWorkerJob(HordeWorkerJob[ImageSingleGeneration, ImageGenerationParameters]):
    """A job containing only image generations."""

    def __init__(
        self,
        *,
        generation: ImageSingleGeneration,
        job_config: HordeWorkerJobConfig | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        preserve_generation_id: bool = False,
    ) -> None:
        """Initialize the image worker job.

        Args:
            generation (ImageSingleGeneration): The generation to use for the job.
            job_config (HordeWorkerJobConfig | None): The configuration for the job.
            job_config (HordeWorkerJobConfig, optional): The configuration for the job. If `None`, the default \
                configuration will be used. Defaults to None.
            job_id (ID_TYPES | None): Optional identifier to associate with the job.
            dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
            dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
                generation.
            preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
                job identifier.
        """
        super().__init__(
            generation=generation,
            generation_cls=ImageSingleGeneration,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

    @override
    @classmethod
    def job_worker_type(cls) -> WORKER_TYPE:
        return WORKER_TYPE.image

    @classmethod
    def from_template(
        cls,
        template: ImageGenerationParametersTemplate,
        *,
        generation_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        base_param_updates: BasicImageGenerationParametersTemplate | None = None,
        additional_param_updates: ImageGenerationComponentContainer | None = None,
        result_ids: Sequence[ID_TYPES] | None = None,
        allocator: ResultIdAllocator | None = None,
        seed: str = "image",
        generation_kwargs: ImageGenerationInitKwargs | None = None,
        job_config: HordeWorkerJobConfig | None = None,
        preserve_generation_id: bool = False,
    ) -> ImageWorkerJob:
        """Instantiate an image job from a template."""
        generation_parameters = template.to_parameters(
            base_param_updates=base_param_updates,
            additional_param_updates=additional_param_updates,
            result_ids=result_ids,
            allocator=allocator,
            seed=seed,
        )
        init_kwargs: ImageGenerationInitKwargs = {}
        if generation_kwargs:
            init_kwargs.update(generation_kwargs)
        if generation_id is not None:
            init_kwargs.setdefault("generation_id", generation_id)
        if dispatch_result_ids is not None:
            init_kwargs.setdefault("dispatch_result_ids", list(dispatch_result_ids))
        init_kwargs.setdefault("result_ids", generation_parameters.result_ids)
        generation = ImageSingleGeneration(
            generation_parameters=generation_parameters,
            **init_kwargs,
        )
        return cls(
            generation=generation,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

generation_cls property 

generation_cls: type[SingleGenerationTypeVar]

The (python) type created by the job.

generation_parameters_cls property 

generation_parameters_cls: type[ComposedParameterSetTypeVar]

The (python) type of the generation parameters.

generation property 

generation: SingleGenerationTypeVar

The individual generations in this job.

job_config property 

job_config: HordeWorkerJobConfig

Return the configuration associated with this job.

job_id property 

job_id: ID_TYPES

Return the identifier assigned to this job.

local_job_id property 

local_job_id: ID_TYPES

Alias for :meth:job_id to emphasize local scope.

dispatch_job_id property 

dispatch_job_id: ID_TYPES | None

Return the identifier provided by the dispatch source, if any.

time_received property 

time_received: float | None

The time the job response was either received or constructed (in epoch time).

Note: This generally will be the time the job was popped from the server. However, manually constructed api responses or jobs that are not popped from a queue may imbue this property with a different meaning.

You can manually set this value with the time_received parameter in the constructor.

time_since_received property 

time_since_received: float | None

The time since the job was popped from the queue in seconds, or None if not yet received.

time_submitted property 

time_submitted: float | None

The time the job was submitted to the API in epoch time or None if not submitted.

time_spent_generating class-attribute instance-attribute 

time_spent_generating: float = 0.0

The time spent generating the job in seconds.

time_to_download_aux_models class-attribute instance-attribute 

time_to_download_aux_models: float = 0.0

The time spent downloading user-specified auxiliary models specific to the job (i.e., LoRas) in seconds.

job_identifier_string property 

job_identifier_string: str

Returns a string that identifies the job.

faulted_reason property 

faulted_reason: WORKER_ERRORS | None

The reason the job was faulted or None if not faulted.

faulted_at property 

faulted_at: float | None

The time the job was faulted in epoch time or None if not faulted.

is_faulted property 

is_faulted: bool

Whether or not the job has been marked as faulted.

should_censor_nsfw property 

should_censor_nsfw: bool

Whether or not the user has requested that NSFW content be censored.

is_job_finalized property 

is_job_finalized: bool

Return true if the generation in the job is finalized.

Note: This means the generation has been submitted as either successful or failed, or has been abandoned. Accordingly, there is nothing more to do with the job.

job_completed_successfully property 

job_completed_successfully: bool

Return true if the generation in the job completed successfully.

__init__ 

__init__(
    *,
    generation: ImageSingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False
) -> None

Initialize the image worker job.

Parameters:

  • generation (ImageSingleGeneration) –

    The generation to use for the job.

  • job_config (HordeWorkerJobConfig | None, default: None ) –

    The configuration for the job.

  • job_config (HordeWorkerJobConfig, default: None ) –

    The configuration for the job. If None, the default configuration will be used. Defaults to None.

  • job_id (ID_TYPES | None, default: None ) –

    Optional identifier to associate with the job.

  • dispatch_job_id (ID_TYPES | None, default: None ) –

    Identifier supplied by dispatch for the job.

  • dispatch_result_ids (Sequence[ID_TYPES] | None, default: None ) –

    Result identifiers supplied by dispatch for the generation.

  • preserve_generation_id (bool, default: False ) –

    Retain the existing generation identifier instead of overwriting it with the job identifier.

Source code in horde_sdk/worker/jobs.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    *,
    generation: ImageSingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False,
) -> None:
    """Initialize the image worker job.

    Args:
        generation (ImageSingleGeneration): The generation to use for the job.
        job_config (HordeWorkerJobConfig | None): The configuration for the job.
        job_config (HordeWorkerJobConfig, optional): The configuration for the job. If `None`, the default \
            configuration will be used. Defaults to None.
        job_id (ID_TYPES | None): Optional identifier to associate with the job.
        dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
        dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
            generation.
        preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
            job identifier.
    """
    super().__init__(
        generation=generation,
        generation_cls=ImageSingleGeneration,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

job_worker_type classmethod 

job_worker_type() -> WORKER_TYPE
Source code in horde_sdk/worker/jobs.py 
78
79
80
81
@override
@classmethod
def job_worker_type(cls) -> WORKER_TYPE:
    return WORKER_TYPE.image

from_template classmethod 

from_template(
    template: ImageGenerationParametersTemplate,
    *,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    base_param_updates: (
        BasicImageGenerationParametersTemplate | None
    ) = None,
    additional_param_updates: (
        ImageGenerationComponentContainer | None
    ) = None,
    result_ids: Sequence[ID_TYPES] | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "image",
    generation_kwargs: (
        ImageGenerationInitKwargs | None
    ) = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False
) -> ImageWorkerJob

Instantiate an image job from a template.

Source code in horde_sdk/worker/jobs.py 
 83
 84
 85
 86
 87
 88
 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
@classmethod
def from_template(
    cls,
    template: ImageGenerationParametersTemplate,
    *,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    base_param_updates: BasicImageGenerationParametersTemplate | None = None,
    additional_param_updates: ImageGenerationComponentContainer | None = None,
    result_ids: Sequence[ID_TYPES] | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "image",
    generation_kwargs: ImageGenerationInitKwargs | None = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False,
) -> ImageWorkerJob:
    """Instantiate an image job from a template."""
    generation_parameters = template.to_parameters(
        base_param_updates=base_param_updates,
        additional_param_updates=additional_param_updates,
        result_ids=result_ids,
        allocator=allocator,
        seed=seed,
    )
    init_kwargs: ImageGenerationInitKwargs = {}
    if generation_kwargs:
        init_kwargs.update(generation_kwargs)
    if generation_id is not None:
        init_kwargs.setdefault("generation_id", generation_id)
    if dispatch_result_ids is not None:
        init_kwargs.setdefault("dispatch_result_ids", list(dispatch_result_ids))
    init_kwargs.setdefault("result_ids", generation_parameters.result_ids)
    generation = ImageSingleGeneration(
        generation_parameters=generation_parameters,
        **init_kwargs,
    )
    return cls(
        generation=generation,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

set_dispatch_job_id 

set_dispatch_job_id(
    dispatch_job_id: ID_TYPES | None,
) -> None

Bind the job to the identifier supplied by dispatch.

Source code in horde_sdk/worker/job_base.py 
222
223
224
225
def set_dispatch_job_id(self, dispatch_job_id: ID_TYPES | None) -> None:
    """Bind the job to the identifier supplied by dispatch."""
    with self._lock:
        self._dispatch_job_id = dispatch_job_id

set_job_faulted 

set_job_faulted(
    faulted_reason: WORKER_ERRORS,
    failure_exception: Exception | None = None,
) -> None

Mark the entire job as faulted.

Note: This will mark all generations in the job as faulted.

Source code in horde_sdk/worker/job_base.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def set_job_faulted(self, faulted_reason: WORKER_ERRORS, failure_exception: Exception | None = None) -> None:
    """Mark the entire job as faulted.

    Note: This will mark all generations in the job as faulted.
    """
    with self._lock:
        if self._faulted:
            logger.warning(
                f"Job {self.job_identifier_string} is already marked faulted with "
                f"reason {self._fault_reason} at {self._faulted_at}",
            )

        self._faulted = True
        self._fault_reason = faulted_reason
        self._faulted_at = time.time()

        self.generation.on_abort(
            failed_message=faulted_reason,
            failure_exception=failure_exception,
        )

AlchemyWorkerJob

Bases: HordeWorkerJob[AlchemySingleGeneration, SingleAlchemyParameters]

A job containing only alchemy generations.

Source code in horde_sdk/worker/jobs.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class AlchemyWorkerJob(HordeWorkerJob[AlchemySingleGeneration, SingleAlchemyParameters]):
    """A job containing only alchemy generations."""

    def __init__(
        self,
        *,
        generation: AlchemySingleGeneration,
        job_config: HordeWorkerJobConfig | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        preserve_generation_id: bool = False,
    ) -> None:
        """Initialize the alchemy worker job.

        Args:
            generation (AlchemySingleGeneration): The response from the API.
            job_config (HordeWorkerJobConfig | None, optional): The configuration for the job. If `None`, the default \
                configuration will be used. Defaults to None.
            job_id (ID_TYPES | None): Optional identifier to associate with the job.
            dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
            dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
                generation.
            preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
                job identifier.
        """
        super().__init__(
            generation=generation,
            generation_cls=AlchemySingleGeneration,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

    @override
    @classmethod
    def job_worker_type(cls) -> WORKER_TYPE:
        return WORKER_TYPE.alchemist

    @classmethod
    def from_template(
        cls,
        template: SingleAlchemyParametersTemplate,
        *,
        source_image: bytes | str | None = None,
        default_form: KNOWN_ALCHEMY_FORMS | str | None = KNOWN_ALCHEMY_FORMS.post_process,
        generation_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        result_id: ID_TYPES | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        allocator: ResultIdAllocator | None = None,
        seed: str = "alchemy",
        generation_kwargs: AlchemyGenerationInitKwargs | None = None,
        job_config: HordeWorkerJobConfig | None = None,
        preserve_generation_id: bool = False,
    ) -> AlchemyWorkerJob:
        """Instantiate an alchemy job from a template."""
        generation_parameters = template.to_parameters(
            result_id=result_id,
            source_image=source_image,
            default_form=default_form,
            allocator=allocator,
            seed=seed,
        )
        init_kwargs: AlchemyGenerationInitKwargs = {}
        if generation_kwargs:
            init_kwargs.update(generation_kwargs)
        if generation_id is not None:
            init_kwargs.setdefault("generation_id", _stringify_id(generation_id))
        if dispatch_result_ids is not None:
            init_kwargs.setdefault(
                "dispatch_result_ids",
                [
                    stringified
                    for identifier in dispatch_result_ids
                    if (stringified := _stringify_id(identifier)) is not None
                ],
            )
        init_kwargs.setdefault("result_ids", [generation_parameters.result_id])
        generation = AlchemySingleGeneration(
            generation_parameters=generation_parameters,
            **init_kwargs,
        )
        return cls(
            generation=generation,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

generation_cls property 

generation_cls: type[SingleGenerationTypeVar]

The (python) type created by the job.

generation_parameters_cls property 

generation_parameters_cls: type[ComposedParameterSetTypeVar]

The (python) type of the generation parameters.

generation property 

generation: SingleGenerationTypeVar

The individual generations in this job.

job_config property 

job_config: HordeWorkerJobConfig

Return the configuration associated with this job.

job_id property 

job_id: ID_TYPES

Return the identifier assigned to this job.

local_job_id property 

local_job_id: ID_TYPES

Alias for :meth:job_id to emphasize local scope.

dispatch_job_id property 

dispatch_job_id: ID_TYPES | None

Return the identifier provided by the dispatch source, if any.

time_received property 

time_received: float | None

The time the job response was either received or constructed (in epoch time).

Note: This generally will be the time the job was popped from the server. However, manually constructed api responses or jobs that are not popped from a queue may imbue this property with a different meaning.

You can manually set this value with the time_received parameter in the constructor.

time_since_received property 

time_since_received: float | None

The time since the job was popped from the queue in seconds, or None if not yet received.

time_submitted property 

time_submitted: float | None

The time the job was submitted to the API in epoch time or None if not submitted.

time_spent_generating class-attribute instance-attribute 

time_spent_generating: float = 0.0

The time spent generating the job in seconds.

time_to_download_aux_models class-attribute instance-attribute 

time_to_download_aux_models: float = 0.0

The time spent downloading user-specified auxiliary models specific to the job (i.e., LoRas) in seconds.

job_identifier_string property 

job_identifier_string: str

Returns a string that identifies the job.

faulted_reason property 

faulted_reason: WORKER_ERRORS | None

The reason the job was faulted or None if not faulted.

faulted_at property 

faulted_at: float | None

The time the job was faulted in epoch time or None if not faulted.

is_faulted property 

is_faulted: bool

Whether or not the job has been marked as faulted.

should_censor_nsfw property 

should_censor_nsfw: bool

Whether or not the user has requested that NSFW content be censored.

is_job_finalized property 

is_job_finalized: bool

Return true if the generation in the job is finalized.

Note: This means the generation has been submitted as either successful or failed, or has been abandoned. Accordingly, there is nothing more to do with the job.

job_completed_successfully property 

job_completed_successfully: bool

Return true if the generation in the job completed successfully.

__init__ 

__init__(
    *,
    generation: AlchemySingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False
) -> None

Initialize the alchemy worker job.

Parameters:

  • generation (AlchemySingleGeneration) –

    The response from the API.

  • job_config (HordeWorkerJobConfig | None, default: None ) –

    The configuration for the job. If None, the default configuration will be used. Defaults to None.

  • job_id (ID_TYPES | None, default: None ) –

    Optional identifier to associate with the job.

  • dispatch_job_id (ID_TYPES | None, default: None ) –

    Identifier supplied by dispatch for the job.

  • dispatch_result_ids (Sequence[ID_TYPES] | None, default: None ) –

    Result identifiers supplied by dispatch for the generation.

  • preserve_generation_id (bool, default: False ) –

    Retain the existing generation identifier instead of overwriting it with the job identifier.

Source code in horde_sdk/worker/jobs.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def __init__(
    self,
    *,
    generation: AlchemySingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False,
) -> None:
    """Initialize the alchemy worker job.

    Args:
        generation (AlchemySingleGeneration): The response from the API.
        job_config (HordeWorkerJobConfig | None, optional): The configuration for the job. If `None`, the default \
            configuration will be used. Defaults to None.
        job_id (ID_TYPES | None): Optional identifier to associate with the job.
        dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
        dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
            generation.
        preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
            job identifier.
    """
    super().__init__(
        generation=generation,
        generation_cls=AlchemySingleGeneration,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

job_worker_type classmethod 

job_worker_type() -> WORKER_TYPE
Source code in horde_sdk/worker/jobs.py 
167
168
169
170
@override
@classmethod
def job_worker_type(cls) -> WORKER_TYPE:
    return WORKER_TYPE.alchemist

from_template classmethod 

from_template(
    template: SingleAlchemyParametersTemplate,
    *,
    source_image: bytes | str | None = None,
    default_form: (
        KNOWN_ALCHEMY_FORMS | str | None
    ) = KNOWN_ALCHEMY_FORMS.post_process,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    result_id: ID_TYPES | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "alchemy",
    generation_kwargs: (
        AlchemyGenerationInitKwargs | None
    ) = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False
) -> AlchemyWorkerJob

Instantiate an alchemy job from a template.

Source code in horde_sdk/worker/jobs.py 
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
@classmethod
def from_template(
    cls,
    template: SingleAlchemyParametersTemplate,
    *,
    source_image: bytes | str | None = None,
    default_form: KNOWN_ALCHEMY_FORMS | str | None = KNOWN_ALCHEMY_FORMS.post_process,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    result_id: ID_TYPES | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "alchemy",
    generation_kwargs: AlchemyGenerationInitKwargs | None = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False,
) -> AlchemyWorkerJob:
    """Instantiate an alchemy job from a template."""
    generation_parameters = template.to_parameters(
        result_id=result_id,
        source_image=source_image,
        default_form=default_form,
        allocator=allocator,
        seed=seed,
    )
    init_kwargs: AlchemyGenerationInitKwargs = {}
    if generation_kwargs:
        init_kwargs.update(generation_kwargs)
    if generation_id is not None:
        init_kwargs.setdefault("generation_id", _stringify_id(generation_id))
    if dispatch_result_ids is not None:
        init_kwargs.setdefault(
            "dispatch_result_ids",
            [
                stringified
                for identifier in dispatch_result_ids
                if (stringified := _stringify_id(identifier)) is not None
            ],
        )
    init_kwargs.setdefault("result_ids", [generation_parameters.result_id])
    generation = AlchemySingleGeneration(
        generation_parameters=generation_parameters,
        **init_kwargs,
    )
    return cls(
        generation=generation,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

set_dispatch_job_id 

set_dispatch_job_id(
    dispatch_job_id: ID_TYPES | None,
) -> None

Bind the job to the identifier supplied by dispatch.

Source code in horde_sdk/worker/job_base.py 
222
223
224
225
def set_dispatch_job_id(self, dispatch_job_id: ID_TYPES | None) -> None:
    """Bind the job to the identifier supplied by dispatch."""
    with self._lock:
        self._dispatch_job_id = dispatch_job_id

set_job_faulted 

set_job_faulted(
    faulted_reason: WORKER_ERRORS,
    failure_exception: Exception | None = None,
) -> None

Mark the entire job as faulted.

Note: This will mark all generations in the job as faulted.

Source code in horde_sdk/worker/job_base.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def set_job_faulted(self, faulted_reason: WORKER_ERRORS, failure_exception: Exception | None = None) -> None:
    """Mark the entire job as faulted.

    Note: This will mark all generations in the job as faulted.
    """
    with self._lock:
        if self._faulted:
            logger.warning(
                f"Job {self.job_identifier_string} is already marked faulted with "
                f"reason {self._fault_reason} at {self._faulted_at}",
            )

        self._faulted = True
        self._fault_reason = faulted_reason
        self._faulted_at = time.time()

        self.generation.on_abort(
            failed_message=faulted_reason,
            failure_exception=failure_exception,
        )

TextWorkerJob

Bases: HordeWorkerJob[TextSingleGeneration, TextGenerationParameters]

A job containing only text generations.

Source code in horde_sdk/worker/jobs.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
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
class TextWorkerJob(HordeWorkerJob[TextSingleGeneration, TextGenerationParameters]):
    """A job containing only text generations."""

    def __init__(
        self,
        generation: TextSingleGeneration,
        job_config: HordeWorkerJobConfig | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        preserve_generation_id: bool = False,
    ) -> None:
        """Initialize the text worker job.

        Args:
            generation (TextSingleGeneration): The response from the API.
            job_config (HordeWorkerJobConfig | None, optional): The configuration for the job. If `None`, the default \
                configuration will be used. Defaults to None.
            job_id (ID_TYPES | None): Optional identifier to associate with the job.
            dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
            dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
                generation.
            preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
                job identifier.
        """
        super().__init__(
            generation=generation,
            generation_cls=TextSingleGeneration,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

    @override
    @classmethod
    def job_worker_type(cls) -> WORKER_TYPE:
        return WORKER_TYPE.text

    @classmethod
    def from_template(
        cls,
        template: TextGenerationParametersTemplate,
        *,
        generation_id: ID_TYPES | None = None,
        dispatch_result_ids: Sequence[ID_TYPES] | None = None,
        job_id: ID_TYPES | None = None,
        dispatch_job_id: ID_TYPES | None = None,
        base_param_updates: BasicTextGenerationParametersTemplate | None = None,
        result_ids: Sequence[ID_TYPES] | None = None,
        allocator: ResultIdAllocator | None = None,
        seed: str = "text",
        generation_kwargs: TextGenerationInitKwargs | None = None,
        job_config: HordeWorkerJobConfig | None = None,
        preserve_generation_id: bool = False,
    ) -> TextWorkerJob:
        """Instantiate a text job from a template."""
        generation_parameters = template.to_parameters(
            base_param_updates=base_param_updates,
            result_ids=result_ids,
            allocator=allocator,
            seed=seed,
        )
        init_kwargs: TextGenerationInitKwargs = {}
        if generation_kwargs:
            init_kwargs.update(generation_kwargs)
        if generation_id is not None:
            init_kwargs.setdefault("generation_id", generation_id)
        if dispatch_result_ids is not None:
            init_kwargs.setdefault("dispatch_result_ids", list(dispatch_result_ids))
        init_kwargs.setdefault("result_ids", generation_parameters.result_ids)
        generation = TextSingleGeneration(
            generation_parameters=generation_parameters,
            **init_kwargs,
        )
        return cls(
            generation=generation,
            job_config=job_config,
            job_id=job_id,
            dispatch_job_id=dispatch_job_id,
            dispatch_result_ids=dispatch_result_ids,
            preserve_generation_id=preserve_generation_id,
        )

generation_cls property 

generation_cls: type[SingleGenerationTypeVar]

The (python) type created by the job.

generation_parameters_cls property 

generation_parameters_cls: type[ComposedParameterSetTypeVar]

The (python) type of the generation parameters.

generation property 

generation: SingleGenerationTypeVar

The individual generations in this job.

job_config property 

job_config: HordeWorkerJobConfig

Return the configuration associated with this job.

job_id property 

job_id: ID_TYPES

Return the identifier assigned to this job.

local_job_id property 

local_job_id: ID_TYPES

Alias for :meth:job_id to emphasize local scope.

dispatch_job_id property 

dispatch_job_id: ID_TYPES | None

Return the identifier provided by the dispatch source, if any.

time_received property 

time_received: float | None

The time the job response was either received or constructed (in epoch time).

Note: This generally will be the time the job was popped from the server. However, manually constructed api responses or jobs that are not popped from a queue may imbue this property with a different meaning.

You can manually set this value with the time_received parameter in the constructor.

time_since_received property 

time_since_received: float | None

The time since the job was popped from the queue in seconds, or None if not yet received.

time_submitted property 

time_submitted: float | None

The time the job was submitted to the API in epoch time or None if not submitted.

time_spent_generating class-attribute instance-attribute 

time_spent_generating: float = 0.0

The time spent generating the job in seconds.

time_to_download_aux_models class-attribute instance-attribute 

time_to_download_aux_models: float = 0.0

The time spent downloading user-specified auxiliary models specific to the job (i.e., LoRas) in seconds.

job_identifier_string property 

job_identifier_string: str

Returns a string that identifies the job.

faulted_reason property 

faulted_reason: WORKER_ERRORS | None

The reason the job was faulted or None if not faulted.

faulted_at property 

faulted_at: float | None

The time the job was faulted in epoch time or None if not faulted.

is_faulted property 

is_faulted: bool

Whether or not the job has been marked as faulted.

should_censor_nsfw property 

should_censor_nsfw: bool

Whether or not the user has requested that NSFW content be censored.

is_job_finalized property 

is_job_finalized: bool

Return true if the generation in the job is finalized.

Note: This means the generation has been submitted as either successful or failed, or has been abandoned. Accordingly, there is nothing more to do with the job.

job_completed_successfully property 

job_completed_successfully: bool

Return true if the generation in the job completed successfully.

__init__ 

__init__(
    generation: TextSingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False,
) -> None

Initialize the text worker job.

Parameters:

  • generation (TextSingleGeneration) –

    The response from the API.

  • job_config (HordeWorkerJobConfig | None, default: None ) –

    The configuration for the job. If None, the default configuration will be used. Defaults to None.

  • job_id (ID_TYPES | None, default: None ) –

    Optional identifier to associate with the job.

  • dispatch_job_id (ID_TYPES | None, default: None ) –

    Identifier supplied by dispatch for the job.

  • dispatch_result_ids (Sequence[ID_TYPES] | None, default: None ) –

    Result identifiers supplied by dispatch for the generation.

  • preserve_generation_id (bool, default: False ) –

    Retain the existing generation identifier instead of overwriting it with the job identifier.

Source code in horde_sdk/worker/jobs.py 
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
def __init__(
    self,
    generation: TextSingleGeneration,
    job_config: HordeWorkerJobConfig | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    preserve_generation_id: bool = False,
) -> None:
    """Initialize the text worker job.

    Args:
        generation (TextSingleGeneration): The response from the API.
        job_config (HordeWorkerJobConfig | None, optional): The configuration for the job. If `None`, the default \
            configuration will be used. Defaults to None.
        job_id (ID_TYPES | None): Optional identifier to associate with the job.
        dispatch_job_id (ID_TYPES | None): Identifier supplied by dispatch for the job.
        dispatch_result_ids (Sequence[ID_TYPES] | None): Result identifiers supplied by dispatch for the
            generation.
        preserve_generation_id (bool): Retain the existing generation identifier instead of overwriting it with the
            job identifier.
    """
    super().__init__(
        generation=generation,
        generation_cls=TextSingleGeneration,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

job_worker_type classmethod 

job_worker_type() -> WORKER_TYPE
Source code in horde_sdk/worker/jobs.py 
262
263
264
265
@override
@classmethod
def job_worker_type(cls) -> WORKER_TYPE:
    return WORKER_TYPE.text

from_template classmethod 

from_template(
    template: TextGenerationParametersTemplate,
    *,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    base_param_updates: (
        BasicTextGenerationParametersTemplate | None
    ) = None,
    result_ids: Sequence[ID_TYPES] | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "text",
    generation_kwargs: (
        TextGenerationInitKwargs | None
    ) = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False
) -> TextWorkerJob

Instantiate a text job from a template.

Source code in horde_sdk/worker/jobs.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
@classmethod
def from_template(
    cls,
    template: TextGenerationParametersTemplate,
    *,
    generation_id: ID_TYPES | None = None,
    dispatch_result_ids: Sequence[ID_TYPES] | None = None,
    job_id: ID_TYPES | None = None,
    dispatch_job_id: ID_TYPES | None = None,
    base_param_updates: BasicTextGenerationParametersTemplate | None = None,
    result_ids: Sequence[ID_TYPES] | None = None,
    allocator: ResultIdAllocator | None = None,
    seed: str = "text",
    generation_kwargs: TextGenerationInitKwargs | None = None,
    job_config: HordeWorkerJobConfig | None = None,
    preserve_generation_id: bool = False,
) -> TextWorkerJob:
    """Instantiate a text job from a template."""
    generation_parameters = template.to_parameters(
        base_param_updates=base_param_updates,
        result_ids=result_ids,
        allocator=allocator,
        seed=seed,
    )
    init_kwargs: TextGenerationInitKwargs = {}
    if generation_kwargs:
        init_kwargs.update(generation_kwargs)
    if generation_id is not None:
        init_kwargs.setdefault("generation_id", generation_id)
    if dispatch_result_ids is not None:
        init_kwargs.setdefault("dispatch_result_ids", list(dispatch_result_ids))
    init_kwargs.setdefault("result_ids", generation_parameters.result_ids)
    generation = TextSingleGeneration(
        generation_parameters=generation_parameters,
        **init_kwargs,
    )
    return cls(
        generation=generation,
        job_config=job_config,
        job_id=job_id,
        dispatch_job_id=dispatch_job_id,
        dispatch_result_ids=dispatch_result_ids,
        preserve_generation_id=preserve_generation_id,
    )

set_dispatch_job_id 

set_dispatch_job_id(
    dispatch_job_id: ID_TYPES | None,
) -> None

Bind the job to the identifier supplied by dispatch.

Source code in horde_sdk/worker/job_base.py 
222
223
224
225
def set_dispatch_job_id(self, dispatch_job_id: ID_TYPES | None) -> None:
    """Bind the job to the identifier supplied by dispatch."""
    with self._lock:
        self._dispatch_job_id = dispatch_job_id

set_job_faulted 

set_job_faulted(
    faulted_reason: WORKER_ERRORS,
    failure_exception: Exception | None = None,
) -> None

Mark the entire job as faulted.

Note: This will mark all generations in the job as faulted.

Source code in horde_sdk/worker/job_base.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def set_job_faulted(self, faulted_reason: WORKER_ERRORS, failure_exception: Exception | None = None) -> None:
    """Mark the entire job as faulted.

    Note: This will mark all generations in the job as faulted.
    """
    with self._lock:
        if self._faulted:
            logger.warning(
                f"Job {self.job_identifier_string} is already marked faulted with "
                f"reason {self._fault_reason} at {self._faulted_at}",
            )

        self._faulted = True
        self._fault_reason = faulted_reason
        self._faulted_at = time.time()

        self.generation.on_abort(
            failed_message=faulted_reason,
            failure_exception=failure_exception,
        )