Calling External APIs from Odoo: Integration Patterns Guide

Why Call External APIs from Odoo?

Odoo often needs to communicate with external services: payment gateways, shipping carriers, CRM platforms, email services, SMS providers, and custom microservices. This guide covers robust patterns for making HTTP requests from Odoo server-side code.

Basic HTTP Requests

Use Python's requests library (included in Odoo dependencies):

import requests
import json
from odoo import models, fields, api
from odoo.exceptions import UserError

class ShippingService(models.Model):
    _name = 'shipping.service'
    _description = 'Shipping Integration'

    def _call_shipping_api(self, endpoint, payload):
        api_key = self.env['ir.config_parameter'].sudo().get_param('shipping.api_key')
        url = f'https://api.shipping-provider.com/v2/{endpoint}'

        try:
            response = requests.post(
                url,
                json=payload,
                headers={
                    'Authorization': f'Bearer {api_key}',
                    'Content-Type': 'application/json',
                },
                timeout=30,  # ALWAYS set a timeout
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            raise UserError('Shipping API timed out. Please try again.')
        except requests.exceptions.HTTPError as e:
            raise UserError(f'Shipping API error: {e.response.status_code} - {e.response.text}')
        except requests.exceptions.ConnectionError:
            raise UserError('Cannot connect to shipping API. Check your internet connection.')

Critical Rule: Always Set Timeout

Never make HTTP requests without a timeout. Without it, a hanging external API blocks an Odoo worker indefinitely:

# DANGEROUS: no timeout — can block the worker forever
response = requests.get('https://api.example.com/data')

# SAFE: 30-second timeout
response = requests.get('https://api.example.com/data', timeout=30)

# BETTER: separate connect and read timeouts
response = requests.get('https://api.example.com/data', timeout=(5, 30))
# 5 seconds to connect, 30 seconds to read

Authentication Patterns

API Key in Header

headers = {'X-API-Key': api_key}
response = requests.get(url, headers=headers, timeout=30)

Bearer Token

headers = {'Authorization': f'Bearer {access_token}'}
response = requests.get(url, headers=headers, timeout=30)

Basic Auth

response = requests.get(url, auth=(username, password), timeout=30)

OAuth2 Client Credentials

def _get_oauth_token(self):
    response = requests.post(
        'https://auth.provider.com/oauth/token',
        data={
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret,
        },
        timeout=10
    )
    response.raise_for_status()
    return response.json()['access_token']

Retry Logic

External APIs can have transient failures. Implement retry with exponential backoff:

import time

def _api_call_with_retry(self, method, url, max_retries=3, **kwargs):
    kwargs.setdefault('timeout', 30)
    last_error = None

    for attempt in range(max_retries):
        try:
            response = getattr(requests, method)(url, **kwargs)
            response.raise_for_status()
            return response.json()
        except (requests.exceptions.Timeout,
                requests.exceptions.ConnectionError) as e:
            last_error = e
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1s, 2s, 4s
                time.sleep(wait)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code in (429, 500, 502, 503, 504):
                last_error = e
                if attempt < max_retries - 1:
                    wait = 2 ** attempt
                    time.sleep(wait)
            else:
                raise  # non-retryable HTTP error

    raise UserError(f'API call failed after {max_retries} attempts: {last_error}')

Webhook Receiver

Receive callbacks from external services:

from odoo import http
from odoo.http import request
import hmac
import hashlib

class WebhookController(http.Controller):

    @http.route('/webhook/payment', type='json', auth='none', csrf=False, methods=['POST'])
    def payment_webhook(self, **kwargs):
        # Verify webhook signature
        secret = request.env['ir.config_parameter'].sudo().get_param('payment.webhook_secret')
        signature = request.httprequest.headers.get('X-Signature')
        body = request.httprequest.get_data()

        expected = hmac.new(
            secret.encode(), body, hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(signature or '', expected):
            return {'status': 'error', 'message': 'Invalid signature'}

        # Process the webhook
        data = request.jsonrequest
        payment_ref = data.get('reference')
        status = data.get('status')

        order = request.env['sale.order'].sudo().search([
            ('name', '=', payment_ref)
        ], limit=1)

        if order and status == 'completed':
            order.action_confirm()

        return {'status': 'ok'}

Async Processing with Cron

For non-urgent API calls, queue them and process asynchronously:

class ApiQueue(models.Model):
    _name = 'api.queue'
    _description = 'API Call Queue'

    endpoint = fields.Char(required=True)
    payload = fields.Text(required=True)
    state = fields.Selection([
        ('pending', 'Pending'),
        ('done', 'Done'),
        ('error', 'Error'),
    ], default='pending')
    error_message = fields.Text()
    retry_count = fields.Integer(default=0)

    def _cron_process_queue(self):
        pending = self.search([
            ('state', '=', 'pending'),
            ('retry_count', '<', 3),
        ], limit=50)

        for item in pending:
            try:
                result = self._call_api(item.endpoint, json.loads(item.payload))
                item.write({'state': 'done'})
                self.env.cr.commit()
            except Exception as e:
                item.write({
                    'state': 'error' if item.retry_count >= 2 else 'pending',
                    'error_message': str(e),
                    'retry_count': item.retry_count + 1,
                })
                self.env.cr.commit()

Storing API Credentials Securely

# Use ir.config_parameter (encrypted at rest in Odoo.sh)
self.env['ir.config_parameter'].sudo().set_param('myservice.api_key', 'sk_live_xxx')

# Or use a dedicated settings model
class ResConfigSettings(models.TransientModel):
    _inherit = 'res.config.settings'

    myservice_api_key = fields.Char(
        config_parameter='myservice.api_key',
        string='API Key',
    )

Logging API Calls

import logging
_logger = logging.getLogger(__name__)

def _call_api(self, endpoint, payload):
    _logger.info('API call: %s %s', endpoint, json.dumps(payload)[:200])
    try:
        response = requests.post(url, json=payload, timeout=30)
        _logger.info('API response: %s %s', response.status_code, response.text[:200])
        response.raise_for_status()
        return response.json()
    except Exception:
        _logger.exception('API call failed: %s', endpoint)
        raise

Best Practices

• ALWAYS set a timeout on HTTP requests (30 seconds is a good default)
• Store API credentials in ir.config_parameter, never hardcoded
• Implement retry logic for transient failures (timeouts, 5xx errors)
• Verify webhook signatures to prevent spoofing
• Use async processing (cron queue) for non-critical API calls
• Log requests and responses for debugging (truncate large payloads)
• Handle all exception types: Timeout, ConnectionError, HTTPError
• Use meaningful error messages that help users understand what went wrong