odoo12-addon-queue-job 12.0.4.1.0

延迟作业

将任务入队到方法的一个快速方法是使用记录或模型上的 with_delay()

def button_done(self):
    self.with_delay().print_confirmation_document(self.state)
    self.write({"state": "done"})
    return True

在这里，方法 print_confirmation_document() 将作为任务异步执行。 with_delay() 可以接受多个参数来更精确地定义任务的执行方式（优先级，……）。

传递给被延迟方法的全部参数都存储在任务中，并在异步执行时传递给该方法，包括 self，因此在工作执行期间保持当前记录（注意：上下文不会保留）。

可以在任务之间表达依赖关系。要启动任务图，请在记录或模型上使用 delayable()

def button_done(self):
    delayable = self.delayable()
    delayable.print_confirmation_document(self.state)
    delayable.delay()
    self.write({"state": "done"})
    return True

Delayable 对象的方法返回自身，因此它可以作为构建器模式使用，在某些情况下允许动态构建任务。

def button_generate_simple_with_delayable(self):
    self.ensure_one()
    # Introduction of a delayable object, using a builder pattern
    # allowing to chain jobs or set properties. The delay() method
    # on the delayable object actually stores the delayable objects
    # in the queue_job table
    (
        self.delayable()
        .generate_thumbnail((50, 50))
        .set(priority=30)
        .set(description=_("generate xxx"))
        .delay()
    )

定义依赖关系的最简单方法是使用 Delayable 上的 .on_done(job)

def button_chain_done(self):
    self.ensure_one()
    job1 = self.browse(1).delayable().generate_thumbnail((50, 50))
    job2 = self.browse(1).delayable().generate_thumbnail((50, 50))
    job3 = self.browse(1).delayable().generate_thumbnail((50, 50))
    # job 3 is executed when job 2 is done which is executed when job 1 is done
    job1.on_done(job2.on_done(job3)).delay()

可以使用 chain() 和 group() 基本操作将 Delayable 链接起来，形成更复杂的图。一个链表示按顺序执行的作业序列，一个组表示可以并行执行的作业。使用 chain() 与使用多个嵌套的 on_done() 具有相同的效果，但更易读。两者可以组合形成一个图，例如，我们可以分组 [A] 的作业，这将阻塞另一个组 [B] 的作业。只有当组 [A] 中的所有作业都执行完毕时，组 [B] 的作业才会执行。代码如下

from odoo.addons.queue_job.delay import group, chain

def button_done(self):
    group_a = group(self.delayable().method_foo(), self.delayable().method_bar())
    group_b = group(self.delayable().method_baz(1), self.delayable().method_baz(2))
    chain(group_a, group_b).delay()
    self.write({"state": "done"})
    return True

当作业图发生故障时，依赖于已失败作业的作业执行将停止。它们将保持 wait_dependencies 状态，直到它们的“父”作业成功。这可以通过两种方式发生：要么父作业重试并在第二次尝试中成功，要么用户手动将父作业“设置为完成”。在这两种情况下，依赖关系得到解决，图将继续处理。或者，用户可以取消失败作业及其所有依赖作业。在所有情况下，不依赖于失败作业的图中的其他作业将继续执行。

注意：delay()必须在图顶层的可延迟对象、链或组上调用。在上面的示例中，如果它在group_a上调用，那么group_b将永远不会被延迟（但会显示警告）。

作业排队选项

• 优先级：默认为10，越接近0，执行越快

• eta：作业到达的估计时间。在此日期/时间之前不会执行

• max_retries：默认为5，在放弃并设置作业状态为‘失败’之前的最大重试次数。0表示无限重试。

• description：作业的人类描述。如果未设置，则描述将从函数文档或方法名称计算得出

• channel：用于处理函数的通道的完整名称。如果指定，则覆盖函数上定义的通道

• identity_key：唯一标识作业的键，如果指定且尚未运行具有相同键的作业，则不会创建新的作业

配置作业的默认选项

在早期版本中，可以使用@job装饰器配置作业。现在已弃用，它们可以使用可选的queue.job.function和queue.job.channel XML记录进行配置。

通道示例

<record id="channel_sale" model="queue.job.channel">
    <field name="name">sale</field>
    <field name="parent_id" ref="queue_job.channel_root" />
</record>

作业函数示例

<record id="job_function_sale_order_action_done" model="queue.job.function">
    <field name="model_id" ref="sale.model_sale_order" />
    <field name="method">action_done</field>
    <field name="channel_id" ref="channel_sale" />
    <field name="related_action" eval='{"func_name": "custom_related_action"}' />
    <field name="retry_pattern" eval="{1: 60, 2: 180, 3: 10, 5: 300}" />
</record>

name的一般形式为：<model.name>.method。

通道、相关操作和重试模式选项是可选的，它们在下面有文档说明。

在编写模块时，如果2个或更多模块添加具有相同名称（以及通道的父级）的作业函数或通道，它们将在同一记录中合并，即使它们有不同的xmlids。在卸载时，当所有使用它的模块都卸载时，合并的记录将被删除。

作业函数：通道

作业将被延迟的通道。默认通道是root。

作业函数：相关操作

“相关操作”作为按钮出现在作业视图中。按钮将执行定义的操作。

默认情况下，该按钮将打开与作业相关的记录视图（存在单个记录时为表单视图，存在多个记录时为列表视图）。在许多情况下，默认相关操作就足够了，不需要自定义，但可以通过在作业函数上提供一个字典来自定义它

{
    "enable": False,
    "func_name": "related_action_partner",
    "kwargs": {"name": "Partner"},
}

• enable：当False时，按钮无效果（默认：True）

• func_name：返回操作的queue.job上方法的名称

• kwargs：传递给相关操作方法的额外参数

相关操作代码示例

class QueueJob(models.Model):
    _inherit = 'queue.job'

    def related_action_partner(self, name):
        self.ensure_one()
        model = self.model_name
        partner = self.records
        action = {
            'name': name,
            'type': 'ir.actions.act_window',
            'res_model': model,
            'view_type': 'form',
            'view_mode': 'form',
            'res_id': partner.id,
        }
        return action

作业函数：重试模式

当作业因可重试的错误类型失败时，它会自动稍后重试。默认情况下，重试总是10分钟后。

可以在作业函数上配置重试模式。一个模式代表的是“从X次尝试开始，延迟Y秒”。它以字典的形式表示，其中键是尝试次数，值是延迟的秒数（整数）

{
    1: 10,
    5: 20,
    10: 30,
    15: 300,
}

根据此配置，我们可以知道

• 前5次重试延迟10秒

• 第5次到第10次重试延迟20秒

• 第10次到第15次重试延迟30秒

• 所有后续的重试延迟5分钟

绕过运行中的Odoo上的作业

当你正在开发（例如：连接器模块）时，你可能想绕过队列作业并立即运行你的代码。

要这样做，你可以在你的环境中设置TEST_QUEUE_JOB_NO_DELAY=1。

在测试中绕过作业

在编写与作业相关的方法的测试时，总是难以处理延迟的记录集。为了使你的测试生活更容易，你可以在上下文中设置test_queue_job_no_delay=True。

提示：你可以在测试用例级别这样做

@classmethod
def setUpClass(cls):
    super().setUpClass()
    cls.env = cls.env(context=dict(
        cls.env.context,
        test_queue_job_no_delay=True,  # no jobs thanks
    ))

然后，所有测试将同步执行作业方法，不会延迟任何作业。

测试

断言已入队的作业

测试作业的推荐方法，而不是直接同步运行它们，是将测试分为两部分

• 一个测试中作业被模拟（使用 trap_jobs() 捕获作业，测试仅验证作业是否已使用预期的参数被延迟

• 一个仅同步调用作业方法的测试，以验证该方法的正确行为

这样进行意味着你可以证明作业将在运行时正确入队，并确保你的代码在测试和生产环境中具有相同的行为（因为同步运行作业可能具有不同的行为，因为它们处于相同的交易中/方法中间）。此外，它为你提供了更多控制权，在调用作业方法的参数（这次是在第二种类型的测试中同步调用），并且使测试更小。

在入队的作业上进行这种断言的最佳方式是使用 odoo.addons.queue_job.tests.common.trap_jobs()。

一个非常小的示例（更多详细信息请参阅 tests/common.py）

# code
def my_job_method(self, name, count):
    self.write({"name": " ".join([name] * count)

def method_to_test(self):
    count = self.env["other.model"].search_count([])
    self.with_delay(priority=15).my_job_method("Hi!", count=count)
    return count

# tests
from odoo.addons.queue_job.tests.common import trap_jobs

# first test only check the expected behavior of the method and the proper
# enqueuing of jobs
def test_method_to_test(self):
    with trap_jobs() as trap:
        result = self.env["model"].method_to_test()
        expected_count = 12

        trap.assert_jobs_count(1, only=self.env["model"].my_job_method)
        trap.assert_enqueued_job(
            self.env["model"].my_job_method,
            args=("Hi!",),
            kwargs=dict(count=expected_count),
            properties=dict(priority=15)
        )
        self.assertEqual(result, expected_count)


 # second test to validate the behavior of the job unitarily
 def test_my_job_method(self):
     record = self.env["model"].browse(1)
     record.my_job_method("Hi!", count=12)
     self.assertEqual(record.name, "Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi!")

如果你愿意，你仍然可以在单个测试中测试整个流程，通过在测试中调用 jobs_tester.perform_enqueued_jobs()。

def test_method_to_test(self):
    with trap_jobs() as trap:
        result = self.env["model"].method_to_test()
        expected_count = 12

        trap.assert_jobs_count(1, only=self.env["model"].my_job_method)
        trap.assert_enqueued_job(
            self.env["model"].my_job_method,
            args=("Hi!",),
            kwargs=dict(count=expected_count),
            properties=dict(priority=15)
        )
        self.assertEqual(result, expected_count)

        trap.perform_enqueued_jobs()

        record = self.env["model"].browse(1)
        record.my_job_method("Hi!", count=12)
        self.assertEqual(record.name, "Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi!")

运行 Odoo 时同步执行作业

当你正在开发（例如：连接器模块）时，你可能想绕过队列作业并立即运行你的代码。

为此，你可以在你的环境中设置 TEST_QUEUE_JOB_NO_DELAY=1。

在测试中同步执行作业

你应该使用 trap_jobs，真的，但如果由于任何原因你不能使用它，并且仍然需要在测试中同步执行作业方法，你可以在上下文中设置 test_queue_job_no_delay=True。

提示：你可以在测试用例级别这样做

@classmethod
def setUpClass(cls):
    super().setUpClass()
    cls.env = cls.env(context=dict(
        cls.env.context,
        test_queue_job_no_delay=True,  # no jobs thanks
    ))

然后，所有测试将同步执行作业方法，不会延迟任何作业。

在测试中，你需要像这样静音记录器

@mute_logger(‘odoo.addons.queue_job.models.base’)

技巧和窍门

• 幂等性 (https://www.restapitutorial.com/lessons/idempotency.html)：queue_job 应该是幂等的，这样它们可以在不影响数据的情况下重试多次。

• 作业应该在开始时测试其相关性：作业将执行的时间是未知的，这是故意的。因此，作业的第一个任务应该是检查相关的工作在执行时刻是否仍然相关。

模式

随着时间的推移，出现了两种主要模式

1. 对于向用户公开的数据，模型应存储数据，模型应该是作业的创建者。作业对用户是隐藏的

2. 对于不向用户公开的技术数据，通常可以直接创建直接带有作为参数传递给作业的数据的作业，而不需要中间模型。