terrakernel-odxproxyclient

Python client SDK for ODXProxy — a reverse proxy that fronts one or more Odoo instances behind a unified JSON-RPC 2.0 API. By Terrakernel.

Importable as terrakernel.odxproxyclient (PEP 420 namespace under terrakernel).

• Sync and async, sharing one core. Pick the one that matches your runtime.
• HTTP/2 + brotli/gzip by default via httpx.
• Typed exceptions for every error code in the spec.
• Framework-friendly: drop-in patterns for FastAPI and Flask, but no framework dep.
• orjson for fast JSON serialization.
• Python >=3.12 (tested on 3.12 and 3.14). Wire protocol fully specified in SYSTEM_ARCHITECTURE.md.

---

Install

pip install terrakernel-odxproxyclient
# or, with uv
uv add terrakernel-odxproxyclient

Then import as:

from terrakernel.odxproxyclient import ODXProxyClient, AsyncODXProxyClient, OdooInstance

---

Two API keys, never confuse them

ODXProxy uses two completely separate credentials. The client mirrors this — they live in different places, intentionally:

Key	Where it goes	What it is
Proxy x-api-key	On the client (api_key=... in the constructor)	Sent as the x-api-key HTTP header. Authenticates you with the proxy.
Odoo api_key	On OdooInstance(api_key=...)	Carried inside the JSON-RPC body. Authenticates the proxy with Odoo as a specific user.

---

Quick start — sync

from terrakernel.odxproxyclient import ODXProxyClient, OdooInstance

with ODXProxyClient(
    base_url="https://proxy.example.com",
    api_key="PROXY_X_API_KEY",
) as client:
    session = client.for_instance(
        url="https://erp.example.com",
        db="prod",
        user_id=2,
        api_key="ODOO_USER_API_KEY",
    )

    partners = session.search_read(
        "res.partner",
        params=[[["is_company", "=", True]]],
        keyword={"fields": ["id", "name", "email"], "limit": 100},
    )
    print(partners)

Quick start — async

import asyncio
from terrakernel.odxproxyclient import AsyncODXProxyClient

async def main() -> None:
    async with AsyncODXProxyClient(
        base_url="https://proxy.example.com",
        api_key="PROXY_X_API_KEY",
    ) as client:
        session = client.for_instance(
            url="https://erp.example.com",
            db="prod",
            user_id=2,
            api_key="ODOO_USER_API_KEY",
        )
        partners = await session.search_read(
            "res.partner",
            params=[[["is_company", "=", True]]],
            keyword={"fields": ["id", "name"], "limit": 100},
        )
        print(partners)

asyncio.run(main())

---

API reference

Client constructors

Sync and async clients share the same signature:

ODXProxyClient(
    base_url: str,
    api_key: str,            # the proxy's x-api-key
    *,
    default_timeout_secs: float = 15.0,
    http_client: httpx.Client | None = None,        # async: httpx.AsyncClient
    http2: bool = True,
)

If http_client is provided, the SDK uses it and will NOT close it on close()/aclose(). Otherwise the SDK owns an internal httpx.Client and closes it.

OdooInstance (frozen dataclass)

All fields are required:

OdooInstance(
    url: str,        # full Odoo URL, e.g. "https://erp.example.com"
    db: str,         # Odoo database name
    user_id: int,    # Odoo user id (not login)
    api_key: str,    # Odoo user's api_key — different from the proxy x-api-key
)

Session factories

Two equivalent ways to build a Session / AsyncSession:

# Canonical: build OdooInstance once, reuse across sessions.
session = client.session(OdooInstance(url=..., db=..., user_id=..., api_key=...))

# Shortcut for one-shot scripts.
session = client.for_instance(url=..., db=..., user_id=..., api_key=...)

Both return the same object. Sessions are cheap (no I/O on construction) — make as many as you need.

Action methods on Session / AsyncSession

All signatures are uniform. First positional arg is model_id (the Odoo model name, e.g. "res.partner"). params and keyword are keyword-only. params is forwarded to Odoo's execute_kw as positional args; keyword is forwarded as kwargs. The client never rewrites Odoo domains.

def search(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[int]
def search_count(  model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> int
def read(          model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[dict]
def fields_get(    model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> dict
def search_read(   model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[dict]
def create(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> int | list[int]
def write(         model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> bool
def unlink(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> bool
def call_method(   model_id: str, fn_name: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> JsonValue

Return types are not statically narrowed. The methods' declared return type is JsonValue (None | bool | int | float | str | list | dict) because the proxy passes through whatever Odoo returns. The annotations above reflect what Odoo actually returns for each action — treat them as a guide, not a guarantee. If you need narrowing, assert isinstance(...) or cast(...) at the call site.

• request_id= — optional client-supplied UUID; if omitted, the SDK generates a UUID4.
• timeout_secs= — per-call only. Sent as the proxy's x-request-timeout header; this is the upstream Odoo timeout, not the local httpx timeout. Non-positive values are dropped (proxy default kicks in).

params / keyword shapes

params is wrapped to become positional args to Odoo's execute_kw(model, method, args, kwargs). The most common shapes:

# search / search_count / search_read: [[domain]]
session.search("res.partner", params=[[["is_company", "=", True]]])

# read: [ids, fields]
session.read("res.partner", params=[[1, 2, 3], ["name", "email"]])

# create: [vals]
session.create("res.partner", params=[{"name": "Acme"}])

# write: [ids, vals]
session.write("res.partner", params=[[1], {"name": "Acme Inc."}])

# unlink: [ids]
session.unlink("res.partner", params=[[1]])

# search_read with fields/limit/offset → use keyword
session.search_read(
    "res.partner",
    params=[[["is_company", "=", True]]],
    keyword={"fields": ["id", "name"], "limit": 100, "offset": 0},
)

# call_method: positional args of the target Odoo method go in params
session.call_method("account.move", "action_post", params=[[invoice_id]])

If a shape isn't here, it matches whatever execute_kw expects — the spec is Odoo's, not the proxy's.

Ops endpoints on the client itself

client.about()              -> BuildInfo(build: str, version: str)
client.license()            -> LicenseInfo(licensee: str, valid_until: str, is_valid: bool)
                               # valid_until is "YYYY-MM-DD" (string, no tz)
client.metrics()            -> str                # Prometheus text format
client.odoo_version(url)    -> JsonValue          # Odoo's version banner (dict)

---

FastAPI integration

Create one AsyncODXProxyClient for the whole app, manage its lifetime with lifespan, and inject sessions via Depends. The single client owns the HTTP/2 connection pool — sharing it across requests is the whole performance story.

# app/main.py
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI, HTTPException, Request
from terrakernel.odxproxyclient import (
    AsyncODXProxyClient,
    AsyncSession,
    AuthError,
    ODXProxyError,
    OdooInstance,
    OdooLogicError,
)

ODOO = OdooInstance(
    url="https://erp.example.com",
    db="prod",
    user_id=2,
    api_key="ODOO_USER_API_KEY",
)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.odx = AsyncODXProxyClient(
        base_url="https://proxy.example.com",
        api_key="PROXY_X_API_KEY",
        default_timeout_secs=15.0,
    )
    try:
        yield
    finally:
        await app.state.odx.aclose()


app = FastAPI(lifespan=lifespan)


async def get_session(request: Request) -> AsyncSession:
    """Inject a session bound to the default Odoo instance."""
    return request.app.state.odx.session(ODOO)


@app.exception_handler(ODXProxyError)
async def odx_error_handler(_: Request, exc: ODXProxyError) -> HTTPException:
    # Translate proxy/Odoo failures into HTTP errors your API wants to expose.
    if isinstance(exc, AuthError):
        raise HTTPException(status_code=502, detail="upstream auth failed")
    if isinstance(exc, OdooLogicError):
        raise HTTPException(status_code=400, detail=exc.message)
    raise HTTPException(status_code=502, detail=exc.message)


@app.get("/partners")
async def list_partners(
    limit: int = 50,
    session: AsyncSession = Depends(get_session),
) -> list[dict]:
    rows = await session.search_read(
        "res.partner",
        params=[[]],
        keyword={"fields": ["id", "name", "email"], "limit": limit},
    )
    assert isinstance(rows, list)
    return rows


@app.post("/invoices/{invoice_id}/post")
async def post_invoice(
    invoice_id: int,
    session: AsyncSession = Depends(get_session),
) -> dict:
    await session.call_method("account.move", "action_post", params=[[invoice_id]])
    return {"ok": True}

Notes

• One AsyncODXProxyClient per process; never instantiate it per request.
• If you talk to multiple Odoo databases, keep one client and call client.session(other_instance) per request — Session/AsyncSession is cheap and stateless.
• For per-tenant credentials, build the OdooInstance from request state (auth token, header, etc.) inside the Depends instead of using a module constant.

---

Flask integration

The sync ODXProxyClient is thread-safe (it wraps a long-lived httpx.Client). Construct one at app-factory time and reuse it across requests; close it at process shutdown.

# app/__init__.py
import atexit

from flask import Flask, current_app, jsonify
from terrakernel.odxproxyclient import (
    ODXProxyClient,
    OdooInstance,
    OdooLogicError,
    ODXProxyError,
)

ODOO = OdooInstance(
    url="https://erp.example.com",
    db="prod",
    user_id=2,
    api_key="ODOO_USER_API_KEY",
)


def create_app() -> Flask:
    app = Flask(__name__)
    app.config.from_mapping(
        ODX_BASE_URL="https://proxy.example.com",
        ODX_API_KEY="PROXY_X_API_KEY",
    )

    odx = ODXProxyClient(
        base_url=app.config["ODX_BASE_URL"],
        api_key=app.config["ODX_API_KEY"],
        default_timeout_secs=15.0,
    )
    app.extensions["odx"] = odx
    atexit.register(odx.close)

    @app.errorhandler(OdooLogicError)
    def _odoo_logic(exc: OdooLogicError):
        return jsonify(error=exc.message, code=exc.code), 400

    @app.errorhandler(ODXProxyError)
    def _odx(exc: ODXProxyError):
        return jsonify(error=exc.message, code=exc.code), 502

    @app.get("/partners")
    def list_partners():
        session = current_app.extensions["odx"].session(ODOO)
        rows = session.search_read(
            "res.partner",
            params=[[]],
            keyword={"fields": ["id", "name", "email"], "limit": 50},
        )
        return jsonify(rows)

    @app.post("/invoices/<int:invoice_id>/post")
    def post_invoice(invoice_id: int):
        session = current_app.extensions["odx"].session(ODOO)
        session.call_method("account.move", "action_post", params=[[invoice_id]])
        return jsonify(ok=True)

    return app

Why atexit instead of teardown_appcontext? The client owns a connection pool meant to live for the whole process. teardown_appcontext fires after every request and would defeat the pool. Close at process exit.

If you're running Flask with a worker model (gunicorn, uwsgi), each worker process gets its own client — that's correct and what you want.

---

Error handling

Every wire-level failure becomes a typed exception, all inheriting from ODXProxyError. Catch the base if you want one handler; catch a subclass if you want to react specifically.

from terrakernel.odxproxyclient import (
    ODXProxyError,
    AuthError,             # -32000 / HTTP 401
    InvalidActionError,    # -32001 / HTTP 400
    MissingFnNameError,    # -32002 / HTTP 400
    OdooTimeoutError,      # -32003 / HTTP 504 (also: local httpx timeout)
    OdooConnectError,      # -32004 / HTTP 502 (also: local httpx connect error)
    InternalProxyError,    # -32005 / HTTP 500
    LicenseError,          # code 0 / HTTP 403
    OdooLogicError,        # any other code / HTTP 200 (Odoo pass-through)
    TransportError,        # unrecognized / malformed envelope
)

try:
    session.write("res.partner", params=[[1], {"name": ""}])
except OdooLogicError as e:
    # Odoo-side validation error — 200 OK with an error envelope.
    print(e.code, e.message, e.data)
except OdooTimeoutError:
    # Either the proxy reported -32003 OR the local httpx call timed out.
    ...
except ODXProxyError as e:
    print(e.code, e.http_status, e.request_id)

HTTP 200 is not always success. The proxy returns Odoo-side logic errors as HTTP 200 with an error object in the body. The client handles this for you — OdooLogicError will be raised. Do not implement your own status-code checks; let the typed exceptions do the work.

---

Performance notes

• One client per process, always. The connection pool, HTTP/2 multiplexing, and TLS session reuse all live on the httpx.Client / AsyncClient that the proxy client owns. Throwing it away per request throws those away too.
• HTTP/2 is on by default (http2=True). Disable with http2=False if your proxy speaks only HTTP/1.1.
• Response compression (gzip, br, deflate) is negotiated automatically by httpx — no SDK-side header juggling.
• orjson is used for both serialization and deserialization. Significantly faster than stdlib json for the kind of payloads search_read returns.
• No retries. This is deliberate. Wrap the client in your own retry policy (e.g. tenacity) if you need one — the SDK won't silently re-fire calls that might have side effects.

---

Advanced: shared httpx client

If you already manage a process-wide httpx.Client / AsyncClient and want the SDK to use it, pass it in. The SDK will not close clients it doesn't own.

import httpx
from terrakernel.odxproxyclient import AsyncODXProxyClient

http = httpx.AsyncClient(base_url="https://proxy.example.com", http2=True, timeout=15.0)
odx = AsyncODXProxyClient(
    base_url="https://proxy.example.com",
    api_key="PROXY_X_API_KEY",
    http_client=http,
)
# ... use odx ...
await odx.aclose()   # does NOT close `http`
await http.aclose()  # your responsibility

---

Testing

pip install -e ".[dev]"

# Unit tests (hermetic, respx-mocked, no network)
pytest -m "not integration"

# Integration tests against a real proxy
cp tests/integration/credentials.example.toml tests/integration/credentials.toml
$EDITOR tests/integration/credentials.toml
pytest -m integration

Integration tests are read-only by design — they exercise about, license, metrics, odoo_version, search_count, search_read, fields_get, plus a negative auth case. They never create, write, or unlink. The integration suite auto-skips when credentials.toml is absent, so a fresh clone runs only the unit tests.

---

License

MIT — see LICENSE. Copyright © 2026 Terrakernel.