TaskConfig model reference

sheppy.models.TaskConfig

Bases: BaseModel

Task configuration

ATTRIBUTE	DESCRIPTION
retry	Number of times to retry the task if it fails. Default is 0 (no retries). TYPE: int
retry_delay	Delay between retries in seconds. If a single float is provided, it will be used for all retries. If a list is provided, it will be used for each retry attempt respectively (exponential backoff). Default is 1.0 seconds. TYPE: float | list[float]

Note

• You should not create TaskConfig instances directly. Instead, use the @task decorator to define a task function, and then call that function to create a Task instance.
• retry must be a non-negative integer.
• retry_delay must be a positive float or a list of positive floats.

Example

from sheppy import task

@task(retry=3, retry_delay=[1, 2, 3])
def my_task():
    raise Exception("Something went wrong")

t = my_task()
print(t.config.retry)  # 3
print(t.config.retry_delay)  # [1.0, 2.0, 3.0]

retry

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

int: Number of times to retry the task if it fails. Default is 0 (no retries).

retry_delay

retry_delay: float | list[float] = Field(default=1.0)

float|list[float]: Delay between retries in seconds. If a single float is provided, it will be used for all retries. If a list is provided, it will be used for each retry attempt respectively (exponential backoff). Default is 1.0 seconds.

validate_retry_delay

validate_retry_delay(
    v: float | list[float],
) -> float | list[float]

Source code in src/sheppy/models.py

@field_validator('retry_delay')
@classmethod
def validate_retry_delay(cls, v: float | list[float]) -> float | list[float]:
    if isinstance(v, list) and len(v) == 0:
        raise ValueError("retry_delay list cannot be empty")
    return v