Celery.Task (celery v0.1.0)

Celery Task behaviour, @task attribute decorator, deftask macro, and runtime self struct (ported from celery.app.task.Task).

Defining Tasks in Elixir

You can define tasks using either the @task module attribute (mirroring Python's @celery.task(...) decorator) or the deftask macro:

defmodule Src.Tasks do
  use Celery.Task

  @task_name "src.tasks.command"
  @task_metadata %{"owner" => "pipeline", "version" => 1}

  @task bind: true, name: @task_name, metadata: @task_metadata
  def command(
    self,
    pipe_result \\ nil,
    input_files \\ nil,
    output_path \\ nil,
    workflow_id \\ nil,
    task_config \\ nil
  ) do
    "This is a command"
  end
end

When bind: true is specified, the first argument (self) is a %Celery.Task{} struct containing:

self.name - the registered task name
self.metadata - the task metadata configured via metadata: ...
self.request - the %Celery.Task.Context{} for the current execution
self.app - the %Celery{} application instance

Summary

Functions

apply_async(app, defn, opts)

Dispatches a task asynchronously (task.apply_async(...)).

apply_local(app, task_or_name, args \\ [], kwargs \\ %{}, opts \\ [])

Executes a task locally in the calling process (task.apply(...)), returning {:ok, %Celery.Result{is_eager: true}}.

bind_call_arguments(defn, self_task, args, kwargs)

Binds positional args and named kwargs to the task function's parameter list, prepending self_task if defn.bind == true.

deftask(call, opts \\ [], do_block)

Defines a Celery task function with options (bind, name, metadata, max_retries, etc.).

delay(app, task_or_name, args \\ [], kwargs \\ %{})

Dispatches a task using positional args and keyword kwargs (task.delay(...)).

invoke(defn, args \\ [], kwargs \\ %{}, opts \\ [])

Invokes a %Celery.Task.Definition{} with positional args (list), keyword kwargs (map), and %Celery.Task.Context{} (request), binding self as the first argument when bind: true and resolving positional, keyword, and default arguments just like Python Celery.

new_self(defn, request, app \\ nil)

Builds a %Celery.Task{} (self) instance for a %Celery.Task.Definition{} and a %Celery.Task.Context{} request.

retry(self_task, opts \\ [])

Retries the currently executing task (ported from Task.retry in celery/app/task.py).

s(task_or_name, args \\ [], kwargs \\ %{})

Creates a %Celery.Canvas.Signature{} for a task definition or task name (task.s(...)).

si(task_or_name, args \\ [], kwargs \\ %{})

Creates an immutable %Celery.Canvas.Signature{} (task.si(...)).

update_state(self_task, state_or_opts, meta \\ nil)

Updates the task's state and metadata in the Redis result backend directly from within a running task (ported from Task.update_state in celery/app/task.py).

Types

t()

@type t() :: %Celery.Task{
  acks_late: boolean(),
  app: Celery.t() | nil,
  default_retry_delay: number(),
  definition: Celery.Task.Definition.t() | nil,
  ignore_result: boolean(),
  max_retries: non_neg_integer() | :infinity | nil,
  metadata: term(),
  name: String.t(),
  request: Celery.Task.Context.t(),
  serializer: String.t(),
  track_started: boolean()
}