Skip to content

Types

Enums

chicory.types.BrokerType

Bases: StrEnum

Supported message broker implementations.

chicory.types.BackendType

Bases: StrEnum

Supported result backend implementations.

chicory.types.TaskState

Bases: StrEnum

Possible states of a task during its lifecycle.

chicory.types.DeliveryMode

Bases: StrEnum

Message delivery guarantee level.

chicory.types.ValidationMode

Bases: StrEnum

When and what to validate on task arguments and return values.

chicory.types.RetryBackoff

Bases: StrEnum

Backoff strategy for retries.

Models

chicory.types.RetryPolicy

Bases: BaseModel

Configuration for task retry behavior.

calculate_delay 

calculate_delay(attempt: int) -> float

Calculate delay for a given retry attempt (1-indexed).

Source code in src/chicory/types.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def calculate_delay(self, attempt: int) -> float:
    """Calculate delay for a given retry attempt (1-indexed)."""

    match self.backoff:
        case RetryBackoff.FIXED:
            delay = self.retry_delay
        case RetryBackoff.LINEAR:
            delay = self.retry_delay * attempt
        case RetryBackoff.EXPONENTIAL:
            delay = self.retry_delay * (2 ** (attempt - 1))
        case _:  # pragma: no cover
            delay = self.retry_delay  # Fallback to fixed

    # Apply max delay cap
    delay = min(delay, self.max_delay)

    # Apply jitter (±25%)
    if self.jitter:
        jitter_range = delay * 0.25
        delay = delay + random.uniform(-jitter_range, jitter_range)

    return max(0, delay)

should_retry 

should_retry(exception: Exception) -> bool

Determine if the given exception should trigger a retry.

Source code in src/chicory/types.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def should_retry(self, exception: Exception) -> bool:
    """Determine if the given exception should trigger a retry."""
    exc_name = type(exception).__name__
    exc_full_name = f"{type(exception).__module__}.{exc_name}"

    # Check ignore list first (takes precedence)
    if self.ignore_on:
        for ignored in self.ignore_on:
            if exc_name == ignored or exc_full_name == ignored:
                return False

    # Check retry list
    if self.retry_on is None:
        return True  # Retry on all exceptions

    for allowed in self.retry_on:
        if exc_name == allowed or exc_full_name == allowed:
            return True

    return False

chicory.types.TaskMessage

Bases: BaseModel

Message schema for task serialization.

chicory.types.TaskResult

Bases: BaseModel, Generic[T]

Result payload stored in backend

chicory.types.TaskOptions

Bases: BaseModel

Configuration options for a task

get_retry_policy 

get_retry_policy() -> RetryPolicy

Get the retry policy, returning a default (no retries) if not configured.

Source code in src/chicory/types.py 
206
207
208
209
210
211
def get_retry_policy(self) -> RetryPolicy:
    """Get the retry policy, returning a default (no retries) if not configured."""
    if self.retry_policy:
        return self.retry_policy

    return RetryPolicy(max_retries=0)

chicory.types.WorkerStats

Bases: BaseModel

Runtime statistics reported by a worker.

uptime_seconds property 

uptime_seconds: float

Worker uptime in seconds.

chicory.types.BrokerStatus

Bases: BaseStatus

Health-check result for a broker connection.

chicory.types.BackendStatus

Bases: BaseStatus

Health-check result for a backend connection.