Celery Task Message Protocol v1 and v2 encoder and decoder (ported from celery.app.amqp
and celery.worker.request).
Protocol v2 (Default in Celery >= 4.0)
headers:%{ "lang" => "py", "task" => name, "id" => task_id, "shadow" => shadow, "eta" => eta, "expires" => expires, "group" => group_id, "group_index" => group_index, "retries" => retries, "timelimit" => [time_limit, soft_time_limit], "root_id" => root_id, "parent_id" => parent_id, "argsrepr" => argsrepr, "kwargsrepr" => kwargsrepr, "origin" => origin, "ignore_result" => ignore_result, "replaced_task_nesting" => 0, "stamped_headers" => stamped_headers, "stamps" => stamps }properties:%{"correlation_id" => task_id, "reply_to" => reply_to || ""}body:[args, kwargs, %{"callbacks" => callbacks, "errbacks" => errbacks, "chain" => chain, "chord" => chord}]
Protocol v1
headers:%{}properties:%{"correlation_id" => task_id, "reply_to" => reply_to || ""}body:%{ "task" => name, "id" => task_id, "args" => args, "kwargs" => kwargs, "retries" => retries, "eta" => eta, "expires" => expires, "utc" => true, "callbacks" => callbacks, "errbacks" => errbacks, "timelimit" => [time_limit, soft_time_limit], "taskset" => group_id, "chord" => chord }
Summary
Functions
Creates a Celery Protocol v1 task message structure (%{headers: ..., properties: ..., body: ...}).
Creates a Celery Protocol v2 task message structure (%{headers: ..., properties: ..., body: ...}).
Decodes an incoming %Kombu.Message{} (or decoded payload/headers/properties map)
into {:ok, %Celery.Task.Context{}} or {:error, reason}.
Decodes decoded body, headers, properties, and delivery_info into {:ok, %Celery.Task.Context{}}.
Returns the current node origin string (e.g. "pid@hostname"), matching Python Celery's
anon@hostname / gen<pid>@hostname convention.
Returns the remaining delay in milliseconds until eta, or 0 if eta is nil or in the past.
Returns true if expires is in the past compared to now.
Formats positional arguments like Python's saferepr(args):
[1, 2] -> "(1, 2)", [1] -> "(1,)", [] -> "()".
Formats keyword arguments like Python's saferepr(kwargs):
%{"a" => 1} -> "{'a': 1}", %{} -> "{}".
Formats eta or countdown (in seconds) into an ISO-8601 UTC string.
Parses an ISO-8601 timestamp string, DateTime, or numeric timestamp into {:ok, %DateTime{}}.
Generates a UUIDv4 task ID compatible with Python Celery.
Types
Functions
@spec as_task_v1(String.t(), list(), map(), keyword()) :: task_message()
Creates a Celery Protocol v1 task message structure (%{headers: ..., properties: ..., body: ...}).
@spec as_task_v2(String.t(), list(), map(), keyword()) :: task_message()
Creates a Celery Protocol v2 task message structure (%{headers: ..., properties: ..., body: ...}).
@spec decode_message(Kombu.Message.t() | map()) :: {:ok, Celery.Task.Context.t()} | {:error, term()}
Decodes an incoming %Kombu.Message{} (or decoded payload/headers/properties map)
into {:ok, %Celery.Task.Context{}} or {:error, reason}.
Supports:
- Protocol v2 (
headers["task"]+body = [args, kwargs, embed]) - Protocol v1 (
body = %{"task" => ..., "id" => ..., "args" => ..., "kwargs" => ...}) - Hybrid messages (
headers["task"]+ map body or v1 headers merged with body)
@spec decode_parts(term(), map(), map(), map()) :: {:ok, Celery.Task.Context.t()} | {:error, term()}
Decodes decoded body, headers, properties, and delivery_info into {:ok, %Celery.Task.Context{}}.
@spec default_origin() :: String.t()
Returns the current node origin string (e.g. "pid@hostname"), matching Python Celery's
anon@hostname / gen<pid>@hostname convention.
@spec eta_delay_ms(term(), DateTime.t()) :: non_neg_integer()
Returns the remaining delay in milliseconds until eta, or 0 if eta is nil or in the past.
@spec expired?(term(), DateTime.t()) :: boolean()
Returns true if expires is in the past compared to now.
Formats positional arguments like Python's saferepr(args):
[1, 2] -> "(1, 2)", [1] -> "(1,)", [] -> "()".
Formats keyword arguments like Python's saferepr(kwargs):
%{"a" => 1} -> "{'a': 1}", %{} -> "{}".
@spec format_time_option( DateTime.t() | NaiveDateTime.t() | String.t() | number() | nil, number() | nil ) :: String.t() | nil
Formats eta or countdown (in seconds) into an ISO-8601 UTC string.
@spec parse_datetime(term()) :: {:ok, DateTime.t()} | :error
Parses an ISO-8601 timestamp string, DateTime, or numeric timestamp into {:ok, %DateTime{}}.
@spec uuid() :: String.t()
Generates a UUIDv4 task ID compatible with Python Celery.