Skip to content
Menu

Documentation

sandloom.sdk.tasks

Background tasks: a typed shared_task for <module>/tasks.py.

Public API 1.0 (spec §2.2, Appendix C).

from sandloom.sdk.tasks import <name>. Signatures in a terminal: sandloom docs <name>. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line # sandloom: advanced <reason> in generated code and examples).

Advanced

Name Kind Summary
cron function A beat schedule from a five-field cron expression ("minute hour day-of-month month day-of-week", e.g. "0 3 * * *"):
shared_task function @shared_task / @shared_task(name=..., acks_late=True, ...).
Task class A registered task: call it inline, or enqueue it.

Advanced names

cron

def cron(expression: str) -> Any

A beat schedule from a five-field cron expression ("minute hour day-of-month month day-of-week", e.g. "0 3 * * *"):

def setup_periodic_tasks(sender: Any, **kwargs: Any) -> None:
    sender.add_periodic_task(cron("0 3 * * *"), purge.s(), name="m.purge")

shared_task

def shared_task(
    fn: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    acks_late: bool | None = None,
    reject_on_worker_lost: bool | None = None,
    **options: Any,
) -> Any

@shared_task / @shared_task(name=..., acks_late=True, ...).

An idempotent, at-least-once task (redelivered when a worker dies) is @shared_task(acks_late=True, reject_on_worker_lost=True); its body runs async work through run_async(...) with open_uow() blocks.

Task

class Task(Protocol)

A registered task: call it inline, or enqueue it.

Task.delay

def delay(self, *args: P.args, **kwargs: P.kwargs) -> Any

Task.s

def s(self, *args: Any, **kwargs: Any) -> Any

A signature (sender.add_periodic_task(schedule, task.s())).

Task.si

def si(self, *args: Any, **kwargs: Any) -> Any

Task.apply_async

def apply_async(
    self,
    args: tuple[Any, ...] | None = None,
    kwargs: dict[str, Any] | None = None,
    **options: Any,
) -> Any