Task API Reference

Complete API reference for the background task system — @task decorator, TaskWrapper, TaskRunner, notification channels, and WebhookConfig.

This page documents the Singularity background task system, built on Celery and Redis. All signatures are taken directly from the source code in singularity/tasks/.

@task() Decorator

tasks.core.base_task.task

Decorator that wraps a plain synchronous function into a TaskWrapper and registers it in the global task registry for auto-discovery by TaskRunner.

Parameters

Parameter	Type	Default	Description
`name`	`str`	(required)	Unique task identifier. Used for submission and Celery routing.
`description`	`str`	`""`	Human-readable description of the task's purpose. Falls back to the function's docstring if empty.
`notify`	`Optional[List[str]]`	`None`	List of notification channel names to dispatch on completion. Built-in channels: `"log"`, `"callback"`, `"websocket"`, `"webhook"`.
`notify_config`	`Optional[Dict[str, Dict[str, Any]]]`	`None`	Per-channel configuration dict. Keys are channel names, values are config dicts passed to the channel. Example: `{"webhook": WebhookConfig(url="...")}`.
`max_retries`	`int`	`3`	Maximum number of retry attempts on failure.
`retry_delay`	`int`	`60`	Seconds between retries.
`time_limit`	`int`	`3600`	Hard timeout in seconds (1 hour default).

Returns

A decorator function that wraps fn in a TaskWrapper and returns it.

Usage

from singularity.tasks import task

@task(name="send_email", notify=["log", "callback"])
def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email to the given address."""
    # ... send logic ...
    return {"message_id": "abc123"}

TaskWrapper

tasks.core.base_task.TaskWrapper

The object returned by the @task decorator. It stores the original function, task configuration, extracted parameter metadata, and any registered callbacks.

Constructor

Parameter	Type	Default	Description
`fn`	`Callable`	(required)	The original task function
`name`	`str`	(required)	Unique task identifier
`description`	`str`	`""`	Human-readable description
`notify`	`Optional[List[str]]`	`None`	Notification channel names
`notify_config`	`Optional[Dict[str, Dict[str, Any]]]`	`None`	Per-channel config
`max_retries`	`int`	`3`	Maximum retry attempts
`retry_delay`	`int`	`60`	Seconds between retries
`time_limit`	`int`	`3600`	Hard timeout in seconds

Attributes

Attribute	Type	Description
`fn`	`Callable`	The original task function
`name`	`str`	Unique task identifier
`description`	`str`	Human-readable description (from `description` param or function docstring)
`notify_channels`	`List[str]`	Notification channel names
`notify_config`	`Dict[str, Dict[str, Any]]`	Per-channel configuration
`max_retries`	`int`	Maximum retry attempts
`retry_delay`	`int`	Seconds between retries
`time_limit`	`int`	Hard timeout in seconds
`on_complete_fn`	`Optional[Callable]`	Callback registered via `.on_complete` (initially `None`)
`params`	`List[Dict[str, Any]]`	Extracted parameter metadata from the function signature

Parameter Metadata Shape

Each entry in the params list has the following keys:

Key	Type	Description
`name`	`str`	Parameter name
`type`	`str`	Type annotation as string (e.g. `"str"`, `"int"`, `"any"`)
`default`	`Any`	Default value, or `None` if no default
`required`	`bool`	Whether the parameter is required
`kind`	`str`	Parameter kind: `"POSITIONAL_ONLY"`, `"POSITIONAL_OR_KEYWORD"`, `"VAR_POSITIONAL"`, `"KEYWORD_ONLY"`, or `"VAR_KEYWORD"`

Methods

Method	Parameters	Return Type	Description
`on_complete(fn)`	`fn: Callable`	`Callable`	Decorator to register a completion callback. The callback receives `(status: str, result: dict, meta: dict)` and runs synchronously inside the Celery worker. Returns the original callback function unchanged.
`__call__(args, *kwargs)`	`args`, `*kwargs`	`Any`	Execute the wrapped task function directly (used internally by the Celery worker).

`.on_complete` Usage

@task(name="generate_report", notify=["callback"])
def generate_report(report_type: str):
    return {"file": f"/reports/{report_type}.pdf"}

@generate_report.on_complete
def on_report_done(status: str, result: dict, meta: dict):
    if status == "success":
        print(f"Report ready: {result['file']}")

TaskRunner

tasks.core.runner.TaskRunner

Discovers, registers, and manages background tasks. This is the main interface used by services to submit tasks. It is injected into the Acquire DI container as acquire.tasks.

Constructor

Takes no parameters. Initializes an empty task_map and registers the four built-in notification channels.

Methods

Method	Parameters	Return Type	Description
`discover_tasks(register)`	`register: bool = True`	`Dict[str, TaskWrapper]`	Scan `tasks/executable/` for `@task`-decorated functions. Supports single-file tasks (`my_task.py`) and multi-file tasks (`complex_task/` directory). If `register=True`, also registers tasks with Celery. Set `register=False` for CLI listing.
`submit(task_name, args, _meta, _countdown, _on_complete, *kwargs)`	See table below	`AsyncResult`	Submit a task for background execution via Celery. Returns a Celery `AsyncResult` with an `.id` attribute for tracking.
`get_task(task_name)`	`task_name: str`	`Optional[TaskWrapper]`	Get a task wrapper by name for introspection.
`list_tasks()`	(none)	`List[Dict[str, Any]]`	List all registered tasks with their metadata (name, description, notify_channels, max_retries, time_limit, params).
`get_task_info(task_name)`	`task_name: str`	`Optional[Dict[str, Any]]`	Get comprehensive info about a task including its parameters, retry config, and whether it has an on_complete callback.
`register_notification_channel(name, channel)`	`name: str`, `channel: NotificationChannel`	`None`	Register a custom notification channel beyond the built-in ones.

`submit()` Parameters

Parameter	Type	Default	Description
`task_name`	`str`	(required)	The unique task identifier (as defined in `@task(name=...)`)
`*args`	`Any`	(positional)	Positional arguments forwarded to the task function
`_meta`	`Optional[Dict[str, Any]]`	`None`	Contextual metadata (user_id, org_id, webhook_url, etc.) passed through to notification channels
`_countdown`	`Optional[int]`	`None`	Delay in seconds before the task starts
`_on_complete`	`Optional[Callable]`	`None`	Ad-hoc callback function invoked on completion (in addition to any `@task.on_complete` callback)
`**kwargs`	`Any`	(keyword)	Keyword arguments forwarded to the task function

`submit()` Exceptions

Exception	Condition
`ValueError`	`task_name` not found in the registry
`TypeError`	Required parameters are missing from the submitted args/kwargs

`get_task_info()` Return Shape

{
    "name": "send_email",
    "description": "Send an email to the given address.",
    "params": [{"name": "to", "type": "str", "default": None, "required": True, "kind": "POSITIONAL_OR_KEYWORD"}],
    "notify_channels": ["log", "callback"],
    "max_retries": 3,
    "retry_delay": 60,
    "time_limit": 3600,
    "has_on_complete": False,
}

NotificationChannel (ABC)

tasks.core.notifications.NotificationChannel

Abstract base class for notification channels. All channels run synchronously inside the Celery worker process.

Abstract Method

Method	Parameters	Return Type	Description
`send(task_name, status, result, error, duration_ms, meta, task_config)`	See table below	`None`	Send a notification about task completion. Must be implemented by subclasses.

`send()` Parameters

Parameter	Type	Default	Description
`task_name`	`str`	(required)	The unique task identifier
`status`	`str`	(required)	`"success"` or `"failure"`
`result`	`Optional[Any]`	`None`	The return value of the task (on success)
`error`	`Optional[str]`	`None`	Error message (on failure)
`duration_ms`	`float`	`0`	Task execution duration in milliseconds
`meta`	`Optional[Dict[str, Any]]`	`None`	Contextual metadata passed at submit time
`task_config`	`Optional[Dict[str, Any]]`	`None`	Per-task channel configuration from `@task(notify_config={...})`

To create a custom notification channel, subclass NotificationChannel, implement send(), and register it with TaskRunner.register_notification_channel(name, instance).

WebhookConfig

tasks.core.notifications.webhook_channel.WebhookConfig

Dataclass for configuring a webhook notification target. Passed via @task(notify_config={"webhook": WebhookConfig(...)}).

Fields

Field	Type	Default	Description
`url`	`str`	(required)	The webhook endpoint URL
`headers`	`Dict[str, str]`	`{}`	HTTP headers to include (e.g. Authorization, API keys)
`method`	`str`	`"POST"`	HTTP method: `"POST"` or `"PUT"`
`timeout`	`int`	`10`	Request timeout in seconds

Usage

from singularity.tasks.notifications.webhook_channel import WebhookConfig

slack_webhook = WebhookConfig(
    url="https://hooks.slack.com/services/xxx/yyy",
    headers={"Content-Type": "application/json"},
)

@task(
    name="process_order",
    notify=["webhook"],
    notify_config={"webhook": slack_webhook},
)
def process_order(order_id: str):
    return {"order_id": order_id, "status": "completed"}

Built-in Notification Channels

LogChannel

tasks.core.notifications.log_channel.LogChannel

Logs task completion as a structured Loguru message with task metadata.

Behavior	Description
Success	Calls `logger.info("Task completed", ...)` with task_name, status, duration_ms, truncated result (200 chars max), and safe meta keys
Failure	Calls `logger.error("Task failed", ...)` with the same fields plus the error message

CallbackChannel

tasks.core.notifications.callback_channel.CallbackChannel

Invokes registered callback functions. Looks for callbacks in two places (both are called if present):

Source	How It Gets There	Signature
`TaskWrapper.on_complete_fn`	Registered via `@task_fn.on_complete` decorator	`(status: str, result: dict, meta: dict) -> None`
Submit-time callback	Passed via `_on_complete` kwarg to `TaskRunner.submit()`	`(status: str, result: dict, meta: dict) -> None`

WebSocketChannel

tasks.core.notifications.websocket_channel.WebSocketChannel

Publishes task completion to a Redis pub/sub channel (task_notifications). A listener in the FastAPI process subscribes and broadcasts via WebSocketManager.

Aspect	Value
Redis channel name	`"task_notifications"`
Message type	`"task_completed"`
Routing keys	`org_id`, `user_id` from `meta` (if present)
Result truncation	Results larger than 1000 chars are truncated with a `_truncated` flag

WebhookChannel

tasks.core.notifications.webhook_channel.WebhookChannel

Sends task completion as an HTTP request to an external URL. Configuration is resolved in priority order:

Priority	Source	Description
1	`task_config`	A `WebhookConfig` instance from `@task(notify_config={"webhook": config})`
2	`task_config` (dict)	A plain dict with a `url` key (backwards compatibility)
3	`meta["webhook_url"]`	Simple URL-only override at submit time

If none of the above are set, the notification is skipped with a warning.

Webhook Payload Shape

{
    "task_name": "process_order",
    "status": "success",
    "duration_ms": 1234.56,
    "result": {"order_id": "123"},   # omitted if None; truncated to 500 chars if not JSON-serializable
    "error": "...",                    # only on failure
    "meta": {"user_id": "..."}        # safe keys only (excludes keys starting with _)
}

NotificationRegistry

tasks.core.notifications.NotificationRegistry

Internal registry of notification channel instances. Channels are registered by name and looked up during dispatch.

Methods

Method	Parameters	Return Type	Description
`register(name, channel)`	`name: str`, `channel: NotificationChannel`	`None`	Register a notification channel by name
`get(name)`	`name: str`	`Optional[NotificationChannel]`	Get a notification channel by name
`dispatch(channel_names, task_name, status, result, error, duration_ms, meta, notify_config)`	`channel_names: List[str]`, `task_name: str`, `status: str`, `result: Optional[Any]`, `error: Optional[str]`, `duration_ms: float`, `meta: Optional[Dict]`, `notify_config: Optional[Dict]`	`None`	Dispatch notifications to the specified channels. Errors in individual channels are logged but do not propagate.

On this page