# `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`](<#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`](<#shared_task>) | function | `@shared_task` / `@shared_task(name=..., acks_late=True, ...)`. |
| [`Task`](<#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
```

## Examples and related guides

- [Background tasks](<https://sandloom.dev/docs/modules/background/tasks/index.md>)
- [Python API overview](<https://sandloom.dev/docs/reference/sdk/index.md>)