Skip to content

manager

django_spire.celery.manager

BaseCeleryTaskManager

Bases: ABC

Used for creating a rigid structure in the client code for calling celery tasks on remote workers

Source code in django_spire/celery/manager.py 
58
59
def __init__(self, model_object: Model | None = None) -> None:
    self.model_object = model_object

task_name instance-attribute

display_name instance-attribute

required_kwargs_keys_types = None class-attribute instance-attribute

send_task_retries = 2 class-attribute instance-attribute

model_object = model_object instance-attribute

reference_key property

model_key property

reference_and_model_key property

class_and_send_task_method property

__init_subclass__

Source code in django_spire/celery/manager.py 
45
46
47
48
49
50
51
52
53
54
55
56
def __init_subclass__(cls, **kwargs) -> None:
    required_class_attributes = ('task_name', 'display_name')

    for attribute_name in required_class_attributes:
        attr_value = getattr(cls, attribute_name, None)
        if not isinstance(attr_value, str):
            message = f'{cls.__name__}.{attribute_name} must be set and type string'
            raise TypeError(message)

    if cls.send_task_retries > _MAX_SEND_TASK_RETRIES:
        message = f'{cls.__name__}.send_task_retries = {cls.send_task_retries} exceeded the maximum number {_MAX_SEND_TASK_RETRIES} of retries allowed'
        raise ValueError(message)

create_pydantic_model_class classmethod

Source code in django_spire/celery/manager.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def create_pydantic_model_class(cls, model_name: str) -> type[BaseModel]:
    if cls.required_kwargs_keys_types is None:
        message = 'Cannot create pydantic model without required kwargs keys'
        raise ValueError(message)

    return create_model(
        model_name,
        **{key: (type_, ...) for key, type_ in cls.required_kwargs_keys_types.items()},
    )

filter_celery_tasks

Source code in django_spire/celery/manager.py 
104
105
def filter_celery_tasks(self) -> QuerySet[CeleryTask]:
    return CeleryTask.objects.by_reference_keys_model_keys({self.reference_key: self.model_key})

send_task

Source code in django_spire/celery/manager.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def send_task(self, **kwargs) -> CeleryTask:
    self._validate_and_kwargs(**kwargs)

    attempt = 0
    last_exception: Exception | None = None

    while attempt <= self.send_task_retries:
        try:
            return self._create_celery_task(send_task(name=self.task_name, kwargs=kwargs))
        except _SEND_RETRYABLE_EXCEPTIONS as e:
            attempt += 1
            last_exception = e

            if attempt > self.send_task_retries:
                break

            time.sleep(1 * (2 ** (attempt - 1)))

    return self._create_failed_celery_task(
        error_message=str(last_exception) if last_exception else 'Unknown error',
        original_args=(),
        original_kwargs=kwargs,
    )