callista-odoo-demo/scripts/odoo_connect.py

"""Generic Odoo JSON-2 connector for callista-odoo-demo.

Odoo 19's external API is **JSON-2**: `POST /json/2/<model>/<method>`, authenticated
with an API key as a bearer token. The old `/xmlrpc`, `/xmlrpc/2` and `/jsonrpc`
endpoints (services common / db / object) are all deprecated and scheduled for removal
in Odoo 22. https://www.odoo.com/documentation/19.0/developer/reference/external_api.html

Three things differ from execute_kw, and they bite:
  1. No uid, no password. The API key alone identifies the user.
  2. **Every argument is named** — there are no positional args. Record ids go in `ids`,
     everything else under the ORM method's own parameter name (`domain`, `fields`,
     `vals`, ...). Your database's exact signatures are at <ODOO_URL>/doc.
  3. Each call is its own SQL transaction. Multi-step work must go through ONE method
     that does it all (`action_confirm`, `search_read`) — see SKILL.md.

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 (only needed when one domain serves several databases)
  ODOO_KEY    an API key (Settings > My Profile > Account Security > New API Key)

Usage:
    from odoo_connect import x, version, UID, URL, DB
    version()["version"]                      # confirm BEFORE building — fields vary by version
    x("res.partner", "search", domain=[["customer_rank", ">", 0]])
    x("res.partner", "read", ids, fields=["name"])
    x("sale.order", "action_confirm", ids)    # buttons, never write(state=...)

Note: keys expire (three months maximum, one day recommended for interactive use), and
Odoo Online restricts the external API to Custom plans — not One App Free or Standard.
"""
import os, json, urllib.request, urllib.error

_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")
KEY =  _cfg("ODOO_KEY")

if not (URL and KEY):
    raise SystemExit("Set ODOO_URL and ODOO_KEY (and ODOO_DB if the domain serves several "
                     "databases) as env vars or in creds.json next to this script.")

_HEADERS = {"Authorization": f"bearer {KEY}",
            "Content-Type": "application/json; charset=utf-8",
            "User-Agent": "callista-odoo-demo python-urllib"}
if DB:
    _HEADERS["X-Odoo-Database"] = DB


class OdooError(RuntimeError):
    """Server-side error. `.name` is the Python exception class, `.debug` the traceback."""
    def __init__(self, payload):
        self.name  = payload.get("name", "UnknownError")
        self.debug = payload.get("debug", "")
        super().__init__(f"{self.name}: {payload.get('message', '')}".strip())


def _post(path, body, timeout):
    req = urllib.request.Request(f"{URL}{path}", data=json.dumps(body).encode(),
                                 headers=_HEADERS, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            raise OdooError(json.loads(raw)) from None
        except json.JSONDecodeError:
            raise OdooError({"name": f"HTTP {e.code}", "message": raw[:500].decode(errors="replace")}) from None


def x(model, method, ids=None, timeout=600, **kwargs):
    """Call `method` on `model` over JSON-2.

    ids     : record id or list of ids. Omit for @api.model methods (search, create).
    kwargs  : every other argument, BY NAME — domain=, fields=, vals=, context=, ...
    timeout : seconds; module installs legitimately take minutes.
    """
    body = dict(kwargs)
    if ids is not None:
        body["ids"] = ids if isinstance(ids, list) else [ids]
    return _post(f"/json/2/{model}/{method}", body, timeout)


def version():
    """Server version — {"version_info": [19, 0, ...], "version": "19.0"}."""
    req = urllib.request.Request(f"{URL}/web/version")
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)


# JSON-2 has no authenticate(); the key identifies the user. context_get hands back its uid.
UID = x("res.users", "context_get").get("uid")
if not UID:
    raise SystemExit("Authentication failed — check ODOO_URL / ODOO_KEY (keys expire).")


def get_or_create(model, domain, vals, label=None):
    """Idempotent create: update if a record matches `domain`, else create.
    Two round trips per record — for more than a handful use get_or_create_many()."""
    ids = x(model, "search", domain=domain, limit=1)
    if ids:
        x(model, "write", ids, vals=vals)
        return ids[0]
    return x(model, "create", vals_list=vals)


def get_or_create_many(model, key_field, rows):
    """Batched get-or-create. `rows` is a list of vals dicts, each carrying `key_field`.

    One search_read for the whole set + one create for everything missing, instead of
    two calls per record. Returns {key_value: id}.
    """
    rows = [r for r in rows if r.get(key_field)]
    if not rows:
        return {}
    keys = [r[key_field] for r in rows]
    found = {r[key_field]: r["id"] for r in
             x(model, "search_read", domain=[[key_field, "in", keys]], fields=[key_field])}
    existing = [r for r in rows if r[key_field] in found]
    missing  = [r for r in rows if r[key_field] not in found]
    if missing:
        ids = x(model, "create", vals_list=missing)
        found.update(zip([r[key_field] for r in missing], ids))
    # ponytail: refreshing existing rows stays one call per record — differing vals can't be
    # batched into a single write. Only costs on re-runs; batch it if a config ever gets big.
    for r in existing:
        x(model, "write", found[r[key_field]], vals=r)
    return found


def fields_named(model, *needles):
    """Return field technical names on `model` whose name contains any needle.
    NOTE: on some v19 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("version:", version().get("version"))
    print("uid:", UID, "url:", URL, "db:", DB or "(from host)")