glenn schrooyenandClaude Opus 5
3fd2c51a3d
JSON-RPC is the recommended Odoo API from 19 on. odoo_connect.py now posts to /jsonrpc over stdlib urllib instead of xmlrpc.client — same x()/get_or_create()/ fields_named() surface, common.version() becomes version(), server errors raise OdooError with Odoo's message, 600s timeout so module installs survive. Also: state changes must go through the model's own button method, never a write on state. Writing the field skips the delivery order, the journal entries, the sequence and the validations a hero feature exists to demonstrate. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Generic Odoo JSON-RPC connector for odoo-demo-architect.
|
|
|
|
JSON-RPC (`POST /jsonrpc`) is the recommended API from Odoo 19 on; XML-RPC still
|
|
works but is legacy. Same call surface either way — only the transport changed.
|
|
|
|
Reads credentials from environment variables (or a creds.json next to this file):
|
|
ODOO_URL e.g. https://my-instance.odoo.com (no trailing slash)
|
|
ODOO_DB database name
|
|
ODOO_LOGIN user login (often an email; "admin" on Odoo.sh trials)
|
|
ODOO_KEY an API key (Settings > My Profile > Account Security > New API Key)
|
|
|
|
Usage:
|
|
from odoo_connect import x, version, UID, URL, DB
|
|
UID # authenticated user id
|
|
version() # confirm server_version BEFORE building (field names vary by version)
|
|
x("res.partner", "search_read", [["customer_rank", ">", 0]], fields=["name"])
|
|
|
|
The x() helper passes keyword args straight into execute_kw's kwargs, so pass
|
|
Odoo kwargs natively: x(model, "search", domain, limit=5, context={"active_test": False})
|
|
Do NOT pass a context dict positionally — that becomes `offset` and Postgres errors.
|
|
|
|
Move records through their BUTTONS, not their fields: x("sale.order",
|
|
"action_confirm", [ids]) — never write state="sale". See SKILL.md.
|
|
"""
|
|
import os, json, itertools, urllib.request
|
|
|
|
_here = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
def _cfg(key, default=None):
|
|
v = os.environ.get(key)
|
|
if v:
|
|
return v
|
|
path = os.path.join(_here, "creds.json")
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
return json.load(f).get(key, default)
|
|
return default
|
|
|
|
URL = (_cfg("ODOO_URL") or "").rstrip("/")
|
|
DB = _cfg("ODOO_DB")
|
|
LOGIN = _cfg("ODOO_LOGIN") or "admin"
|
|
KEY = _cfg("ODOO_KEY")
|
|
|
|
if not (URL and DB and KEY):
|
|
raise SystemExit("Set ODOO_URL, ODOO_DB, ODOO_KEY (and optionally ODOO_LOGIN) "
|
|
"as env vars or in creds.json next to this script.")
|
|
|
|
_ids = itertools.count(1)
|
|
|
|
|
|
class OdooError(RuntimeError):
|
|
"""Server-side error, carrying Odoo's own message + traceback."""
|
|
|
|
|
|
def _rpc(service, method, args, timeout=600):
|
|
"""One JSON-RPC call. Module installs take minutes — hence the long default timeout."""
|
|
payload = json.dumps({"jsonrpc": "2.0", "method": "call", "id": next(_ids),
|
|
"params": {"service": service, "method": method, "args": args}})
|
|
req = urllib.request.Request(f"{URL}/jsonrpc", data=payload.encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
res = json.load(r)
|
|
if "error" in res:
|
|
e = res["error"]
|
|
d = e.get("data") or {}
|
|
raise OdooError(d.get("message") or e.get("message") or json.dumps(e)[:500])
|
|
return res["result"]
|
|
|
|
|
|
def version():
|
|
"""Server info dict — check ["server_version"] before building anything."""
|
|
return _rpc("common", "version", [])
|
|
|
|
|
|
UID = _rpc("common", "authenticate", [DB, LOGIN, KEY, {}])
|
|
if not UID:
|
|
raise SystemExit("Authentication failed — check DB / LOGIN / API key.")
|
|
|
|
|
|
def x(model, method, *args, **kwargs):
|
|
"""execute_kw wrapper. Positional args -> Odoo positional params;
|
|
keyword args (limit, fields, context, ...) -> Odoo kwargs."""
|
|
return _rpc("object", "execute_kw", [DB, UID, KEY, model, method, list(args), kwargs])
|
|
|
|
|
|
def get_or_create(model, domain, vals, label=None):
|
|
"""Idempotent create: update if a record matches `domain`, else create."""
|
|
ids = x(model, "search", domain, limit=1)
|
|
if ids:
|
|
x(model, "write", ids, vals)
|
|
return ids[0]
|
|
return x(model, "create", vals)
|
|
|
|
|
|
def fields_named(model, *needles):
|
|
"""Return field technical names on `model` whose name/label contains any needle.
|
|
NOTE: on some Odoo 19 builds fields_get attribute VALUES come back null, so match
|
|
on the technical KEYS, which are always reliable."""
|
|
keys = list(x(model, "fields_get", [], {}).keys())
|
|
nl = [n.lower() for n in needles]
|
|
return sorted(k for k in keys if any(n in k.lower() for n in nl))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("server_version:", version().get("server_version"))
|
|
print("uid:", UID, "url:", URL, "db:", DB)
|