Skip to content

querysets

django_spire.celery.querysets

CeleryTaskQuerySet

Bases: QuerySet

by_model_keys

Source code in django_spire/celery/querysets.py 
8
9
def by_model_keys(self, model_keys: list[str]) -> QuerySet:
    return self.filter(model_key__in=model_keys)

by_reference_keys

Source code in django_spire/celery/querysets.py 
11
12
def by_reference_keys(self, reference_keys: list[str]) -> QuerySet:
    return self.filter(reference_key__in=reference_keys)

by_reference_keys_model_keys

Source code in django_spire/celery/querysets.py 
14
15
16
17
18
19
20
21
22
23
24
25
def by_reference_keys_model_keys(
    self, reference_keys_model_keys: dict[str, str | None]
) -> QuerySet:
    query = Q()

    for reference_key, model_key in reference_keys_model_keys.items():
        if model_key is not None:
            query |= Q(reference_key=reference_key, model_key=model_key)
        else:
            query |= Q(reference_key=reference_key, model_key__isnull=True)

    return self.filter(query)

by_unready

Source code in django_spire/celery/querysets.py 
27
28
def by_unready(self) -> QuerySet:
    return self.exclude(state__in=[*states.READY_STATES, *states.EXCEPTION_STATES])

by_completed

Source code in django_spire/celery/querysets.py 
30
31
def by_completed(self) -> QuerySet:
    return self.filter(state__in=states.SUCCESS)