Connector and Job Queue Best Practices

Savepoints and Requeuing

When running a job with a batch of records to import, sometimes the import could fail due to a data entry error from the source, or edge case we haven't handled. We should design our connectors so that these failures do not fail the entire batch. We can do this using exception handling and savepoints.

Reduce Lookups by Pre-Mapping Relationships

Avoid doing repeated lookups to tables like product.product in your import_record method. Instead, in your import_batch method, you can do pre-map the relationships by passing a list of external ID codes.

# Call the API for data.
api_records = adapter.get_page(page_number, adapter.item_url, params=params)

# Build a product mapping of the Productid to Odoo product Ids.
product_dict = mapper._get_map_products(product_codes=[x['Productid'] for x in api_records])

# The product_dict dictionary should contain a key value for every product that exists in Odoo.
# If your batch has a 1000 products, this dictionary will be built in one very fast query, vs
# 1000 small queries. This saves database resources, and speeds up your code.
# { # OpenTaps  # Odoo
#  '10106202': {
#     'binding_id': 1145, # cnf_product.product.template ID
#     'product_id': 1578 # product.tempate ID
#     'default_code': '10106202' # OpenTaps ID
#  }
# }

# With these results, you can fetch the mapped record using the product dictionary mapping before 
# passing your record onto the mapper.
record['product_id'] = products_dict.get(record['Productid'], {})

# Pass the record onto the mapper. The mapper can parse the rest, without having to do any DB 
# lookups.
internal_data = mapper.map_record(record).values()

Code Example

# Iterate over each record
for rec in api_records:
  try:
    # Use a savepoint. This is the equivelant of SAVEPOINT command in Postgresql, 
    # which allows the transaction to rollback gracefully.
    with self.env.cr.savepoint():
      # Mapping data and creating
      data_to_save = mapper.map_record(rec).values()
      self.create(data_to_save)

  except (IntegrityError, ValidationError, UserError, TypeError, AttributeError) as e:
    # Handle exception. IntegrityErrors from Postgres, Validation and UserError from Odoo, and python exceptions TypeError and AttributeError.
    # It is important to specify the type of error. Do not handle all exceptions using Exception.
    if len(api_records) == 1:
      # Since we will be requeuing individual failures, we'll check if the batch is one and raise the error
      raise e

    # Requeue the failing record so that the exception can be traced, while still allowing the rest of the batch to complete.
    self.with_delay(description=f"Sync: Import Rec {rec['RecordId']}").import_record([rec], backend)

Avoid Writing Everything

Because of the complexity of relationships between Odoo records and computed fields, it is important to be mindful when writing to the database. Each column write causes Odoo to evaluate whether to trigger a compute on other document types that have computed fields that api.depends() on the field you are updating. This can get expensive and slow down your sync.

To work around this, a method in the Base Adapter class has been modified to compare the data between your API record and the DB record, and then return a dictionary of just the values that should be written to the database.

# Get the values that need to be updated.
diff_dict = adapter.is_data_different(api_record=internal_data, db_record=binding)
if diff_dict:
  # If there are any different values, write the difference to the database.
  binding.with_context(no_need=True).write(diff_dict)

Dispatch to Channels

You should configure your method to use a specific channel. You can do this by defining an XML record. Your module should inherit from cnf_queue_job in order to refer to the CNF job queue channels.

<record id="job_function_sale_import_record" model="queue.job.function">
    <field name="model_id" ref="connector_cnf_inventory.model_cnf_pos_order" />
    <field name="method">import_record</field>
    <field name="channel_id" ref="cnf_queue_job.channel_record" />
    <field name="related_action" eval='{"func_name": "custom_related_action"}' />
    <field name="retry_pattern" eval="{1: 60, 2: 180, 3: 10, 5: 300}" />
</record>