The queue_job module is a critical component for implementing asynchronous task processing within the Odoo ecosystem. It facilitates the handling of resource-intensive operations—such as high-volume data imports, complex report generation, or batch updates—by executing them in the background. This approach ensures that the main Odoo threads remain unblocked, thereby significently improving system performance and user responsiveness.

Primary use cases for this module include:

• Bulk data import and export operations
• Generation and delivery of reports
• Batch processing of records
• Scheduled recurring tasks

Basic Implementation

The following example demonstrates how to define a model that utilizes the job decorator to enable asynchronous execution. The apply_async method is used to dispatch the task to the queue.

from odoo import models, fields, api
from odoo.addons.queue_job.job import job
import logging

_logger = logging.getLogger(__name__)

class AsyncJobDemo(models.Model):
    _name = 'async.job.demo'
    _description = 'Asynchronous Job Demo'

    label = fields.Char(string='Task Label', default="Async Task Demo")
    details = fields.Text(string='Details')

    @job(priority=10)
    def process_heavy_task(self, input_data, config_options):
        """
        The actual logic executed asynchronously.
        """
        _logger.info(f"Starting task: {self.label}, Data: {input_data}, Config: {config_options}")

        # Simulating a long-running process or potential failure
        if config_options.get('simulate_error'):
            raise ValueError("An error occurred during the asynchronous process.")
        
        _logger.info(f"Task {self.label} completed successfully.")
        return True

    def initiate_job(self):
        """
        Triggers the asynchronous method using apply_async.
        Calling this method directly does not execute the logic immediately.
        """
        self.process_heavy_task.apply_async(
            args=('sample_payload', {'simulate_error': False})
        )

    def schedule_recurring_job(self):
        """
        Schedules a task to run periodically.
        """
        self.process_heavy_task.apply_async(
            interval=3600,  # Run every hour
            nextcall=fields.Datetime.now(),
            args=('scheduled_payload', {'simulate_error': False})
        )

Error Handling and Retry Logic

To ensure robustness, jobs can be configured with automatic retry mechanisms. The decorator accepts parameters for maximum retry attempts and delays between attempts. This prevents transient failures from stopping the workflow entirely.

from odoo import models, fields
from odoo.addons.queue_job.job import job
import random

class ResilientJobDemo(models.Model):
    _name = 'resilient.job.demo'
    _description = 'Resilient Job Demo'

    name = fields.Char(string='Job Name', default="Resilient Task")
    notes = fields.Text(string='Notes')

    @job(max_retries=3, default_retry_delay=60)
    def execute_with_retry(self, param_a, param_b):
        """
        Executes a task with built-in retry logic.
        """
        _logger.info(f"Processing {self.name} with params: {param_a}, {param_b}")

        # Simulate a random failure to test the retry mechanism
        if random.choice([True, False]):
            _logger.warning("Task failed, queuing for retry...")
            raise Exception("Simulated failure for testing retries.")

        _logger.info(f"Task {self.name} finished.")
        return True

    def launch_resilient_task(self):
        """
        Initiates a task designed to handle failures gracefully.
        """
        self.execute_with_retry.apply_async(
            args=('test_data', 'retry_config')
        )

Modern Odoo Versions (v18 and later)

In newer versions of Odoo, the models.Model class includes the with_delay method. This provides a cleaner, more Pythonic way to enqueue jobs with various options such as priority, scheduled execution time (eta), and deduplication keys.

from datetime import datetime, timedelta

class ModernAsyncDemo(models.Model):
    _name = 'modern.async.demo'
    _description = 'Modern Async Demo'

    file_hash = fields.Char(string='File Hash')
    
    def _enqueue_analysis(self):
        # Method to encapsulate the delay configuration
        return self.with_delay(
            priority=20,  # Higher priority number
            eta=datetime.now() + timedelta(minutes=5),  # Delay execution by 5 minutes
            max_retries=5,
            description=f"Analyzing File Hash [{self.file_hash}]",
            channel='root.data_processing',  # Specific channel for workers
            identity_key=self.file_hash,  # Ensures unique job per hash
        )

    def start_analysis(self):
        """
        Enqueues the actual analysis method.
        """
        self._enqueue_analysis().perform_complex_analysis()
        
    def perform_complex_analysis(self):
        # Logic for the analysis goes here
        pass