Use the real Odoo 19 external API (JSON-2), and batch calls

The previous commit moved to /jsonrpc, which the v19 docs list as deprecated
alongside /xmlrpc — both go away in Odoo 22. The actual v19 external API is
JSON-2: POST /json/2/<model>/<method>, API key as a bearer token.

Three consequences, all of which change calling code:
  - no uid and no password; UID now comes from res.users/context_get
  - every argument is named, positional args do not exist, so x() becomes
    x(model, method, ids=None, **kwargs) and callers pass domain=/vals=/fields=
  - one call is one transaction and nothing chains, which is the mechanical
    reason state changes must go through a button method

Batching: no multi-call envelope exists, so batching means widening a call, not
bundling calls. Added get_or_create_many() (one search_read + one create for a
whole set) and put build_demo.py's product, customer, vehicle, invoice and
headroom loops through set-based calls.

test_odoo_connect.py checks the call shape and the batching offline, with a
faked urlopen — none of this can be exercised without a live instance otherwise.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
glenn schrooyen
2026-08-09 02:53:03 +02:00
co-authored by Claude Opus 5
parent 3fd2c51a3d
commit 93edf9296e
6 changed files with 295 additions and 129 deletions
+49 -55
View File
@@ -9,7 +9,7 @@ Idempotent: get-or-create by name. Run order matters (currency before invoices).
Requires odoo_connect.py in the same folder + creds (env or creds.json).
"""
import json, os
from odoo_connect import x, get_or_create, URL, DB
from odoo_connect import x, get_or_create, get_or_create_many, OdooError, URL, DB
CFG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "demo_config.json")
cfg = json.load(open(CFG, encoding="utf-8"))
@@ -35,91 +35,85 @@ if unknown:
# 1. currency (must run before invoices)
cur = cfg.get("currency")
if cur:
ids = x("res.currency", "search", [["name", "=", cur]], context={"active_test": False})
ids = x("res.currency", "search", domain=[["name", "=", cur]], context={"active_test": False})
if not ids:
ids = [x("res.currency", "create", {"name": cur, "rounding": 0.01})]
x("res.currency", "write", ids, {"active": True})
comp = x("res.company", "search", [], limit=1)[0]
ids = [x("res.currency", "create", vals_list={"name": cur, "rounding": 0.01})]
x("res.currency", "write", ids, vals={"active": True})
comp = x("res.company", "search", domain=[], limit=1)[0]
try:
x("res.company", "write", [comp], {"currency_id": ids[0]})
x("res.company", "write", comp, vals={"currency_id": ids[0]})
except Exception as e:
print(" ! currency switch blocked (invoices exist?):", str(e)[:120])
print("currency:", cur)
# 2. settings
s = x("res.config.settings", "create", {
s = x("res.config.settings", "create", vals_list={
"account_use_credit_limit": bool(cfg.get("use_credit_limit", True)),
"group_product_pricelist": True,
})
x("res.config.settings", "execute", [s])
x("res.config.settings", "execute", s)
# 3. products
prod = {}
for p in cfg.get("products", []):
vals = {"name": p["name"], "list_price": p.get("price", 0.0),
"weight": p.get("weight", 0.0), "type": "consu", "is_storable": True}
try:
prod[p["name"]] = get_or_create("product.template", [["name", "=", p["name"]]], vals)
except Exception:
vals.pop("is_storable", None)
prod[p["name"]] = get_or_create("product.template", [["name", "=", p["name"]]], vals)
# 3. products (batched: one search_read + one create for the whole catalogue)
rows = [{"name": p["name"], "list_price": p.get("price", 0.0), "weight": p.get("weight", 0.0),
"type": "consu", "is_storable": True} for p in cfg.get("products", [])]
try:
prod = get_or_create_many("product.template", "name", rows)
except OdooError:
for r in rows: # older builds have no is_storable — retry the batch without it
r.pop("is_storable", None)
prod = get_or_create_many("product.template", "name", rows)
print("products:", len(prod))
# 4. vehicle categories + vehicles
brand = None
if cfg.get("vehicles"):
# 4. vehicle categories + vehicles — batched one level at a time (each level needs the ids above it)
veh = cfg.get("vehicles", [])
if veh:
brand = get_or_create("fleet.vehicle.model.brand", [["name", "=", "Demo Fleet"]], {"name": "Demo Fleet"})
for v in cfg.get("vehicles", []):
cat = get_or_create("fleet.vehicle.model.category", [["name", "=", v["category"]]],
{"name": v["category"], "weight_capacity": v.get("capacity_kg", 0.0)})
model = get_or_create("fleet.vehicle.model", [["name", "=", v["model"]]],
{"name": v["model"], "brand_id": brand, "category_id": cat})
get_or_create("fleet.vehicle", [["license_plate", "=", v["plate"]]],
{"model_id": model, "license_plate": v["plate"], "category_id": cat})
print("vehicles:", len(cfg.get("vehicles", [])))
cats = get_or_create_many("fleet.vehicle.model.category", "name",
[{"name": v["category"], "weight_capacity": v.get("capacity_kg", 0.0)} for v in veh])
models = get_or_create_many("fleet.vehicle.model", "name",
[{"name": v["model"], "brand_id": brand, "category_id": cats[v["category"]]} for v in veh])
get_or_create_many("fleet.vehicle", "license_plate",
[{"license_plate": v["plate"], "model_id": models[v["model"]],
"category_id": cats[v["category"]]} for v in veh])
print("vehicles:", len(veh))
# 5. customers with credit limits
cust = {}
for c in cfg.get("customers", []):
cust[c["name"]] = get_or_create("res.partner", [["name", "=", c["name"]]],
{"name": c["name"], "is_company": True, "customer_rank": 1,
"use_partner_credit_limit": True, "credit_limit": c.get("credit_limit", 0.0)})
# 5. customers with credit limits (batched)
cust = get_or_create_many("res.partner", "name",
[{"name": c["name"], "is_company": True, "customer_rank": 1, "use_partner_credit_limit": True,
"credit_limit": c.get("credit_limit", 0.0)} for c in cfg.get("customers", [])])
print("customers:", len(cust))
# 6. B2B pricelist (optional global % discount)
pl_cfg = cfg.get("pricelist")
if pl_cfg:
pl = get_or_create("product.pricelist", [["name", "=", pl_cfg["name"]]], {"name": pl_cfg["name"]})
if pl_cfg.get("percent") and not x("product.pricelist.item", "search", [["pricelist_id", "=", pl]]):
x("product.pricelist.item", "create", {"pricelist_id": pl, "applied_on": "3_global",
if pl_cfg.get("percent") and not x("product.pricelist.item", "search", domain=[["pricelist_id", "=", pl]]):
x("product.pricelist.item", "create", vals_list={"pricelist_id": pl, "applied_on": "3_global",
"compute_price": "percentage", "percent_price": pl_cfg["percent"]})
for cname in pl_cfg.get("assign_to", []):
if cname in cust:
x("res.partner", "write", [cust[cname]], {"property_product_pricelist": pl})
assign = [cust[c] for c in pl_cfg.get("assign_to", []) if c in cust]
if assign: # same vals for all of them -> one write
x("res.partner", "write", assign, vals={"property_product_pricelist": pl})
# 7. pre-stage receivables (post unpaid invoices)
def first_variant(tmpl_id):
return x("product.product", "search", [["product_tmpl_id", "=", tmpl_id]], limit=1)[0]
return x("product.product", "search", domain=[["product_tmpl_id", "=", tmpl_id]], limit=1)[0]
any_prod = first_variant(next(iter(prod.values()))) if prod else None
for stage in cfg.get("prestage_receivables", []):
cid = cust.get(stage["customer"])
if not cid:
continue
mv = x("account.move", "create", {
"move_type": "out_invoice", "partner_id": cid,
stages = [st for st in cfg.get("prestage_receivables", []) if cust.get(st["customer"])]
if stages:
mvs = x("account.move", "create", vals_list=[{
"move_type": "out_invoice", "partner_id": cust[st["customer"]],
"invoice_line_ids": [(0, 0, {"product_id": any_prod, "quantity": 1,
"price_unit": stage["amount"], "tax_ids": [(6, 0, [])]})]})
"price_unit": st["amount"], "tax_ids": [(6, 0, [])]})]}
for st in stages])
try:
x("account.move", "action_post", [mv])
except Exception as e:
print(" ! post for", stage["customer"], str(e)[:100])
x("account.move", "action_post", mvs) # one button call over every invoice
except OdooError as e:
print(" ! posting invoices:", str(e)[:160])
# 8. verify credit headroom
print("\n-- credit headroom --")
for c in cfg.get("customers", []):
cid = cust[c["name"]]
r = x("res.partner", "read", [cid], ["credit", "credit_limit"])[0]
for r in x("res.partner", "read", list(cust.values()), fields=["name", "credit", "credit_limit"]):
flag = "OVER" if r["credit"] > r["credit_limit"] else "ok"
print(f" {c['name']:30} owes={r['credit']:>12,.0f} limit={r['credit_limit']:>12,.0f} {flag}")
print(f" {r['name']:30} owes={r['credit']:>12,.0f} limit={r['credit_limit']:>12,.0f} {flag}")
print("\nDONE. Instance:", URL, "DB:", DB)