Odoo REST API: 実用的な例と統合チュートリアル

Postman の 2025 State of APIs レポートによると、現在、企業の 73% が ERP を少なくとも 3 つの外部システムと統合しています。世界中で 1,200 万人を超えるユーザーをサポートする Odoo は、豊富な API レイヤーを通じてデータ モデル全体を公開しています。しかし、このドキュメントでは、多くの開発者が認証フロー、バッチ操作、運用レベルのエラー処理に苦労していることになります。

このチュートリアルでは、一般的な統合パターンごとに、コピー＆ペースト可能なサンプルを Python と Node.js の両方で提供します。 Shopify の注文を同期する場合でも、モバイル アプリからデータをプッシュする場合でも、カスタム ダッシュボードを構築する場合でも、このガイドはすべてをカバーします。

重要なポイント

• Odoo は 3 つの API アクセス方法を提供します。XML-RPC (レガシー、すべてのバージョン)、JSON-RPC (Web クライアント プロトコル)、および REST API (API キーを備えた Odoo 17 以降) であり、それぞれ認証と使用例が異なります。
• API キー認証 (Odoo 17 以降) は、サーバー間の統合に推奨されるアプローチです。セッション管理も CSRF トークンも必要なく、単純な HTTP ヘッダーも必要ありません。
• 検索ドメイン は、フィルタリングに Odoo の強力なポーランド語表記を使用します。演算子をマスターすれば、あらゆるデータの組み合わせをクエリできます。
• バッチ操作はパフォーマンスにとって重要です。1 回の API 呼び出しで 1,000 件のレコードを作成すると、1,000 回の個別呼び出しよりも 50 倍高速になります。
• エラー処理とレート制限は運用環境の統合に不可欠です。Odoo は構造化されたエラー応答を返します。コードはこれを解析して適切に処理する必要があります。

1. 認証方法

方法 1: API キー (Odoo 17 以降に推奨)

API キーは、サーバー間通信の最も簡単かつ安全な方法です。

# Generate an API key in Odoo:
# Settings → Users → Select user → Account Security → New API Key

# Test authentication
curl -X GET "https://your-odoo.com/api/res.partner" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json"

API キーを使用した Python:

import requests

class OdooAPI:
    def __init__(self, url, api_key):
        self.url = url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
        })

    def get(self, model, params=None):
        response = self.session.get(
            f'{self.url}/api/{model}',
            params=params or {}
        )
        response.raise_for_status()
        return response.json()

    def post(self, model, data):
        response = self.session.post(
            f'{self.url}/api/{model}',
            json=data
        )
        response.raise_for_status()
        return response.json()

# Usage
odoo = OdooAPI('https://your-odoo.com', 'your-api-key-here')
partners = odoo.get('res.partner', {'limit': 10})

API キーを使用した Node.js:

const axios = require('axios');

class OdooAPI {
  constructor(url, apiKey) {
    this.client = axios.create({
      baseURL: `${url}/api`,
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Content-Type': 'application/json',
      },
      timeout: 30000,
    });
  }

  async get(model, params = {}) {
    const { data } = await this.client.get(`/${model}`, { params });
    return data;
  }

  async post(model, body) {
    const { data } = await this.client.post(`/${model}`, body);
    return data;
  }

  async put(model, id, body) {
    const { data } = await this.client.put(`/${model}/${id}`, body);
    return data;
  }

  async delete(model, id) {
    const { data } = await this.client.delete(`/${model}/${id}`);
    return data;
  }
}

// Usage
const odoo = new OdooAPI('https://your-odoo.com', 'your-api-key');
const partners = await odoo.get('res.partner', { limit: 10 });

方法 2: JSON-RPC (すべてのバージョン)

JSON-RPC は、Odoo の Web クライアントによって内部的に使用されるプロトコルです。すべての Odoo バージョンで動作します。

import requests
import json

class OdooJsonRpc:
    def __init__(self, url, db, username, password):
        self.url = url.rstrip('/')
        self.db = db
        self.session = requests.Session()
        self.uid = self._authenticate(username, password)

    def _authenticate(self, username, password):
        response = self._call('/web/session/authenticate', {
            'db': self.db,
            'login': username,
            'password': password,
        })
        if not response.get('uid'):
            raise Exception(f"Authentication failed: {response}")
        return response['uid']

    def _call(self, endpoint, params):
        payload = {
            'jsonrpc': '2.0',
            'method': 'call',
            'params': params,
            'id': None,
        }
        response = self.session.post(
            f'{self.url}{endpoint}',
            json=payload,
            headers={'Content-Type': 'application/json'}
        )
        result = response.json()
        if result.get('error'):
            raise Exception(result['error']['data']['message'])
        return result.get('result')

    def search_read(self, model, domain=None, fields=None, limit=80, offset=0, order=None):
        return self._call('/web/dataset/call_kw', {
            'model': model,
            'method': 'search_read',
            'args': [domain or []],
            'kwargs': {
                'fields': fields or [],
                'limit': limit,
                'offset': offset,
                'order': order or '',
            },
        })

    def create(self, model, values):
        return self._call('/web/dataset/call_kw', {
            'model': model,
            'method': 'create',
            'args': [values],
            'kwargs': {},
        })

    def write(self, model, ids, values):
        return self._call('/web/dataset/call_kw', {
            'model': model,
            'method': 'write',
            'args': [ids, values],
            'kwargs': {},
        })

# Usage
odoo = OdooJsonRpc('https://your-odoo.com', 'mydb', 'admin', 'password')
orders = odoo.search_read('sale.order', [('state', '=', 'sale')],
                          fields=['name', 'amount_total', 'partner_id'],
                          limit=20)

方法 3: XML-RPC (レガシー、ユニバーサル)

import xmlrpc.client

url = 'https://your-odoo.com'
db = 'mydb'
username = 'admin'
password = 'password'

# Authenticate
common = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/common')
uid = common.authenticate(db, username, password, {})

# Create models proxy
models = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')

# Search and read
partners = models.execute_kw(db, uid, password,
    'res.partner', 'search_read',
    [[('is_company', '=', True), ('country_id.code', '=', 'US')]],
    {'fields': ['name', 'email', 'phone'], 'limit': 10}
)

2. CRUD 操作

レコードの作成

# Create a single contact
partner_id = odoo.create('res.partner', {
    'name': 'Acme Corporation',
    'is_company': True,
    'email': '[email protected]',
    'phone': '+1-555-0123',
    'street': '123 Main Street',
    'city': 'San Francisco',
    'state_id': 5,  # California
    'country_id': 233,  # United States
    'category_id': [(6, 0, [1, 3])],  # Tags: replace all with IDs 1 and 3
})

# Create a sale order with lines
order_id = odoo.create('sale.order', {
    'partner_id': partner_id,
    'date_order': '2026-03-23',
    'order_line': [
        (0, 0, {
            'product_id': 42,
            'product_uom_qty': 5,
            'price_unit': 99.99,
        }),
        (0, 0, {
            'product_id': 43,
            'product_uom_qty': 2,
            'price_unit': 149.99,
        }),
    ],
})

Node.js と同等:

// Create a contact
const partnerId = await odoo.post('res.partner', {
  name: 'Acme Corporation',
  is_company: true,
  email: '[email protected]',
  phone: '+1-555-0123',
  street: '123 Main Street',
  city: 'San Francisco',
  country_id: 233,
});

// Create sale order with lines
const orderId = await odoo.post('sale.order', {
  partner_id: partnerId,
  date_order: '2026-03-23',
  order_line: [
    [0, 0, { product_id: 42, product_uom_qty: 5, price_unit: 99.99 }],
    [0, 0, { product_id: 43, product_uom_qty: 2, price_unit: 149.99 }],
  ],
});

レコードの読み取り

# Read specific fields from specific records
data = odoo.search_read('sale.order',
    domain=[('state', '=', 'sale'), ('amount_total', '>', 500)],
    fields=['name', 'partner_id', 'amount_total', 'date_order', 'state'],
    limit=50,
    offset=0,
    order='date_order desc'
)

# Read a single record by ID (REST API)
# GET /api/sale.order/42?fields=name,amount_total

// Node.js — read with pagination
async function fetchAllOrders(odoo) {
  const pageSize = 100;
  let offset = 0;
  let allOrders = [];

  while (true) {
    const orders = await odoo.get('sale.order', {
      domain: JSON.stringify([['state', '=', 'sale']]),
      fields: 'name,partner_id,amount_total',
      limit: pageSize,
      offset,
      order: 'date_order desc',
    });

    allOrders = allOrders.concat(orders);
    if (orders.length < pageSize) break;
    offset += pageSize;
  }

  return allOrders;
}

レコードを更新する

# Update a single record
odoo.write('res.partner', [partner_id], {
    'phone': '+1-555-9999',
    'website': 'https://acme.com',
})

# Update multiple records at once
draft_orders = odoo.search_read('sale.order',
    [('state', '=', 'draft'), ('date_order', '<', '2026-01-01')],
    fields=['id']
)
ids = [o['id'] for o in draft_orders]
odoo.write('sale.order', ids, {'note': 'Reviewed Q1 2026'})

レコードの削除

# Delete records (use with caution)
odoo.write('res.partner', [obsolete_id], {'active': False})  # Prefer archiving

# Actually delete (rarely needed)
# models.execute_kw(db, uid, password, 'res.partner', 'unlink', [[obsolete_id]])

3. 高度な検索ドメイン

Odoo の検索ドメインは、条件を組み合わせるためにポーランド語表記 (接頭辞表記) を使用します。条件間のデフォルトの演算子は AND です。パイプ '|' を使用しますOR の場合はアンパサンド「&」、明示的な AND の場合は「&」。各リーフは、フィールド名、演算子、値のタプルです。 Odoo は、顧客の国ごとに注文をフィルタリングするための「partner_id.country_id.code」など、関連するモデル フィールドをフィルタリングするためのドット表記をサポートしています。

# Complex domain examples

# Orders from US customers with amount > $1000, created this year
domain = [
    ('partner_id.country_id.code', '=', 'US'),
    ('amount_total', '>', 1000),
    ('date_order', '>=', '2026-01-01'),
    ('state', 'in', ['sale', 'done']),
]

# OR condition: email contains 'gmail' OR 'yahoo'
domain = [
    '|',
    ('email', 'ilike', 'gmail.com'),
    ('email', 'ilike', 'yahoo.com'),
]

# Complex: (state=sale AND amount>1000) OR (state=done AND amount>5000)
domain = [
    '|',
    '&', ('state', '=', 'sale'), ('amount_total', '>', 1000),
    '&', ('state', '=', 'done'), ('amount_total', '>', 5000),
]

# Negation: NOT archived
domain = [('active', '!=', False)]

# NULL check: has no email
domain = [('email', '=', False)]

# Hierarchical: all child categories of 'Electronics'
domain = [('categ_id', 'child_of', electronics_id)]

# Date ranges
from datetime import datetime, timedelta
domain = [
    ('create_date', '>=', (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')),
    ('create_date', '<', datetime.now().strftime('%Y-%m-%d')),
]

演算子のリファレンス

4. バッチ操作

バッチ操作はパフォーマンスにとって不可欠です。ループ内で一度に 1 つずつレコードを作成しないでください。

# BAD: 1000 API calls (slow, ~300 seconds)
for customer in customers:
    odoo.create('res.partner', customer)

# GOOD: 1 API call with batch (fast, ~3 seconds)
# Using JSON-RPC batch create
partner_ids = odoo._call('/web/dataset/call_kw', {
    'model': 'res.partner',
    'method': 'create',
    'args': [customers],  # Pass list of dicts
    'kwargs': {},
})

// Node.js batch pattern with chunking
async function batchCreate(odoo, model, records, chunkSize = 200) {
  const results = [];

  for (let i = 0; i < records.length; i += chunkSize) {
    const chunk = records.slice(i, i + chunkSize);
    console.log(`Creating ${model} batch ${i / chunkSize + 1}/${Math.ceil(records.length / chunkSize)}`);

    const ids = await odoo.post(model, chunk);
    results.push(...(Array.isArray(ids) ? ids : [ids]));

    // Respect rate limits
    if (i + chunkSize < records.length) {
      await new Promise(resolve => setTimeout(resolve, 500));
    }
  }

  return results;
}

// Usage
const customers = generateCustomerData(); // Array of 5000 objects
const ids = await batchCreate(odoo, 'res.partner', customers);
console.log(`Created ${ids.length} partners`);

5. エラー処理

本番環境の統合では、エラーを適切に処理する必要があります。

import requests
import logging
import time

logger = logging.getLogger(__name__)

class OdooAPIClient:
    MAX_RETRIES = 3
    RETRY_DELAY = 2  # seconds

    def __init__(self, url, api_key):
        self.url = url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
        })

    def _request(self, method, endpoint, **kwargs):
        last_error = None

        for attempt in range(self.MAX_RETRIES):
            try:
                response = self.session.request(
                    method,
                    f'{self.url}{endpoint}',
                    timeout=30,
                    **kwargs
                )

                if response.status_code == 429:
                    retry_after = int(response.headers.get('Retry-After', 60))
                    logger.warning(f"Rate limited. Retrying after {retry_after}s")
                    time.sleep(retry_after)
                    continue

                if response.status_code == 401:
                    raise AuthenticationError("Invalid API key or session expired")

                if response.status_code == 403:
                    raise PermissionError(f"Access denied: {response.text}")

                if response.status_code == 404:
                    raise RecordNotFoundError(f"Record not found: {endpoint}")

                if response.status_code >= 500:
                    logger.error(f"Server error {response.status_code}: {response.text}")
                    if attempt < self.MAX_RETRIES - 1:
                        time.sleep(self.RETRY_DELAY * (attempt + 1))
                        continue
                    raise ServerError(f"Server error after {self.MAX_RETRIES} retries")

                response.raise_for_status()
                return response.json()

            except requests.exceptions.ConnectionError as e:
                last_error = e
                logger.warning(f"Connection error (attempt {attempt + 1}): {e}")
                if attempt < self.MAX_RETRIES - 1:
                    time.sleep(self.RETRY_DELAY * (attempt + 1))

            except requests.exceptions.Timeout as e:
                last_error = e
                logger.warning(f"Request timeout (attempt {attempt + 1}): {e}")
                if attempt < self.MAX_RETRIES - 1:
                    time.sleep(self.RETRY_DELAY * (attempt + 1))

        raise ConnectionError(f"Failed after {self.MAX_RETRIES} attempts: {last_error}")

class AuthenticationError(Exception): pass
class PermissionError(Exception): pass
class RecordNotFoundError(Exception): pass
class ServerError(Exception): pass

// Node.js error handling with axios interceptors
const axios = require('axios');

function createOdooClient(url, apiKey) {
  const client = axios.create({
    baseURL: `${url}/api`,
    headers: { Authorization: `Bearer ${apiKey}` },
    timeout: 30000,
  });

  // Response interceptor for error handling
  client.interceptors.response.use(
    (response) => response,
    async (error) => {
      const { config, response } = error;
      config.__retryCount = config.__retryCount || 0;

      // Rate limiting
      if (response?.status === 429) {
        const retryAfter = parseInt(response.headers['retry-after'] || '60', 10);
        console.warn(`Rate limited. Waiting ${retryAfter}s...`);
        await new Promise((r) => setTimeout(r, retryAfter * 1000));
        return client(config);
      }

      // Retry on server errors (max 3)
      if (response?.status >= 500 && config.__retryCount < 3) {
        config.__retryCount += 1;
        const delay = config.__retryCount * 2000;
        console.warn(`Server error ${response.status}. Retry ${config.__retryCount}/3 in ${delay}ms`);
        await new Promise((r) => setTimeout(r, delay));
        return client(config);
      }

      // Structured error response
      const errorMessage = response?.data?.error?.message
        || response?.data?.message
        || error.message;

      throw new Error(`Odoo API Error [${response?.status}]: ${errorMessage}`);
    }
  );

  return client;
}

6. 実際の統合例

Shopify と Odoo の注文同期

class ShopifyOdooSync:
    def __init__(self, odoo_client, shopify_client):
        self.odoo = odoo_client
        self.shopify = shopify_client

    def sync_order(self, shopify_order):
        # 1. Find or create customer
        partner = self._get_or_create_partner(shopify_order['customer'])

        # 2. Map products
        order_lines = []
        for item in shopify_order['line_items']:
            product_id = self._find_product_by_sku(item['sku'])
            if not product_id:
                logger.warning(f"Product not found for SKU: {item['sku']}")
                continue
            order_lines.append((0, 0, {
                'product_id': product_id,
                'product_uom_qty': item['quantity'],
                'price_unit': float(item['price']),
                'discount': self._calc_discount(item),
            }))

        # 3. Create sale order
        order_id = self.odoo.create('sale.order', {
            'partner_id': partner['id'],
            'client_order_ref': shopify_order['name'],  # Shopify order #
            'order_line': order_lines,
            'note': f"Shopify Order: {shopify_order['id']}",
        })

        # 4. Auto-confirm if paid
        if shopify_order['financial_status'] == 'paid':
            self.odoo._call('/web/dataset/call_kw', {
                'model': 'sale.order',
                'method': 'action_confirm',
                'args': [[order_id]],
                'kwargs': {},
            })

        return order_id

    def _get_or_create_partner(self, customer):
        # Search by email first
        existing = self.odoo.search_read('res.partner',
            [('email', '=', customer['email'])],
            fields=['id', 'name'], limit=1)

        if existing:
            return existing[0]

        return {'id': self.odoo.create('res.partner', {
            'name': f"{customer['first_name']} {customer['last_name']}",
            'email': customer['email'],
            'phone': customer.get('phone'),
        })}

    def _find_product_by_sku(self, sku):
        products = self.odoo.search_read('product.product',
            [('default_code', '=', sku)],
            fields=['id'], limit=1)
        return products[0]['id'] if products else None

Webhook レシーバー (Flask の例)

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhook/odoo/order', methods=['POST'])
def handle_odoo_webhook():
    # Verify signature
    signature = request.headers.get('X-Odoo-Signature')
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        request.data,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({'error': 'Invalid signature'}), 401

    payload = request.json
    event_type = payload.get('event')

    if event_type == 'sale.order.confirmed':
        handle_order_confirmed(payload['data'])
    elif event_type == 'stock.picking.done':
        handle_shipment_complete(payload['data'])

    return jsonify({'status': 'ok'}), 200

7. パフォーマンスのヒント

よくある質問

Odoo の XML-RPC、JSON-RPC、REST API の違いは何ですか?

XML-RPC は、すべての Odoo バージョンで利用できるレガシー プロトコルです。冗長ですが、広くサポートされています。 JSON-RPC は Odoo の Web クライアントで使用されるプロトコルであり、JSON ペイロードで同じ機能を提供します。 REST API は Odoo 17 で導入され、API キー認証を備えた標準の HTTP エンドポイントを提供するため、最新の統合にとって最も簡単なオプションになります。新しいプロジェクトの場合、Odoo 17 以降を使用している場合は REST API を使用してください。

Odoo API のレート制限はどのように処理すればよいですか?

Odoo.sh は、プラン階層に基づいてレート制限を適用します。 429 応答を受信した場合は、Retry-After ヘッダーを読み取り、再試行する前に待機します。大規模な統合の場合は、指数関数的バックオフを実装し、操作をバッチ処理して API 呼び出しの数を減らし、重要でない同期にはリアルタイムの API 呼び出しの代わりに Odoo のスケジュールされたアクションを一括処理に使用することを検討してください。

API を通じてカスタム Python メソッドを呼び出すことはできますか?

はい。 Odoo モデルのパブリック メソッドは、execute_kw を使用して XML-RPC または JSON-RPC 経由で呼び出すことができます。 REST API の場合は、@http.route を使用してカスタム コントローラー エンドポイントを作成する必要があります。アンダースコアで始まるメソッドはプライベートであり、XML-RPC を通じて外部から呼び出すことはできません。インジェクション攻撃を防ぐために、カスタム メソッドの入力を常に検証してください。

大規模なデータセットを効率的に同期するにはどうすればよいですか?

戦略を組み合わせて使用​​します。バッチ操作とページネーションによる最初の完全同期 (リクエストごとに 200 レコードを制限)、次に write_date フィルタリングを使用した増分同期で、最後の同期以降に変更されたレコードのみを取得します。最終同期タイムスタンプを保存し、ドメイン フィルターとして使用します。 100,000 レコードを超える非常に大規模なデータセットの場合は、API 同期ではなくデータベースの直接レプリケーションを検討してください。

Odoo REST API は Odoo Community Edition で利用できますか?

API キー認証を備えたネイティブ REST API が Odoo 17 Enterprise で導入されました。 Odoo Community の場合、すべてのエディションで利用可能な XML-RPC または JSON-RPC を使用することも、RESTful エンドポイントを追加する OCA の REST フレームワークなどのコミュニティ モジュールをインストールすることもできます。 ECOSIRE の統合サービスは、すべての Odoo エディションと API プロトコルをサポートします。

API 呼び出しで Many2many フィールドと One2many フィールドを処理するにはどうすればよいですか?

リレーショナル フィールドは特別なコマンド タプルを使用します: (0, 0,values) は新しいレコードの作成とリンク、(1, id,values) はリンクされたレコードの更新、(2, id, 0) はリンクされたレコードの削除、(3, id, 0) は削除せずにリンク解除、(4, id, 0) は既存のレコードのリンク、(5, 0, 0) はすべてのリンクを解除、(6, 0, [ids]) すべてのリンクを置き換えます。読み取りの場合、これらのフィールドはデフォルトで ID のリストを返します。完全なデータを取得するには、フィールド名を指定して search_read を使用します。

次のステップ

API 統合は、最新のビジネス システムのバックボーンです。単純なデータ同期を構築する場合でも、複雑なマルチプラットフォーム オーケストレーションを構築する場合でも、このガイドのパターンは役に立ちます。

関連リソース:

• Odoo Python 開発ガイド — Odoo の Python バックエンドの詳細
• Webhook デバッグ ガイド — Webhook 統合のトラブルシューティング
• ECOSIRE Marketplace Connectors — 主要なプラットフォーム用の事前構築された統合

Odoo API の統合についてサポートが必要ですか? ECOSIRE の統合チーム は、Shopify、Amazon、Salesforce、カスタム ERP を含む 50 以上の外部プラットフォームに Odoo を接続しました。単純なデータ同期からリアルタイムの双方向オーケストレーションまで、拡張性のある統合を構築します。 無料の技術相談を予約する。