Canvas：设计工作流¶

Canvas: Designing Work-flows

签名¶

Signatures

中文

Added in version 2.0.

你已经在 calling 指南中学习了如何使用任务的 delay 方法来调用任务，这通常已足够，但有时你可能希望将任务调用的签名传递给另一个进程，或者作为参数传递给另一个函数。

signature() 会将一次任务调用的参数、关键字参数及执行选项封装起来，以便它可以被传递给其他函数，甚至被序列化并通过网络发送。

你可以使用任务名称为 add 任务创建一个签名，如下所示：
```
>>> from celery import signature
>>> signature('tasks.add', args=(2, 2), countdown=10)
tasks.add(2, 2)
```
此任务的签名为 arity 为 2（两个参数）：(2, 2)，并设置执行选项 countdown 为 10。
或者你可以使用任务的 signature 方法来创建签名：
```
>>> add.signature((2, 2), countdown=10)
tasks.add(2, 2)
```
也可以使用星号参数的快捷方式：
```
>>> add.s(2, 2)
tasks.add(2, 2)
```

同样也支持关键字参数：

>>> add.s(2, 2, debug=True)
tasks.add(2, 2, debug=True)

你可以从任意签名实例中检查不同的字段：

>>> s = add.signature((2, 2), {'debug': True}, countdown=10)
>>> s.args
(2, 2)
>>> s.kwargs
{'debug': True}
>>> s.options
{'countdown': 10}

它支持 delay、apply_async 等“调用 API”，包括可直接调用（即 __call__）。

调用该签名将在当前进程中直接执行任务：
```
>>> add(2, 2)
4
>>> add.s(2, 2)()
4
```
delay 是我们喜爱的快捷方式，相当于使用星号参数调用 apply_async：
```
>>> result = add.delay(2, 2)
>>> result.get()
4
```
apply_async 接受的参数与 app.Task.apply_async() 方法相同：
```
>>> add.apply_async(args, kwargs, **options)
>>> add.signature(args, kwargs, **options).apply_async()

>>> add.apply_async((2, 2), countdown=1)
>>> add.signature((2, 2), countdown=1).apply_async()
```
你不能通过 s() 方法定义选项，但可以通过链式调用 set 方法来设置：
```
>>> add.s(2, 2).set(countdown=1)
proj.tasks.add(2, 2)
```

英文

Added in version 2.0.

You just learned how to call a task using the tasks delay method in the calling guide, and this is often all you need, but sometimes you may want to pass the signature of a task invocation to another process or as an argument to another function.

A signature() wraps the arguments, keyword arguments, and execution options of a single task invocation in a way such that it can be passed to functions or even serialized and sent across the wire.

You can create a signature for the add task using its name like this:
```
>>> from celery import signature
>>> signature('tasks.add', args=(2, 2), countdown=10)
tasks.add(2, 2)
```
This task has a signature of arity 2 (two arguments): (2, 2), and sets the countdown execution option to 10.
or you can create one using the task's signature method:
```
>>> add.signature((2, 2), countdown=10)
tasks.add(2, 2)
```
There's also a shortcut using star arguments:
```
>>> add.s(2, 2)
tasks.add(2, 2)
```

Keyword arguments are also supported:

>>> add.s(2, 2, debug=True)
tasks.add(2, 2, debug=True)

From any signature instance you can inspect the different fields:

>>> s = add.signature((2, 2), {'debug': True}, countdown=10)
>>> s.args
(2, 2)
>>> s.kwargs
{'debug': True}
>>> s.options
{'countdown': 10}

It supports the "Calling API" of delay, apply_async, etc., including being called directly (__call__).

Calling the signature will execute the task inline in the current process:
```
>>> add(2, 2)
4
>>> add.s(2, 2)()
4
```
delay is our beloved shortcut to apply_async taking star-arguments:
```
>>> result = add.delay(2, 2)
>>> result.get()
4
```
apply_async takes the same arguments as the app.Task.apply_async() method:
```
>>> add.apply_async(args, kwargs, **options)
>>> add.signature(args, kwargs, **options).apply_async()

>>> add.apply_async((2, 2), countdown=1)
>>> add.signature((2, 2), countdown=1).apply_async()
```
You can't define options with s(), but a chaining set call takes care of that:
```
>>> add.s(2, 2).set(countdown=1)
proj.tasks.add(2, 2)
```

部分代码¶

Partials

中文

使用签名，你可以在 worker 中执行任务：

>>> add.s(2, 2).delay()
>>> add.s(2, 2).apply_async(countdown=1)

或者也可以在当前进程中直接调用：

>>> add.s(2, 2)()
4

向 apply_async/delay 提供额外的 args、kwargs 或 options，会创建部分签名（partial）：

所添加的位置参数会被追加到签名中的已有参数前面：

>>> partial = add.s(2)          # 不完整的签名
>>> partial.delay(4)            # 4 + 2
>>> partial.apply_async((4,))  # 同上

所添加的关键字参数会与签名中的关键字参数合并，新添加的关键字参数将覆盖已有参数：

>>> s = add.s(2, 2)
>>> s.delay(debug=True)                    # -> add(2, 2, debug=True)
>>> s.apply_async(kwargs={'debug': True})  # 同上

所添加的选项会与签名中的选项合并，新添加的选项将覆盖已有选项：
```
>>> s = add.signature((2, 2), countdown=10)
>>> s.apply_async(countdown=1)  # countdown 现在为 1
```

你还可以克隆签名以创建派生签名：

>>> s = add.s(2)
proj.tasks.add(2)

>>> s.clone(args=(4,), kwargs={'debug': True})
proj.tasks.add(4, 2, debug=True)

英文

With a signature, you can execute the task in a worker:

>>> add.s(2, 2).delay()
>>> add.s(2, 2).apply_async(countdown=1)

Or you can call it directly in the current process:

>>> add.s(2, 2)()
4

Specifying additional args, kwargs, or options to apply_async/delay creates partials:

Any arguments added will be prepended to the args in the signature:

>>> partial = add.s(2)          # incomplete signature
>>> partial.delay(4)            # 4 + 2
>>> partial.apply_async((4,))  # same

Any keyword arguments added will be merged with the kwargs in the signature, with the new keyword arguments taking precedence:

>>> s = add.s(2, 2)
>>> s.delay(debug=True)                    # -> add(2, 2, debug=True)
>>> s.apply_async(kwargs={'debug': True})  # same

Any options added will be merged with the options in the signature, with the new options taking precedence:
```
>>> s = add.signature((2, 2), countdown=10)
>>> s.apply_async(countdown=1)  # countdown is now 1
```

You can also clone signatures to create derivatives:

>>> s = add.s(2)
proj.tasks.add(2)

>>> s.clone(args=(4,), kwargs={'debug': True})
proj.tasks.add(4, 2, debug=True)

不变性¶

Immutability

中文

Added in version 3.0.

部分签名（Partials）主要用于回调函数。任何通过 link 关联的任务或 chord 的回调函数，都会接收到父任务的返回结果。但有时你可能希望指定一个**不接受附加参数**的回调函数，此时可以将签名设置为不可变（immutable）：

>>> add.apply_async((2, 2), link=reset_buffers.signature(immutable=True))

你也可以使用 .si() 这个快捷方式来创建不可变签名：

>>> add.apply_async((2, 2), link=reset_buffers.si())

当一个签名是不可变的时，只能设置其执行选项，因此无法通过传递额外的参数或关键字参数来调用该签名。

备注

在本教程中，我有时会对签名使用前缀操作符 ~。在生产环境中你可能不应使用这个操作符，但在 Python 交互式 shell 中做实验时它非常方便：

>>> ~sig

>>> # 等价于
>>> sig.delay().get()

英文

Added in version 3.0.

Partials are meant to be used with callbacks, any tasks linked, or chord callbacks will be applied with the result of the parent task. Sometimes you want to specify a callback that doesn't take additional arguments, and in that case you can set the signature to be immutable:

>>> add.apply_async((2, 2), link=reset_buffers.signature(immutable=True))

The .si() shortcut can also be used to create immutable signatures:

>>> add.apply_async((2, 2), link=reset_buffers.si())

Only the execution options can be set when a signature is immutable, so it's not possible to call the signature with partial args/kwargs.

备注

In this tutorial I sometimes use the prefix operator ~ to signatures. You probably shouldn't use it in your production code, but it's a handy shortcut when experimenting in the Python shell:

>>> ~sig

>>> # is the same as
>>> sig.delay().get()

回调函数¶

Callbacks

中文

Added in version 3.0.

你可以使用 apply_async 的 link 参数，为任何任务添加回调函数：

add.apply_async((2, 2), link=other_task.s())

该回调任务仅在前一个任务成功完成时才会被执行，且其参数将是前一个任务的返回值。

如前所述，任何添加到签名上的参数都会被追加在签名原本定义的参数之前！

如果你有以下签名：

>>> sig = add.s(10)

那么 sig.delay(result) 实际相当于：

>>> add.apply_async(args=(result, 10))

...

现在让我们通过部分参数，给 add 任务设置一个回调函数：

>>> add.apply_async((2, 2), link=add.s(8))

如预期，这首先会启动一个任务来计算 2 + 2，然后启动另一个任务来计算 8 + 4。

英文

Added in version 3.0.

Callbacks can be added to any task using the link argument to apply_async:

add.apply_async((2, 2), link=other_task.s())

The callback will only be applied if the task exited successfully, and it will be applied with the return value of the parent task as argument.

As I mentioned earlier, any arguments you add to a signature, will be prepended to the arguments specified by the signature itself!

If you have the signature:

>>> sig = add.s(10)

then sig.delay(result) becomes:

>>> add.apply_async(args=(result, 10))

...

Now let's call our add task with a callback using partial arguments:

>>> add.apply_async((2, 2), link=add.s(8))

As expected this will first launch one task calculating 2 + 2, then another task calculating 8 + 4.

标记¶

Stamping

中文

Added in version 5.3.

Stamping API 的目标是提供为签名及其组成部分打标签的能力，以便于调试信息的标注。例如，在 canvas 是一个复杂结构时，可能需要为结构中的部分或全部元素打标签。复杂性在展开嵌套 group 或替换 chain 元素时进一步增加。在此类情况下，可能需要知道某个元素属于哪个 group 或它处于哪个嵌套层级。这就需要一个遍历 canvas 元素并以特定元数据标注它们的机制。 Stamping API 允许基于 Visitor 模式实现这一功能。

例如：

>>> sig1 = add.si(2, 2)
>>> sig1_res = sig1.freeze()
>>> g = group(sig1, add.si(3, 3))
>>> g.stamp(stamp='your_custom_stamp')
>>> res = g.apply_async()
>>> res.get(timeout=TIMEOUT)
[4, 6]
>>> sig1_res._get_task_meta()['stamp']
['your_custom_stamp']

这将初始化一个 group g，并为其组件打上 your_custom_stamp 标签。

要使该功能生效，你需要将配置项 result_extended 设置为 True，或使用配置指令 result_extended = True。

英文

Added in version 5.3.

The goal of the Stamping API is to give an ability to label the signature and its components for debugging information purposes. For example, when the canvas is a complex structure, it may be necessary to label some or all elements of the formed structure. The complexity increases even more when nested groups are rolled-out or chain elements are replaced. In such cases, it may be necessary to understand which group an element is a part of or on what nested level it is. This requires a mechanism that traverses the canvas elements and marks them with specific metadata. The stamping API allows doing that based on the Visitor pattern.

For example,

>>> sig1 = add.si(2, 2)
>>> sig1_res = sig1.freeze()
>>> g = group(sig1, add.si(3, 3))
>>> g.stamp(stamp='your_custom_stamp')
>>> res = g.apply_async()
>>> res.get(timeout=TIMEOUT)
[4, 6]
>>> sig1_res._get_task_meta()['stamp']
['your_custom_stamp']

will initialize a group g and mark its components with stamp your_custom_stamp.

For this feature to be useful, you need to set the result_extended configuration option to True or directive result_extended = True.

Canvas 标记¶

Canvas stamping

中文

你还可以通过自定义 stamping 逻辑来标记 canvas，方法是继承 visitor 类 StampingVisitor 以实现自定义 stamping visitor。

英文

We can also stamp the canvas with custom stamping logic, using the visitor class StampingVisitor as the base class for the custom stamping visitor.

自定义标记¶

Custom stamping

中文

如果需要更复杂的标记（stamping）逻辑，可以基于访问者模式（Visitor pattern）实现自定义标记行为。实现此类自定义逻辑的类必须继承自 StampingVisitor 并实现相应的方法。

例如，下面这个名为 InGroupVisitor 的示例将会给某个 group 中的任务打上 in_group 的标签：

class InGroupVisitor(StampingVisitor):
    def __init__(self):
        self.in_group = False

    def on_group_start(self, group, **headers) -> dict:
        self.in_group = True
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

    def on_group_end(self, group, **headers) -> None:
        self.in_group = False

    def on_chain_start(self, chain, **headers) -> dict:
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

    def on_signature(self, sig, **headers) -> dict:
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

下面是另一个自定义的标记访问器示例，它会给所有任务加上一个自定义的 monitoring_id。这个 monitoring_id 可以代表外部监控系统中的 UUID 值，可用于通过包含此 ID 来追踪任务执行情况。该 ID 可以是随机生成的 UUID，或是监控系统中的唯一 span ID 等标识符。

class MonitoringIdStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'monitoring_id': uuid4().hex}

备注

在 on_signature （或任何其他访问器方法）中返回的 stamped_headers 键用于指定应该打到任务上的 header。如果未指定此键，访问器会假设返回的字典中的所有键都是要标记的 header。

这意味着，下面的代码块行为等同于前面的示例。

class MonitoringIdStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'monitoring_id': uuid4().hex, 'stamped_headers': ['monitoring_id']}

接下来，我们看看如何使用 MonitoringIdStampingVisitor 示例访问器对不同的画布结构（canvas）进行打标：

sig_example = signature('t1')
sig_example.stamp(visitor=MonitoringIdStampingVisitor())

group_example = group([signature('t1'), signature('t2')])
group_example.stamp(visitor=MonitoringIdStampingVisitor())

chord_example = chord([signature('t1'), signature('t2')], signature('t3'))
chord_example.stamp(visitor=MonitoringIdStampingVisitor())

chain_example = chain(signature('t1'), group(signature('t2'), signature('t3')), signature('t4'))
chain_example.stamp(visitor=MonitoringIdStampingVisitor())

最后需要说明的是，上述示例中的每个任务会获得不同的 monitoring id。

英文

If more complex stamping logic is required, it is possible to implement custom stamping behavior based on the Visitor pattern. The class that implements this custom logic must inherit StampingVisitor and implement appropriate methods.

For example, the following example InGroupVisitor will label tasks that are in side of some group by label in_group.

class InGroupVisitor(StampingVisitor):
    def __init__(self):
        self.in_group = False

    def on_group_start(self, group, **headers) -> dict:
        self.in_group = True
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

    def on_group_end(self, group, **headers) -> None:
        self.in_group = False

    def on_chain_start(self, chain, **headers) -> dict:
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

    def on_signature(self, sig, **headers) -> dict:
        return {"in_group": [self.in_group], "stamped_headers": ["in_group"]}

The following example shows another custom stamping visitor, which labels all tasks with a custom monitoring_id which can represent a UUID value of an external monitoring system, that can be used to track the task execution by including the id with such a visitor implementation. This monitoring_id can be a randomly generated UUID, or a unique identifier of the span id used by the external monitoring system, etc.

class MonitoringIdStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'monitoring_id': uuid4().hex}

备注

The stamped_headers key returned in on_signature (or any other visitor method) is used to specify the headers that will be stamped on the task. If this key is not specified, the stamping visitor will assume all keys in the returned dictionary are the stamped headers from the visitor.

This means the following code block will result in the same behavior as the previous example.

class MonitoringIdStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'monitoring_id': uuid4().hex, 'stamped_headers': ['monitoring_id']}

Next, let's see how to use the MonitoringIdStampingVisitor example stamping visitor.

sig_example = signature('t1')
sig_example.stamp(visitor=MonitoringIdStampingVisitor())

group_example = group([signature('t1'), signature('t2')])
group_example.stamp(visitor=MonitoringIdStampingVisitor())

chord_example = chord([signature('t1'), signature('t2')], signature('t3'))
chord_example.stamp(visitor=MonitoringIdStampingVisitor())

chain_example = chain(signature('t1'), group(signature('t2'), signature('t3')), signature('t4'))
chain_example.stamp(visitor=MonitoringIdStampingVisitor())

Lastly, it's important to mention that each monitoring id stamp in the example above would be different from each other between tasks.

回调函数标记¶

Callbacks stamping

中文

标记 API 也支持对回调（callback）进行隐式标记。这意味着当某个回调添加到任务上时，访问器也会应用到该回调上。

警告

回调必须在标记之前链接到签名（signature）上。

例如，下面是一个自定义标记访问器的定义：

class CustomStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'header': 'value'}

    def on_callback(self, callback, **header) -> dict:
        return {'on_callback': True}

    def on_errback(self, errback, **header) -> dict:
        return {'on_errback': True}

这个自定义访问器将会对签名、回调和错误回调打上 {'header': 'value'}，并分别为回调和错误回调添加 {'on_callback': True} 与 {'on_errback': True}：

c = chord([add.s(1, 1), add.s(2, 2)], xsum.s())
callback = signature('sig_link')
errback = signature('sig_link_error')
c.link(callback)
c.link_error(errback)
c.stamp(visitor=CustomStampingVisitor())

该示例的最终结果为：

>>> c.options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.tasks.tasks[0].options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.tasks.tasks[1].options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.body.options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.body.options['link'][0].options
{'header': 'value', 'on_callback': True, 'stamped_headers': ['header', 'on_callback']}
>>> c.body.options['link_error'][0].options
{'header': 'value', 'on_errback': True, 'stamped_headers': ['header', 'on_errback']}

英文

The stamping API also supports stamping callbacks implicitly. This means that when a callback is added to a task, the stamping visitor will be applied to the callback as well.

警告

The callback must be linked to the signature before stamping.

For example, let's examine the following custom stamping visitor.

class CustomStampingVisitor(StampingVisitor):
    def on_signature(self, sig, **headers) -> dict:
        return {'header': 'value'}

    def on_callback(self, callback, **header) -> dict:
        return {'on_callback': True}

    def on_errback(self, errback, **header) -> dict:
        return {'on_errback': True}

This custom stamping visitor will stamp the signature, callbacks, and errbacks with {'header': 'value'} and stamp the callbacks and errbacks with {'on_callback': True} and {'on_errback': True} respectively as shown below.

c = chord([add.s(1, 1), add.s(2, 2)], xsum.s())
callback = signature('sig_link')
errback = signature('sig_link_error')
c.link(callback)
c.link_error(errback)
c.stamp(visitor=CustomStampingVisitor())

This example will result in the following stamps:

>>> c.options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.tasks.tasks[0].options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.tasks.tasks[1].options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.body.options
{'header': 'value', 'stamped_headers': ['header']}
>>> c.body.options['link'][0].options
{'header': 'value', 'on_callback': True, 'stamped_headers': ['header', 'on_callback']}
>>> c.body.options['link_error'][0].options
{'header': 'value', 'on_errback': True, 'stamped_headers': ['header', 'on_errback']}

Canvas：设计工作流¶

签名¶

部分代码¶

不变性¶

回调函数¶

原语¶

链¶

任务 ID¶

图表¶

组¶

组回调函数和错误处理¶

组结果¶

组展开¶

合集¶

错误处理¶

重要说明¶

map 和 starmap¶

数据块¶

标记¶

Canvas 标记¶

自定义标记¶

回调函数标记¶