Job Queue

Asegúrate de tener la librería requests.

• Usando variables de entorno y línea de comandos:
• Ajustar las variables de entorno (opcional):
• ODOO_QUEUE_JOB_CHANNELS=root:4 o cualquier otra configuración de canales. El valor predeterminado es root:1
• ODOO_QUEUE_JOB_PORT=8069, predeterminado --http-port
• ODOO_QUEUE_JOB_SCHEME=https, predeterminado http
• ODOO_QUEUE_JOB_HOST=load-balancer, predeterminado --http-interface o localhost si no se establece
• ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner, predeterminado vacío
• ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t, predeterminado vacío
• Iniciar Odoo con --load=web,queue_job y --workers mayores que 1.[^1]
• Usando el archivo de configuración de Odoo:

[options]
(...)
workers = 6
server_wide_modules = web,queue_job

(...)
[queue_job]
channels = root:2
scheme = https
host = load-balancer
port = 443
http_auth_user = jobrunner
http_auth_password = s3cr3t

• Confirmar que el runner se inicia correctamente comprobando el archivo de registro de odoo:

...INFO...queue_job.jobrunner.runner: starting
...INFO...queue_job.jobrunner.runner: initializing database connections
...INFO...queue_job.jobrunner.runner: queue job runner ready for db <dbname>
...INFO...queue_job.jobrunner.runner: database connections ready

• Crear trabajos (por ejemplo, usando base_import_async) y observar que se inician inmediatamente y en paralelo.
• Consejo: para habilitar el registro de depuración para el trabajo de cola, usar --log-handler=odoo.addons.queue_job:DEBUG

[^1]: También funciona con el servidor Odoo con hilos, aunque esta forma de ejecutar Odoo obviamente no es para propósitos de producción.

• Los trabajos que permanecen en estado enqueued o started (porque, por ejemplo, su worker ha sido eliminado) se volverán a poner en cola automáticamente.

Para usar este módulo, necesita:

• Ir al menú Job Queue

Desarrolladores

Retrasar trabajos

La forma rápida de encolar un trabajo para un método es usar with_delay() en un registro o modelo:

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

Aquí, el método print_confirmation_document() se ejecutará de forma asíncrona como un trabajo. with_delay() puede tomar varios parámetros para definir con más precisión cómo se ejecuta el trabajo (prioridad, ...).

Todos los argumentos pasados al método que se retrasa se almacenan en el trabajo y se pasan al método cuando se ejecuta de forma asíncrona, incluido self, por lo que el registro actual se mantiene durante la ejecución del trabajo (advertencia: el contexto no se conserva).

Se pueden expresar dependencias entre trabajos. Para iniciar un grafo de trabajos, use delayable() en un registro o modelo. Lo siguiente es el equivalente de with_delay() pero usando la forma larga:

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

Los métodos de los objetos Delayable se devuelven a sí mismos, por lo que se puede usar como patrón builder, lo que en algunos casos permite construir los trabajos dinámicamente:

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()
    )

La forma más sencilla de definir una dependencia es usar .on_done(job) en un Delayable:

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()

Los Delayables se pueden encadenar para formar grafos más complejos usando las primitivas chain() y group(). Una cadena representa una secuencia de trabajos a ejecutar en orden, un grupo representa trabajos que se pueden ejecutar en paralelo. Usar chain() tiene el mismo efecto que usar varios on_done() anidados pero es más legible. Ambos se pueden combinar para formar un grafo, por ejemplo podemos agrupar [A] de trabajos, que bloquea otro grupo [B] de trabajos. Cuando y solo cuando todos los trabajos del grupo [A] se ejecuten, se ejecutarán los trabajos del grupo [B]. El código se vería así:

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

Cuando ocurre un fallo en un grafo de trabajos, la ejecución de los trabajos que dependen del trabajo fallido se detiene. Permanecen en un estado wait_dependencies hasta que su trabajo "padre" tenga éxito. Esto puede ocurrir de dos formas: o bien el trabajo padre se reintenta y tiene éxito en un segundo intento, o bien el trabajo padre es "marcado como hecho" manualmente por un usuario. En estos dos casos, la dependencia se resuelve y el grafo continuará procesándose. Alternativamente, el trabajo fallido y todos sus trabajos dependientes pueden ser cancelados por un usuario. Los demás trabajos del grafo que no dependen del trabajo fallido continúan su ejecución en cualquier caso.

Nota: delay() debe llamarse en el delayable, chain o group que está en la parte superior del grafo. En el ejemplo anterior, si se llamara en group_a, entonces group_b nunca se retrasaría (pero se mostraría una advertencia).

También es posible dividir un trabajo en varios trabajos, cada uno procesando una parte del trabajo. Esto puede ser útil para evitar trabajos muy largos, paralelizar alguna tarea y obtener errores más específicos. El uso es el siguiente:

def button_split_delayable(self):
    (
        self  # Can be a big recordset, let's say 1000 records
        .delayable()
        .generate_thumbnail((50, 50))
        .set(priority=30)
        .set(description=_("generate xxx"))
        .split(50)  # Split the job in 20 jobs of 50 records each
        .delay()
    )

El método split() toma un argumento de palabra clave booleano chain. Si se establece en True, los trabajos se encadenarán, lo que significa que el siguiente trabajo solo comenzará cuando el anterior haya terminado:

def button_increment_var(self):
    (
        self
        .delayable()
        .increment_counter()
        .split(1, chain=True) # Will exceute the jobs one after the other
        .delay()
    )

Opciones de encolado de trabajos

• priority: el valor predeterminado es 10, cuanto más cerca esté de 0, más rápido se ejecutará
• eta: Estimated Time of Arrival del trabajo. No se ejecutará antes de esta fecha/hora
• max_retries: el valor predeterminado es 5, número máximo de reintentos antes de darse por vencido y establecer el estado del trabajo en 'failed'. Un valor de 0 significa reintentos infinitos.
• description: descripción humana del trabajo. Si no se establece, la descripción se calcula a partir de la documentación de la función o el nombre del método
• channel: el nombre completo del canal a usar para procesar la función. Si se especifica, anula el definido en la función
• identity_key: clave que identifica de forma única el trabajo, si se especifica y un trabajo con la misma clave aún no se ha ejecutado, el nuevo trabajo no se creará

Configurar opciones predeterminadas para los trabajos

En versiones anteriores, los trabajos se podían configurar usando el decorador @job. Esto ahora está obsoleto, se pueden configurar usando registros XML opcionales queue.job.function y queue.job.channel.

Ejemplo de canal:

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

Ejemplo de función de trabajo:

<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>

La forma general para el name es: <model.name>.method.

Las opciones de channel, related action y retry pattern son opcionales, se documentan a continuación.

Al escribir módulos, si 2 o más módulos añaden una función de trabajo o un canal con el mismo nombre (y padre para los canales), se fusionarán en el mismo registro, incluso si tienen xmlids diferentes. Al desinstalar, el registro fusionado se elimina cuando todos los módulos que lo usan se desinstalan.

Función de trabajo: model

Si la función se define en un modelo abstracto, no puede escribir <field name="model_id" ref="xml_id_of_the_abstract_model"</field> sino que tiene que definir una función para cada modelo que herede del modelo abstracto.

Función de trabajo: channel

El canal donde se retrasará el trabajo. El canal predeterminado es root.

Función de trabajo: related action

La Related Action aparece como un botón en la vista del Job. El botón ejecutará la acción definida.

La predeterminada es abrir la vista del registro relacionado con el trabajo (vista de formulario cuando hay un único registro, vista de lista para varios registros). En muchos casos, la related action predeterminada es suficiente y no necesita personalización, pero se puede personalizar proporcionando un diccionario en la función de trabajo:

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

• enable: cuando False, el botón no tiene efecto (predeterminado: True)
• func_name: nombre del método en queue.job que devuelve una acción
• kwargs: argumentos adicionales a pasar al método de related action

Ejemplo de código de related action:

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

Función de trabajo: retry pattern

Cuando un trabajo falla con un tipo de error reintentable, se reintenta automáticamente más tarde. De forma predeterminada, el reintento es siempre 10 minutos más tarde.

Se puede configurar un retry pattern en la función de trabajo. Lo que representa un patrón es "desde X intentos, posponer a Y segundos". Se expresa como un diccionario donde las claves son intentos y los valores son segundos a posponer como enteros:

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

Basándonos en esta configuración, podemos decir que:

• los 5 primeros reintentos se posponen 10 segundos más tarde
• los reintentos 5 a 10 se posponen 20 segundos más tarde
• los reintentos 10 a 15 se posponen 30 segundos más tarde
• todos los reintentos posteriores se posponen 5 minutos más tarde

Job Context

El contexto del recordset del trabajo, o de cualquier recordset pasado en argumentos de un trabajo, se transfiere al trabajo según una allow-list.

La allow-list predeterminada es ("tz", "lang", "allowed_company_ids", "force_company", "active_test"). Se puede personalizar en Base._job_prepare_context_before_enqueue_keys. Omitir trabajos en Odoo en ejecución

Cuando está desarrollando (es decir: módulos conector) puede que quiera omitir el queue job y ejecutar su código inmediatamente.

Para hacerlo puede establecer QUEUE_JOB__NO_DELAY=1 en su entorno.

Omitir trabajos en las pruebas

Cuando se escriben pruebas sobre métodos relacionados con trabajos, siempre es complicado lidiar con recordsets retrasados. Para facilitarle la vida en las pruebas puede establecer queue_job__no_delay=True en el contexto.

Consejo: puede hacer esto a nivel de caso de prueba así

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

Entonces todas sus pruebas ejecutan los métodos de trabajo de forma síncrona sin retrasar ningún trabajo.

Testing

Afirmar trabajos encolados

La forma recomendada de probar trabajos, en lugar de ejecutarlos directamente y de forma síncrona, es dividir las pruebas en dos partes:

> - una prueba donde el trabajo se simula (capturar trabajos con trap_jobs() y > la prueba solo verifica que el trabajo se ha retrasado con los > argumentos esperados > - una prueba que solo llama al método del trabajo de forma síncrona, para > validar únicamente el comportamiento adecuado de este método

Proceder de esta manera significa que puede demostrar que los trabajos se encolarán correctamente en tiempo de ejecución, y garantiza que su código no tenga un comportamiento diferente en las pruebas y en producción (porque ejecutar sus trabajos de forma síncrona puede tener un comportamiento diferente, ya que están en la misma transacción / en medio del método). Además, da más control sobre los argumentos que quiere pasar al llamar al método del trabajo (de forma síncrona, esta vez, en el segundo tipo de pruebas), y hace que las pruebas sean más pequeñas.

La mejor forma de ejecutar tales aserciones sobre los trabajos encolados es usar odoo.addons.queue_job.tests.common.trap_jobs().

Un ejemplo muy pequeño (más detalles en 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!")

Si lo prefiere, todavía puede probar todo el conjunto en una sola prueba, llamando a jobs_tester.perform_enqueued_jobs() en su prueba.

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!")

Ejecutar trabajos de forma síncrona cuando Odoo está en ejecución

Cuando está desarrollando (es decir: módulos conector) puede que quiera omitir el queue job y ejecutar su código inmediatamente.

Para hacerlo puede establecer QUEUE_JOB__NO_DELAY=1 en su entorno.

Advertencia

No haga esto en producción

Ejecutar trabajos de forma síncrona en las pruebas

Debería usar trap_jobs, de verdad, pero si por alguna razón no pudiera usarlo, y aún necesita que los métodos de trabajo se ejecuten de forma síncrona en sus pruebas, puede hacerlo estableciendo queue_job__no_delay=True en el contexto.

Consejo: puede hacer esto a nivel de caso de prueba así

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

Entonces todas sus pruebas ejecutan los métodos de trabajo de forma síncrona sin retrasar ningún trabajo.

En las pruebas tendrá que silenciar el logger así:

> @mute_logger('odoo.addons.queue_job.models.base')

Nota

en grafos de trabajos, la clave de contexto queue_job__no_delay debe estar en al menos un env de trabajo del grafo para que todo el grafo se ejecute de forma síncrona

Consejos y trucos

• Idempotencia (): El queue_job debe ser idempotente para que se pueda reintentar varias veces sin impacto en los datos.
• El trabajo debe probar al principio mismo su relevancia: el momento en que se ejecutará el trabajo es desconocido por diseño. Por lo tanto, la primera tarea de un trabajo debe ser comprobar si el trabajo relacionado sigue siendo relevante en el momento de la ejecución.

Patrones

A lo largo del tiempo, surgieron dos patrones principales:

• Para datos expuestos a los usuarios, un modelo debe almacenar los datos y el modelo debe ser el creador del trabajo. El trabajo se mantiene oculto a los usuarios
• Para datos técnicos, que no están expuestos a los usuarios, generalmente está bien crear directamente trabajos con datos pasados como argumentos al trabajo, sin modelos intermediarios.