Integrating Odoo with External Systems via REST APIs

Odoo is powerful, but modern businesses need integrations. Here's our approach to connecting Odoo with payment processors, shipping providers, CRM systems, and custom applications using REST APIs and webhooks.

The Integration Challenge

Out of the box, Odoo provides XML-RPC and JSON-RPC APIs. While functional, they have limitations:

• Not RESTful: External developers expect REST conventions (GET, POST, PUT, DELETE)
• Authentication complexity: XML-RPC sessions are cumbersome for modern apps
• No webhooks: Can't push real-time events to external systems
• Limited filtering: Domain filters are Odoo-specific syntax

Our REST API Layer

Architecture Overview

External System → API Gateway → REST Controller → Odoo Models → PostgreSQL
                     ↑                                    ↓
                     └──────────── Webhooks ──────────────┘

1. RESTful Controllers

We built custom Odoo controllers that expose RESTful endpoints:

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

class RentalAPI(http.Controller):

    @http.route('/api/v1/rentals', type='http', auth='api_key',
                methods=['GET'], csrf=False)
    def get_rentals(self, **params):
        """Get rental reservations with filtering"""

        # Parse query parameters
        customer_id = params.get('customer_id')
        status = params.get('status')
        start_date = params.get('start_date')
        limit = int(params.get('limit', 100))
        offset = int(params.get('offset', 0))

        # Build domain
        domain = []
        if customer_id:
            domain.append(('customer_id', '=', int(customer_id)))
        if status:
            domain.append(('state', '=', status))
        if start_date:
            domain.append(('start_date', '>=', start_date))

        # Query Odoo
        Rental = request.env['rental.reservation'].sudo()
        rentals = Rental.search(domain, limit=limit, offset=offset)
        total = Rental.search_count(domain)

        # Serialize response
        return request.make_response(
            json.dumps({
                'data': [r._to_json() for r in rentals],
                'total': total,
                'limit': limit,
                'offset': offset
            }),
            headers={'Content-Type': 'application/json'}
        )

    @http.route('/api/v1/rentals', type='json', auth='api_key',
                methods=['POST'], csrf=False)
    def create_rental(self, **data):
        """Create new rental reservation"""

        try:
            Rental = request.env['rental.reservation'].sudo()
            rental = Rental.create({
                'customer_id': data['customer_id'],
                'product_id': data['product_id'],
                'start_date': data['start_date'],
                'end_date': data['end_date'],
                'pickup_location': data.get('pickup_location'),
                'return_location': data.get('return_location')
            })

            return {
                'success': True,
                'rental_id': rental.id,
                'rental': rental._to_json()
            }

        except Exception as e:
            return {
                'success': False,
                'error': str(e)
            }

2. API Key Authentication

We implemented token-based authentication instead of username/password:

class APIAuth(http.Controller):

    def _authenticate_api_key(self, api_key):
        """Validate API key and return user"""
        APIKey = request.env['api.key'].sudo()
        key_record = APIKey.search([
            ('key', '=', api_key),
            ('active', '=', True),
            ('expiry_date', '>=', fields.Date.today())
        ], limit=1)

        if not key_record:
            raise werkzeug.exceptions.Unauthorized('Invalid API key')

        # Update last used
        key_record.last_used = fields.Datetime.now()

        return key_record.user_id

# Custom auth method
def api_key_auth(self):
    api_key = request.httprequest.headers.get('X-API-Key')
    if not api_key:
        raise werkzeug.exceptions.Unauthorized('API key required')

    user = self._authenticate_api_key(api_key)
    request.uid = user.id

3. Webhooks for Real-Time Events

When events occur in Odoo, we push notifications to external systems:

class RentalReservation(models.Model):
    _inherit = 'rental.reservation'

    @api.model
    def create(self, vals):
        """Trigger webhook on new reservation"""
        rental = super().create(vals)

        # Send webhook
        self._trigger_webhook('rental.created', {
            'rental_id': rental.id,
            'customer_id': rental.customer_id.id,
            'product_id': rental.product_id.id,
            'start_date': rental.start_date.isoformat(),
            'end_date': rental.end_date.isoformat(),
            'total_amount': rental.total_amount
        })

        return rental

    def write(self, vals):
        """Trigger webhook on status change"""
        old_state = self.state
        result = super().write(vals)

        if 'state' in vals and vals['state'] != old_state:
            self._trigger_webhook('rental.status_changed', {
                'rental_id': self.id,
                'old_status': old_state,
                'new_status': self.state,
                'timestamp': fields.Datetime.now().isoformat()
            })

        return result

    def _trigger_webhook(self, event_type, data):
        """Send webhook to registered endpoints"""
        Webhook = self.env['webhook.subscription'].sudo()
        subscriptions = Webhook.search([
            ('event_type', '=', event_type),
            ('active', '=', True)
        ])

        for subscription in subscriptions:
            # Async webhook delivery
            self.env['webhook.delivery'].create({
                'subscription_id': subscription.id,
                'event_type': event_type,
                'payload': json.dumps(data),
                'target_url': subscription.url
            }).send_async()

Real-World Integrations

Payment Processing (Stripe)

Synchronize rental invoices with Stripe for payment collection:

class AccountMove(models.Model):
    _inherit = 'account.move'

    stripe_invoice_id = fields.Char('Stripe Invoice ID')
    stripe_payment_intent = fields.Char('Stripe Payment Intent')

    def action_post(self):
        """Create Stripe invoice when Odoo invoice is posted"""
        result = super().action_post()

        for invoice in self:
            if invoice.move_type == 'out_invoice' and not invoice.stripe_invoice_id:
                # Create Stripe invoice
                stripe_invoice = stripe.Invoice.create(
                    customer=invoice.partner_id.stripe_customer_id,
                    description=f'Rental Invoice {invoice.name}',
                    metadata={'odoo_invoice_id': invoice.id}
                )

                # Add line items
                for line in invoice.invoice_line_ids:
                    stripe.InvoiceItem.create(
                        invoice=stripe_invoice.id,
                        customer=invoice.partner_id.stripe_customer_id,
                        amount=int(line.price_total * 100),  # cents
                        currency='usd',
                        description=line.name
                    )

                # Finalize and send
                stripe.Invoice.finalize_invoice(stripe_invoice.id)
                invoice.stripe_invoice_id = stripe_invoice.id

        return result

Shipping Integration (ShipStation)

Auto-create shipping labels for rental equipment delivery:

class StockPicking(models.Model):
    _inherit = 'stock.picking'

    shipstation_order_id = fields.Char('ShipStation Order ID')
    tracking_number = fields.Char('Tracking Number')

    def action_confirm(self):
        """Create ShipStation order when delivery is confirmed"""
        result = super().action_confirm()

        for picking in self:
            if picking.picking_type_code == 'outgoing' and not picking.shipstation_order_id:
                # Create ShipStation order
                response = requests.post(
                    'https://ssapi.shipstation.com/orders/createorder',
                    auth=(API_KEY, API_SECRET),
                    json={
                        'orderNumber': picking.name,
                        'orderDate': picking.scheduled_date.isoformat(),
                        'orderStatus': 'awaiting_shipment',
                        'customerEmail': picking.partner_id.email,
                        'shipTo': {
                            'name': picking.partner_id.name,
                            'street1': picking.partner_id.street,
                            'city': picking.partner_id.city,
                            'state': picking.partner_id.state_id.code,
                            'postalCode': picking.partner_id.zip,
                            'country': picking.partner_id.country_id.code
                        },
                        'items': [{
                            'sku': move.product_id.default_code,
                            'name': move.product_id.name,
                            'quantity': move.product_uom_qty
                        } for move in picking.move_ids_without_package]
                    }
                )

                if response.status_code == 200:
                    picking.shipstation_order_id = response.json()['orderId']

        return result

CRM Integration (HubSpot)

Sync customer data and rental activity to HubSpot:

class ResPartner(models.Model):
    _inherit = 'res.partner'

    hubspot_contact_id = fields.Char('HubSpot Contact ID')

    @api.model
    def create(self, vals):
        """Create HubSpot contact when customer is created"""
        partner = super().create(vals)

        if partner.customer_rank > 0:
            # Create HubSpot contact
            response = requests.post(
                'https://api.hubapi.com/contacts/v1/contact',
                headers={'Authorization': f'Bearer {HUBSPOT_API_KEY}'},
                json={
                    'properties': [
                        {'property': 'email', 'value': partner.email},
                        {'property': 'firstname', 'value': partner.name.split()[0]},
                        {'property': 'lastname', 'value': ' '.join(partner.name.split()[1:])},
                        {'property': 'phone', 'value': partner.phone},
                        {'property': 'odoo_customer_id', 'value': str(partner.id)}
                    ]
                }
            )

            if response.status_code == 200:
                partner.hubspot_contact_id = response.json()['vid']

        return partner

Performance Considerations

Rate Limiting

from odoo.addons.web.controllers.main import ratelimit

@http.route('/api/v1/rentals', auth='api_key')
@ratelimit(limit=1000, interval=3600)  # 1000 req/hour
def get_rentals(self, **params):
    # ... implementation

Caching

from werkzeug.contrib.cache import RedisCache
cache = RedisCache(host='localhost', port=6379)

def get_rentals(self, **params):
    cache_key = f"rentals:{customer_id}:{status}"
    cached = cache.get(cache_key)

    if cached:
        return cached

    # Query database
    result = # ...
    cache.set(cache_key, result, timeout=300)  # 5 min
    return result

API Documentation

We auto-generate OpenAPI (Swagger) documentation:

openapi: 3.0.0
info:
  title: CleverBusiness Rental API
  version: 1.0.0
paths:
  /api/v1/rentals:
    get:
      summary: List rental reservations
      parameters:
        - name: customer_id
          in: query
          schema:
            type: integer
        - name: status
          in: query
          schema:
            type: string
            enum: [draft, confirmed, picked_up, returned]
      security:
        - ApiKeyAuth: []
      responses:
        200:
          description: List of rentals
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/Rental'

Monitoring & Logging

Track API usage and errors with custom logging:

class APILog(models.Model):
    _name = 'api.log'

    timestamp = fields.Datetime(default=fields.Datetime.now)
    endpoint = fields.Char()
    method = fields.Char()
    api_key_id = fields.Many2one('api.key')
    status_code = fields.Integer()
    response_time = fields.Float()  # milliseconds
    error_message = fields.Text()

# Middleware logging
def log_api_request(endpoint, method, start_time, status, error=None):
    request.env['api.log'].sudo().create({
        'endpoint': endpoint,
        'method': method,
        'api_key_id': request.api_key_id,
        'status_code': status,
        'response_time': (time.time() - start_time) * 1000,
        'error_message': error
    })

Results

Since implementing our REST API layer:

• 12 external integrations connected (Stripe, ShipStation, HubSpot, Zapier, custom apps)
• 450K+ API calls/month processed
• avg 85ms response time
• 99.97% uptime for API endpoints
• Zero manual data entry between systems

Best Practices

1. Version your APIs: Use /api/v1/ prefix to allow backwards-compatible changes.

2. Use sudo() carefully: Always validate permissions even with API keys.

3. Async webhook delivery: Don't block requests waiting for external systems.

4. Comprehensive logging: Track every API call for debugging and analytics.

Need API integrations for your Odoo instance? Contact us to discuss your requirements.