name	odoo-connector-module-creator
description	Creates and enhances Odoo 16.0 connector modules that integrate with external systems (e-commerce, logistics, accounting, CRM) using the `generic_connector` framework

Odoo Connector Module Creator and Enhancer

Description

Creates and enhances Odoo 16.0 connector modules that integrate with external systems (e-commerce, logistics, accounting, CRM) using the generic_connector framework. This skill handles:

New Connector Creation: Build complete integration modules for Shopify, WooCommerce, Amazon, or any external API
Connector Enhancement: Add features like inventory sync, webhook support, or new entity types to existing connectors
Troubleshooting: Debug sync issues, API errors, authentication problems, and queue job failures
Architecture Implementation: Properly implement binding models, adapters, mappers, and importers/exporters

The skill leverages production-tested patterns from reference connectors (zid_connector_v2, beatroute_connector) and provides automated scripts for generating boilerplate code.

Overview

Create production-ready Odoo 16.0 connector modules that integrate with external systems using the generic_connector framework. Handle creation of new connectors, enhancement of existing connectors, troubleshooting sync issues, and debugging integration problems.

When to Use This Skill

Use this skill when the user requests:

Creating new connectors: "Create a Shopify connector", "Build WooCommerce integration", "Connect to Amazon API"
Enhancing connectors: "Add inventory sync to zid_connector", "Implement webhooks for orders", "Add product export"
Adding entities: "Add customer sync to the connector", "Import invoices from the external system"
Troubleshooting: "Orders aren't importing", "Webhook signature verification failing", "Fix sync errors"
Debugging: "Why is the API returning 401?", "Products are duplicating", "Queue jobs not running"

Key Concepts

Generic Connector Framework

All connector modules extend generic_connector, which provides:

Backend Model - Configuration and orchestration
Binding Models - Link Odoo records to external entities
Adapter Component - HTTP client for API communication
Mapper Components - Data transformation (import/export)
Importer/Exporter Components - Sync logic
Webhook System - Real-time event processing
Queue Job Integration - Async operations

Reference Code

Three production connectors serve as references:

/Users/jamshid/PycharmProjects/Siafa/odoo16e_simc/addons-connector/generic_connector - Base framework
/Users/jamshid/PycharmProjects/Siafa/odoo16e_simc/addons-connector/zid_connector_v2 - E-commerce example
/Users/jamshid/PycharmProjects/Siafa/odoo16e_simc/addons-connector/beatroute_connector - Logistics example

Workflow

Creating a New Connector

When the user requests a new connector:

Step 1: Gather Requirements

External system name (e.g., "Shopify", "WooCommerce")
Connector type: ecommerce, logistics, accounting, crm
Entities to sync: products, orders, customers, inventory
Sync direction: import, export, or bidirectional
Authentication method: API key, OAuth, basic auth
API documentation URL (if available)

Step 2: Initialize Module

# Use the init_connector.py script
python3 scripts/init_connector.py <connector_name> --path <output_path> --type <connector_type>

# Example:
python3 scripts/init_connector.py shopify --path ~/odoo/addons --type ecommerce

Step 3: Review Generated Structure

The script creates:

shopify_connector/
├── __manifest__.py              # Module metadata
├── __init__.py                  # Python imports
├── models/
│   ├── backend.py              # Backend configuration
│   ├── adapter.py              # API client
│   ├── product_binding.py      # Product sync
│   └── __init__.py
├── views/
│   ├── backend_views.xml       # Backend UI
│   ├── binding_views.xml       # Binding UI
│   └── menu_views.xml          # Menu structure
├── security/
│   ├── security.xml            # Access groups
│   └── ir.model.access.csv     # Access rules
├── wizards/
│   ├── sync_wizard.py          # Manual sync wizard
│   └── __init__.py
├── data/
│   ├── ir_cron_data.xml        # Scheduled jobs
│   └── queue_job_function_data.xml
└── README.md

Step 4: Customize Backend Model

Edit models/backend.py:

Update API configuration fields to match the external system:

# Example for Shopify
shop_url = fields.Char(string='Shop URL', required=True)
api_version = fields.Selection([
    ('2024-01', '2024-01'),
    ('2024-04', '2024-04'),
], default='2024-04')

Implement template methods:

def _test_connection_implementation(self):
    """Test API connection."""
    adapter = self.get_adapter('shopify.adapter')
    return adapter.test_connection()

def _sync_orders_implementation(self):
    """Import orders."""
    with self.work_on('shopify.sale.order') as work:
        importer = work.component(usage='batch.importer')
        return importer.run()

Step 5: Implement Adapter

Edit models/adapter.py:

Configure authentication (see references/authentication.md for patterns):

def get_api_headers(self):
    headers = super().get_api_headers()
    headers.update({
        'X-Shopify-Access-Token': self.backend_record.api_key,
        'Content-Type': 'application/json',
    })
    return headers

Add CRUD methods for each entity type:

def get_products(self, filters=None):
    """Fetch products from Shopify."""
    return self.get('/admin/api/2024-01/products.json', params=filters)

def create_order(self, data):
    """Create order in Shopify."""
    return self.post('/admin/api/2024-01/orders.json', data={'order': data})

Handle pagination (see references/api_integration.md):

def get_all_products(self):
    """Fetch all products with pagination."""
    # Implement based on API pagination style

Step 6: Create Mapper Components

Create components/mapper.py:

from odoo.addons.generic_connector.components.mapper import GenericImportMapper

class ProductImportMapper(GenericImportMapper):
    _name = 'shopify.product.import.mapper'
    _inherit = 'generic.import.mapper'
    _apply_on = 'shopify.product.template'

    direct = [
        ('title', 'name'),
        ('vendor', 'manufacturer'),
    ]

    @mapping
    def backend_id(self, record):
        return {'backend_id': self.backend_record.id}

    @mapping
    def price(self, record):
        variants = record.get('variants', [])
        if variants:
            return {'list_price': float(variants[0].get('price', 0))}
        return {}

Step 7: Implement Importer Components

Create components/importer.py:

from odoo.addons.generic_connector.components.importer import GenericImporter

class ProductImporter(GenericImporter):
    _name = 'shopify.product.importer'
    _inherit = 'generic.importer'
    _apply_on = 'shopify.product.template'

    def _import_record(self, external_id, force=False):
        # Fetch from external system
        adapter = self.component(usage='backend.adapter')
        external_data = adapter.get_product(external_id)

        # Transform data
        mapper = self.component(usage='import.mapper')
        mapped_data = mapper.map_record(external_data).values()

        # Create or update binding
        binding = self._get_binding()
        if binding:
            binding.write(mapped_data)
        else:
            binding = self.model.create(mapped_data)

        return binding

Step 8: Register Components

Create components/__init__.py:

from . import adapter
from . import mapper
from . import importer
from . import exporter

Update main __init__.py:

from . import models
from . import wizards
from . import components

Step 9: Test the Connector

# Install module
odoo-bin -c odoo.conf -d test_db -i shopify_connector

# Test in Odoo UI
# 1. Go to Connector > Shopify > Backends
# 2. Create a new backend
# 3. Configure API credentials
# 4. Click "Test Connection"
# 5. Click "Sync All"

Enhancing an Existing Connector

When the user wants to add functionality to an existing connector:

Step 1: Identify Enhancement Type

Adding a new entity (orders, customers, invoices)
Adding a new feature (webhooks, batch export)
Fixing bugs or improving performance
Adding authentication method

Step 2: Add New Entity Binding

Use the add_binding.py script:

python3 scripts/add_binding.py <connector_path> <entity_name> --odoo-model <model>

# Example:
python3 scripts/add_binding.py ~/odoo/addons/shopify_connector customer --odoo-model res.partner

This generates:

models/customer_binding.py - Binding model
views/customer_views.xml - UI views
Updates to __manifest__.py and security files
Adapter methods to implement manually

Step 3: Implement Components

Follow steps 6-7 from "Creating a New Connector" to implement mapper and importer/exporter for the new entity.

Step 4: Add to Backend Orchestration

Update models/backend.py:

def _sync_customers_implementation(self):
    """Import customers."""
    with self.work_on('shopify.res.partner') as work:
        importer = work.component(usage='batch.importer')
        return importer.run()

def action_sync_all(self):
    """Override to include customers."""
    super().action_sync_all()
    self.with_delay().sync_customers()

Implementing Webhooks

When the user requests webhook support:

Step 1: Create Webhook Controller

Create controllers/webhook_controller.py:

from odoo import http
from odoo.http import request
import json
import logging

_logger = logging.getLogger(__name__)

class ShopifyWebhookController(http.Controller):
    @http.route('/shopify/webhook', type='json', auth='none', csrf=False)
    def webhook(self):
        """Handle Shopify webhooks."""
        try:
            payload = request.httprequest.get_data(as_text=True)
            topic = request.httprequest.headers.get('X-Shopify-Topic')
            hmac_header = request.httprequest.headers.get('X-Shopify-Hmac-SHA256')

            # Find backend
            shop_domain = request.httprequest.headers.get('X-Shopify-Shop-Domain')
            backend = request.env['shopify.backend'].sudo().search([
                ('shop_url', 'ilike', shop_domain)
            ], limit=1)

            if not backend:
                return {'error': 'Backend not found'}, 404

            # Verify signature
            if not self._verify_webhook(payload, hmac_header, backend.webhook_secret):
                return {'error': 'Invalid signature'}, 401

            # Create webhook record
            webhook = request.env['generic.webhook'].sudo().create({
                'backend_id': backend.id,
                'event_type': topic,
                'payload': payload,
                'signature': hmac_header,
                'processing_status': 'pending',
            })

            # Process asynchronously
            webhook.with_delay().process_webhook()

            return {'status': 'accepted', 'webhook_id': webhook.id}

        except Exception as e:
            _logger.exception("Webhook processing failed")
            return {'error': str(e)}, 500

    def _verify_webhook(self, payload, hmac_header, secret):
        """Verify HMAC-SHA256 signature."""
        import hmac
        import hashlib
        import base64

        computed = hmac.new(
            secret.encode('utf-8'),
            payload.encode('utf-8'),
            hashlib.sha256
        ).digest()

        computed_base64 = base64.b64encode(computed).decode()

        return hmac.compare_digest(computed_base64, hmac_header)

Step 2: Add Webhook Processing to Backend

Update models/backend.py:

def process_webhook(self, webhook):
    """Process webhook by topic."""
    handlers = {
        'orders/create': self._handle_order_created,
        'orders/updated': self._handle_order_updated,
        'products/update': self._handle_product_updated,
    }

    handler = handlers.get(webhook.event_type)
    if handler:
        try:
            handler(webhook)
            webhook.mark_as_processed()
        except Exception as e:
            _logger.exception("Webhook handler failed")
            webhook.mark_as_failed(str(e))
    else:
        webhook.mark_as_ignored(f"No handler for {webhook.event_type}")

def _handle_order_created(self, webhook):
    """Handle orders/create webhook."""
    payload = json.loads(webhook.payload)
    order_id = payload['id']

    # Import the order
    self.env['shopify.sale.order'].import_record(
        backend=self,
        external_id=str(order_id)
    )

Implementing Export Using Shared Wizard

The connector_base_backend module provides a shared export wizard (connector.export.wizard) that works across all connectors without requiring custom UI for each one.

Architecture Overview

The export system uses delegation inheritance to route export requests:

User clicks "Export to Connectors" on product.product
    ↓
connector.export.wizard opens (shared UI)
    ↓
User selects backend (e.g., ZID, Shopify)
    ↓
wizard.action_export() calls backend.export_records(model_name, record_ids)
    ↓
connector.base.backend routes to concrete implementation
    ↓ (via _inherits delegation chain)
generic.backend (intermediate)
    ↓
zid.backend.export_product_product(record_ids)
    ↓
Creates bindings + queues async exports

Inheritance Chain:

connector.base.backend (has export_records() router)
    ↓ _inherits via base_backend_id
generic.backend (intermediate layer)
    ↓ _inherits via generic_backend_id
your_connector.backend (concrete implementation)

Step-by-Step Implementation

Step 1: Understand the Routing Mechanism

The connector.base.backend.export_records() method automatically routes to your backend:

# In connector_base_backend/models/connector_base_backend.py
def export_records(self, model_name, record_ids):
    """Generic export method that routes to specific connector implementations"""
    method_name = f'export_{model_name.replace(".", "_")}'

    # Find concrete backend via _inherits chain
    concrete_backend = self
    for model in self._inherits_children:
        child = self.env[model].search([('base_backend_id', '=', self.id)], limit=1)
        if child:
            concrete_backend = child
            break

    # Call export_product_product(), export_sale_order(), etc.
    if hasattr(concrete_backend, method_name):
        return getattr(concrete_backend, method_name)(record_ids)
    else:
        raise UserError(_(
            "Export not implemented for model %s in connector %s"
        ) % (model_name, concrete_backend.name))

Step 2: Implement Export Methods in Your Backend

For each Odoo model you want to export, implement export_<model_name>() in your backend model:

Example: Export Products

Add to models/backend.py:

def export_product_product(self, record_ids):
    """
    Export product.product records to external system.

    Called by connector.export.wizard when exporting products.

    Args:
        record_ids: List of product.product IDs to export

    Returns:
        dict: Notification action
    """
    self.ensure_one()

    if not record_ids:
        return self._build_notification(
            _('Export Products'),
            _('No products selected for export'),
            'warning'
        )

    products = self.env['product.product'].browse(record_ids)
    exported_count = 0
    created_bindings = 0
    skipped_count = 0
    errors = []

    for product in products:
        try:
            # Find or create binding
            binding = self.env['shopify.product.product'].search([
                ('backend_id', '=', self.id),
                ('odoo_id', '=', product.id)
            ], limit=1)

            if not binding:
                # Create new binding
                binding_vals = {
                    'backend_id': self.id,
                    'odoo_id': product.id,
                    'external_sku': product.default_code or '',
                    'external_name': product.name,
                    'external_price': product.list_price,
                    'external_status': 'active' if product.active else 'inactive',
                }
                binding = self.env['shopify.product.product'].create(binding_vals)
                created_bindings += 1

            # Skip if marked as no_export
            if binding.no_export:
                skipped_count += 1
                continue

            # Queue async export
            binding.with_delay()._export_to_external()
            exported_count += 1

        except Exception as e:
            errors.append(f'Product {product.name}: {str(e)}')
            _logger.error(f'Export failed for {product.name}: {e}', exc_info=True)

    # Build response message
    message_parts = []
    if exported_count > 0:
        message_parts.append(
            _('%d product(s) scheduled for export') % exported_count
        )
    if created_bindings > 0:
        message_parts.append(_('%d new binding(s) created') % created_bindings)
    if skipped_count > 0:
        message_parts.append(_('%d skipped (no_export)') % skipped_count)
    if errors:
        message_parts.append(_('Errors: %d') % len(errors))

    message = '. '.join(message_parts)
    notif_type = 'success' if exported_count > 0 and not errors else 'warning'

    # Update statistics
    if exported_count > 0:
        self.last_export_date = datetime.now()

    return self._build_notification(_('Export Products'), message, notif_type)

Example: Export Partners

def export_res_partner(self, record_ids):
    """Export res.partner records to external system."""
    self.ensure_one()

    partners = self.env['res.partner'].browse(record_ids)
    exported_count = 0

    for partner in partners:
        # Find or create partner binding
        binding = self.env['shopify.res.partner'].search([
            ('backend_id', '=', self.id),
            ('odoo_id', '=', partner.id)
        ], limit=1)

        if not binding:
            binding = self.env['shopify.res.partner'].create({
                'backend_id': self.id,
                'odoo_id': partner.id,
            })

        # Queue export
        binding.with_delay()._export_to_external()
        exported_count += 1

    return self._build_notification(
        _('Export Customers'),
        _('%d customer(s) scheduled for export') % exported_count,
        'success'
    )

Example: Export Sale Orders

def export_sale_order(self, record_ids):
    """Export sale.order records to external system."""
    self.ensure_one()

    orders = self.env['sale.order'].browse(record_ids)
    exported_count = 0

    for order in orders:
        # Validate order state
        if order.state not in ['sale', 'done']:
            _logger.warning(f'Skipping order {order.name}: not confirmed')
            continue

        # Find or create order binding
        binding = self.env['shopify.sale.order'].search([
            ('backend_id', '=', self.id),
            ('odoo_id', '=', order.id)
        ], limit=1)

        if not binding:
            binding = self.env['shopify.sale.order'].create({
                'backend_id': self.id,
                'odoo_id': order.id,
            })

        # Export dependencies first (customer, products)
        self._export_order_dependencies(order)

        # Queue order export
        binding.with_delay()._export_to_external()
        exported_count += 1

    return self._build_notification(
        _('Export Orders'),
        _('%d order(s) scheduled for export') % exported_count,
        'success'
    )

def _export_order_dependencies(self, order):
    """Export customer and products before exporting order."""
    # Export customer
    if order.partner_id:
        self.export_res_partner([order.partner_id.id])

    # Export products
    product_ids = order.order_line.mapped('product_id').ids
    if product_ids:
        self.export_product_product(product_ids)

Step 3: The Shared Wizard is Already Configured

The connector_base_backend module already includes action bindings:

<!-- In connector_base_backend/wizards/connector_export_wizard_view.xml -->

<!-- Export action for products -->
<record id="action_connector_export_wizard_product" model="ir.actions.act_window">
    <field name="name">Export to Connectors</field>
    <field name="res_model">connector.export.wizard</field>
    <field name="view_mode">form</field>
    <field name="target">new</field>
    <field name="binding_model_id" ref="product.model_product_product"/>
    <field name="binding_view_types">list,form</field>
</record>

<!-- Export action for partners -->
<record id="action_connector_export_wizard" model="ir.actions.act_window">
    <field name="name">Export to Connectors</field>
    <field name="res_model">connector.export.wizard</field>
    <field name="view_mode">form</field>
    <field name="target">new</field>
    <field name="binding_model_id" ref="base.model_res_partner"/>
    <field name="binding_view_types">list,form</field>
</record>

To add export for other models, create similar actions in your connector or in connector_base_backend:

<!-- Export action for sale orders -->
<record id="action_connector_export_wizard_sale_order" model="ir.actions.act_window">
    <field name="name">Export to Connectors</field>
    <field name="res_model">connector.export.wizard</field>
    <field name="view_mode">form</field>
    <field name="target">new</field>
    <field name="binding_model_id" ref="sale.model_sale_order"/>
    <field name="binding_view_types">list,form</field>
</record>

Step 4: Testing the Export

# 1. Update your connector module
odoo-bin -c odoo.conf -d your_db -u your_connector

# 2. Test from UI
# Navigate to: Inventory → Products → Products
# Select one or more products
# Click: Action → Export to Connectors
# Select your backend
# Click: Export

# 3. Verify in logs
tail -f /var/log/odoo/odoo.log | grep "export\|binding"

# 4. Check queue jobs
# Navigate to: Queue Jobs → Jobs
# Look for: your_connector.product.product._export_to_external

# 5. Check bindings created
# Navigate to: Connector → Your Connector → Products
# Verify bindings were created with correct external_id

Step 5: Advanced Export Patterns

Pattern 1: Conditional Export

def export_product_product(self, record_ids):
    """Export only published products."""
    products = self.env['product.product'].browse(record_ids)

    # Filter products
    exportable_products = products.filtered(
        lambda p: p.active and getattr(p, 'website_published', True)
    )

    if len(exportable_products) < len(products):
        skipped = len(products) - len(exportable_products)
        _logger.info(f'Skipped {skipped} unpublished products')

    # Export only exportable products
    for product in exportable_products:
        # ... create binding and export

Pattern 2: Batch Export with Progress

def export_product_product(self, record_ids):
    """Export products in batches."""
    products = self.env['product.product'].browse(record_ids)
    batch_size = 50

    for i in range(0, len(products), batch_size):
        batch = products[i:i + batch_size]
        # Process batch with delay
        self.with_delay()._export_product_batch(batch.ids)

    return self._build_notification(
        _('Export Products'),
        _('Queued %d products in %d batches') % (
            len(products),
            (len(products) + batch_size - 1) // batch_size
        ),
        'success'
    )

def _export_product_batch(self, product_ids):
    """Process a batch of products."""
    for product_id in product_ids:
        # Create binding and export
        pass

Pattern 3: Export with Validation

def export_sale_order(self, record_ids):
    """Export orders with validation."""
    orders = self.env['sale.order'].browse(record_ids)
    validation_errors = []

    for order in orders:
        # Validate before export
        if not order.partner_id:
            validation_errors.append(f'{order.name}: Missing customer')
            continue

        if not order.order_line:
            validation_errors.append(f'{order.name}: No order lines')
            continue

        if order.state not in ['sale', 'done']:
            validation_errors.append(f'{order.name}: Not confirmed')
            continue

        # Export if valid
        # ... create binding and export

    if validation_errors:
        message = '\n'.join(validation_errors[:10])
        return self._build_notification(
            _('Export Validation Errors'),
            message,
            'warning'
        )

Method Naming Convention

The export method name must follow this pattern:

export_{model_name_with_underscores}

# Examples:
export_product_product      # for product.product
export_product_template     # for product.template
export_sale_order           # for sale.order
export_res_partner          # for res.partner
export_stock_picking        # for stock.picking
export_account_move         # for account.move

The wizard automatically converts model names:

Replaces dots (.) with underscores (_)
product.product → calls export_product_product()
sale.order → calls export_sale_order()

Benefits of Shared Export Wizard

Single UI: One wizard works for all models across all connectors
Consistent UX: Users learn once, use everywhere
No Custom Code: No need to create custom wizards or actions
Multi-Backend: Users can export to multiple backends
Extensible: Add new models just by implementing one method
Backend Filtering: Wizard shows only relevant backends via domain

Complete Implementation Checklist

When implementing export for a new model:

Implement export_<model_name>() method in backend model
Handle binding creation (find or create)
Queue export using with_delay()._export_to_external()
Handle errors gracefully with try/except
Return notification with detailed status
Update backend statistics (last_export_date)
Add export action binding (if not exists for this model)
Test from UI (Action → Export to Connectors)
Verify queue jobs are created
Check bindings are created correctly
Review logs for errors

Troubleshooting Export Issues

Issue: "Export not implemented" error

# Solution: Check method name matches pattern
# Model: product.product → Method: export_product_product()
# Model: sale.order → Method: export_sale_order()

Issue: Backend not showing in wizard

# Solution: Check inheritance chain
# Ensure your backend inherits from generic.backend
# which inherits from connector.base.backend

class YourBackend(models.Model):
    _name = 'your.backend'
    _inherits = {'generic.backend': 'generic_backend_id'}

Issue: Export creates duplicates

# Solution: Ensure unique constraint on binding
# In binding model:
_sql_constraints = [
    ('backend_odoo_uniq',
     'unique(backend_id, odoo_id)',
     'A binding already exists for this record on this backend.')
]

Issue: Export completes but nothing happens

# Solution: Check queue_job is running
# 1. Verify queue_job channel exists
# 2. Start queue job worker:
odoo-bin gevent -c odoo.conf --workers=2

# 3. Or run job manually:
>>> job = env['queue.job'].search([...])
>>> job.requeue()

Troubleshooting

When the user reports sync issues or errors:

Step 1: Identify the Problem

Common issues:

Connection/authentication failures → Check references/authentication.md
Import not working → Check component registration
Duplicates being created → Check SQL constraints
Queue jobs not running → Check queue_job configuration
Webhooks not received → Check controller route

Step 2: Use Diagnostic Tools

# Test in Odoo shell
odoo-bin shell -c odoo.conf -d your_db

# Find backend
>>> backend = env['shopify.backend'].browse(1)

# Test connection
>>> backend.action_test_connection()

# Test adapter
>>> with backend.work_on('shopify.product.template') as work:
...     adapter = work.component(usage='backend.adapter')
...     products = adapter.get_products()
...     print(f"Fetched {len(products)} products")

# Test mapper
...     mapper = work.component(usage='import.mapper')
...     if products:
...         mapped = mapper.map_record(products[0])
...         print(mapped.values())

Step 3: Enable Debug Logging

Add to backend or adapter:

import logging
_logger = logging.getLogger(__name__)
_logger.setLevel(logging.DEBUG)

def make_request(self, method, endpoint, **kwargs):
    _logger.debug("API Request: %s %s", method, self.build_url(endpoint))
    _logger.debug("Params: %s", kwargs.get('params'))
    _logger.debug("Data: %s", kwargs.get('data'))

    response = super().make_request(method, endpoint, **kwargs)

    _logger.debug("Response: %s", str(response)[:500])
    return response

Step 4: Check Reference Documentation

Refer the user to:

references/troubleshooting.md - Common issues and solutions
references/architecture.md - Component structure
references/patterns.md - Design patterns
references/api_integration.md - API communication patterns
references/authentication.md - Authentication methods

Available Scripts

init_connector.py

Generate a complete new connector module.

Usage:

python3 scripts/init_connector.py <connector_name> --path <output_path> --type <connector_type>

Arguments:

connector_name: Name (e.g., 'shopify', 'woocommerce')
--path: Output directory (default: current directory)
--type: Connector type - 'ecommerce', 'logistics', 'accounting', 'crm'

Output: Complete module with backend, adapter, binding, views, security

add_binding.py

Add a new entity binding to existing connector.

Usage:

python3 scripts/add_binding.py <connector_path> <entity_name> --odoo-model <model>

Arguments:

connector_path: Path to existing connector module
entity_name: Entity name (e.g., 'order', 'customer')
--odoo-model: Odoo model to bind (e.g., 'sale.order', 'res.partner')

Output: Binding model, views, security rules, adapter methods template

validate_connector.py

Validate connector module structure.

Usage:

python3 scripts/validate_connector.py <connector_path>

Checks:

Required files and directories
Manifest dependencies
Backend model structure
Component registration
Security configuration

Reference Documentation

Load references as needed using the Read tool:

references/architecture.md

Comprehensive guide to generic_connector architecture:

Backend model patterns
Binding model structure
Adapter, mapper, importer, exporter components
Queue job integration
Security model
View patterns

When to read: Creating new connectors, understanding component relationships

references/patterns.md

Design patterns used in connectors:

Template Method, Adapter, Strategy, Factory patterns
Observer pattern for webhooks
Retry and circuit breaker patterns
Rate limiting patterns
Anti-patterns to avoid

When to read: Implementing complex sync logic, handling failures

references/api_integration.md

API integration techniques:

REST, GraphQL, SOAP integrations
Pagination handling (offset, cursor, link header)
Response envelope handling
Webhook integration
Rate limiting implementation
Error handling and retries

When to read: Implementing adapters, handling API specifics

references/authentication.md

Authentication patterns:

API key authentication
OAuth 2.0 (authorization code flow)
Bearer token
Basic auth
HMAC signatures
JWT tokens
Webhook signature verification

When to read: Configuring authentication, debugging 401 errors

references/troubleshooting.md

Common issues and solutions:

Connection issues
Authentication failures
Import/export problems
Queue job issues
Webhook problems
Data mapping errors
Performance optimization
Debugging tips

When to read: Debugging sync issues, performance problems

Best Practices

Always extend generic_connector - Never build from scratch
Use bindings - Never directly modify Odoo records from external data
Queue long operations - Use with_delay() for anything >2 seconds
Implement retry logic - Use binding's retry_count and max_retries
Log extensively - Debug logging helps troubleshoot production issues
Handle API errors - Wrap adapter calls in try/except
Validate data - Check required fields before creating records
Test connection - Always implement _test_connection_implementation()
Use transactions - Leverage Odoo's automatic transaction management
Document the API - Add docstrings to all adapter methods

Component Registration Checklist

When creating components, ensure:

class MyComponent(BaseComponent):
    _name = 'unique.component.name'      # ✓ Unique identifier
    _inherit = 'parent.component'        # ✓ Parent component
    _apply_on = 'model.name'             # ✓ Model this applies to
    _usage = 'component.usage'           # ✓ Usage context

Common usages:

backend.adapter - API communication
record.importer - Single record import
batch.importer - Batch import
record.exporter - Single record export
batch.exporter - Batch export
import.mapper - Import data transformation
export.mapper - Export data transformation

Testing Checklist

Before delivering a connector:

Backend Configuration:

Backend configuration form loads
"Test Connection" button works
Backend inherits from generic.backend correctly
Backend statistics update (last_sync_date, counters)

Import Functionality:

Manual sync imports data
No duplicate records created on import
External IDs are set correctly
Bindings link Odoo records to external records
Import handles API pagination correctly
Import handles API errors gracefully

Export Functionality:

"Export to Connectors" action appears on models (Action menu)
Export wizard shows only relevant backends
export_<model_name>() methods implemented in backend
Export creates bindings if they don't exist
Export queues async jobs via with_delay()
Export notifications show correct counts (exported, created, skipped, errors)
Export respects no_export flag on bindings
Export updates backend statistics (last_export_date)

Queue Jobs:

Queue jobs are registered and visible
Jobs execute successfully in queue_job worker
Failed jobs can be retried
Job logs provide useful debugging info

Scheduled Jobs:

Scheduled cron jobs exist (disabled by default)
Cron jobs can be enabled and run on schedule

Security & Access:

Security access rules allow users to view data
Users can access backend, bindings, and wizards
Proper groups assigned (connector_manager, connector_user)

Integration:

Webhooks received and processed (if applicable)
Webhook signature verification works
Components registered correctly (adapters, mappers, importers, exporters)

Error Handling:

Error handling works (test with invalid credentials)
API errors don't crash Odoo
User-friendly error messages displayed
Detailed errors logged for debugging

Logging & Debugging:

Logging provides useful debug information
Log levels appropriate (INFO for success, ERROR for failures)
Sensitive data (tokens, passwords) not logged

Module Update Process

When updating an existing connector:

# 1. Update module files
# 2. Upgrade module
odoo-bin -c odoo.conf -d your_db -u connector_module_name

# 3. Test thoroughly
# 4. Check logs for errors
tail -f /var/log/odoo/odoo.log

Common Workflows

Workflow: Add Product Sync

Generate binding: python3 scripts/add_binding.py <path> product --odoo-model product.template
Implement adapter methods in models/adapter.py
Create mapper in components/mapper.py
Create importer in components/importer.py
Update backend _sync_products_implementation()
Update module: odoo-bin -u connector_name
Test sync

Workflow: Add Order Import

Generate binding: python3 scripts/add_binding.py <path> order --odoo-model sale.order
Implement adapter methods
Create import mapper (transform external order to Odoo format)
Create importer (handle order lines, customer lookup)
Update backend _sync_orders_implementation()
Configure webhook for real-time import (optional)
Test import

Workflow: Debug Sync Failure

Check logs: tail -f /var/log/odoo/odoo.log
Enable debug logging in adapter
Test in Odoo shell
Check component registration
Verify API credentials
Test adapter methods directly
Check mapper output
Review binding constraints
Refer to references/troubleshooting.md

Output Format

When creating or enhancing connectors:

Use scripts whenever possible (init_connector.py, add_binding.py)
Provide code for custom components (mappers, importers, exporters)
Show configuration (backend fields, view changes)
Include testing steps (how to verify it works)
Reference docs when needed (point to specific reference sections)
Explain patterns used (why this approach was chosen)

Error Prevention

Common mistakes to avoid:

❌ Missing _apply_on in components
❌ Wrong model name in _apply_on
❌ Forgetting to register components in __init__.py
❌ Not setting external_id in mapper
❌ Missing SQL constraint on bindings
❌ Using synchronous operations for long tasks
❌ Not handling API pagination
❌ Hardcoding configuration instead of using backend fields
❌ Not implementing retry logic
❌ Insufficient error handling

Success Criteria

A successfully created/enhanced connector should:

✅ Install without errors
✅ Test connection successfully
✅ Import/export data correctly
✅ Handle API errors gracefully
✅ Log useful information for debugging
✅ Use queue jobs for async operations
✅ Not create duplicate records
✅ Follow generic_connector patterns
✅ Have proper security configuration
✅ Be maintainable and extensible

When to Use Each Reference

Situation	Reference
Creating new connector	architecture.md
Implementing OAuth	authentication.md
Adding webhooks	api_integration.md
Sync not working	troubleshooting.md
Implementing retry logic	patterns.md
Understanding components	architecture.md
API pagination	api_integration.md
401 errors	authentication.md, troubleshooting.md
Performance issues	troubleshooting.md, patterns.md
Best practices	patterns.md (anti-patterns section)

Final Notes

Always test in a development database first
Use the reference connectors (zid, beatroute) as examples
Leverage the scripts to generate boilerplate code
Refer to documentation for specific patterns
Focus on extensibility and maintainability
Follow Odoo and generic_connector conventions

Install Skill

SKILL.md

Odoo Connector Module Creator and Enhancer

Description

Overview

When to Use This Skill

Key Concepts

Generic Connector Framework

Reference Code

Workflow

Creating a New Connector

Enhancing an Existing Connector

Implementing Webhooks

Implementing Export Using Shared Wizard

Architecture Overview

Step-by-Step Implementation

Method Naming Convention

Benefits of Shared Export Wizard

Complete Implementation Checklist

Troubleshooting Export Issues

Troubleshooting

Available Scripts

init_connector.py

add_binding.py

validate_connector.py

Reference Documentation

references/architecture.md

references/patterns.md

references/api_integration.md

references/authentication.md

references/troubleshooting.md

Best Practices

Component Registration Checklist

Testing Checklist

Module Update Process

Common Workflows

Workflow: Add Product Sync

Workflow: Add Order Import

Workflow: Debug Sync Failure

Output Format

Error Prevention

Success Criteria

When to Use Each Reference

Final Notes