Elixir port of Python's Celery distributed task queue (celery.Celery),
wire-compatible with Python Celery (Protocol v1 & v2) and backed by Kombu and Redis
for transport and result storage.
Quick Start
Define a task module using use Celery.Task and @task:
defmodule Src.Tasks do
use Celery.Task
@task_name "src.tasks.command"
@task_metadata %{"service" => "pipeline"}
@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
endCreate a Celery application and start a worker:
celery = Celery.new(
broker: "redis://localhost:6379/0",
backend: "redis://localhost:6379/0",
include: ["src.tasks"]
)
{:ok, worker} = Celery.start_worker(celery)Summary
Functions
Convenience function to dispatch a task with full options (apply_async).
Creates a %Celery.Result{} (AsyncResult) bound to app.
Executes a chain of %Celery.Canvas.Signature{} tasks.
Executes a chord (parallel group header + callback body signature).
Convenience function to dispatch a task with positional args and keyword kwargs (delay).
Executes a parallel group of %Celery.Canvas.Signature{} tasks.
Creates a new %Celery{} application instance.
Registers a %Celery.Task.Definition{} or a module using Celery.Task on the %Celery{} app.
Creates a %Celery.Canvas.Signature{} (s) with args and kwargs.
Sends a task by name (celery.send_task(name, args=..., kwargs=..., ...)).
Creates an immutable %Celery.Canvas.Signature{} (si) with args and kwargs.
Creates a %Celery.Canvas.Signature{} bound to app.
Starts a Celery.Worker consuming tasks for app.
Registers an inline task function on the %Celery{} application (matching @celery.task(...)).
Updates the configuration of a %Celery{} application (app.conf.update(...)).
Types
@type t() :: %Celery{ config: Celery.Config.t(), main: String.t(), tasks: %{required(String.t()) => Celery.Task.Definition.t()} }
Functions
@spec apply_async(t(), Celery.Task.Definition.t() | String.t(), keyword()) :: {:ok, Celery.Result.t()} | {:error, term()}
Convenience function to dispatch a task with full options (apply_async).
@spec async_result(t(), String.t()) :: Celery.Result.t()
Creates a %Celery.Result{} (AsyncResult) bound to app.
@spec chain([Celery.Canvas.Signature.t()], keyword()) :: {:ok, Celery.Result.t()} | {:error, term()}
Executes a chain of %Celery.Canvas.Signature{} tasks.
@spec chord([Celery.Canvas.Signature.t()], Celery.Canvas.Signature.t(), keyword()) :: {:ok, Celery.Result.t()} | {:error, term()}
Executes a chord (parallel group header + callback body signature).
@spec delay(t(), Celery.Task.Definition.t() | String.t(), list(), map() | keyword()) :: {:ok, Celery.Result.t()} | {:error, term()}
Convenience function to dispatch a task with positional args and keyword kwargs (delay).
@spec group([Celery.Canvas.Signature.t()], keyword()) :: {:ok, Celery.Result.GroupResult.t()} | {:error, term()}
Executes a parallel group of %Celery.Canvas.Signature{} tasks.
Creates a new %Celery{} application instance.
Accepts Python Celery-style options:
:brokeror:broker_url- Redis broker URL (e.g."redis://localhost:6379/0"):backendor:result_backend- Redis result backend URL (e.g."redis://localhost:6379/0"):include- list of task modules ([Src.Tasks]or Python-style["src.tasks"])- any other
%Celery.Config{}option (:task_serializer,:result_serializer,:task_acks_late, etc.)
@spec register_task(t(), Celery.Task.Definition.t() | module() | String.t()) :: t()
Registers a %Celery.Task.Definition{} or a module using Celery.Task on the %Celery{} app.
@spec s(String.t() | Celery.Task.Definition.t(), list(), map() | keyword()) :: Celery.Canvas.Signature.t()
Creates a %Celery.Canvas.Signature{} (s) with args and kwargs.
@spec send_task(t(), String.t(), keyword()) :: {:ok, Celery.Result.t()} | {:error, term()}
Sends a task by name (celery.send_task(name, args=..., kwargs=..., ...)).
If app.config.task_always_eager is true, executes the task locally via Celery.Task.apply_local/5.
Otherwise encodes a Celery Protocol v2 (or v1) task message and publishes it over Kombu to Redis.
Returns {:ok, %Celery.Result{id: task_id}}.
@spec si(String.t() | Celery.Task.Definition.t(), list(), map() | keyword()) :: Celery.Canvas.Signature.t()
Creates an immutable %Celery.Canvas.Signature{} (si) with args and kwargs.
@spec signature(t(), String.t(), keyword()) :: Celery.Canvas.Signature.t()
Creates a %Celery.Canvas.Signature{} bound to app.
@spec start_worker(t() | keyword(), keyword()) :: GenServer.on_start()
Starts a Celery.Worker consuming tasks for app.
@spec task(t(), keyword(), function()) :: {t(), Celery.Task.Definition.t()}
Registers an inline task function on the %Celery{} application (matching @celery.task(...)).
Example
celery =
Celery.task(
celery,
[bind: true, name: "src.tasks.command", metadata: %{"v" => 1}],
fn self, pipe_result, input_files, output_path, workflow_id, task_config ->
"This is a command"
end
)Updates the configuration of a %Celery{} application (app.conf.update(...)).