53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
import os
|
|
import requests
|
|
import random
|
|
from typing import Any, Dict, List
|
|
|
|
|
|
class OdooClient:
|
|
def __init__(self):
|
|
self.base_url = "https://www.odoo.com"
|
|
self.session = requests.Session()
|
|
if session_id := os.getenv("ODOO_SESSION_ID"):
|
|
self.session.cookies.set("session_id", session_id)
|
|
else:
|
|
raise Exception("ODOO_SESSION_ID is not set")
|
|
self.session.cookies.set("cids", "1")
|
|
self.session.cookies.set("frontend_lang", "en_US")
|
|
|
|
def rpc(
|
|
self, model: str, method: str, *args: Any, **kwargs: Any
|
|
) -> requests.Response:
|
|
body: Dict[str, Any] = {
|
|
"jsonrpc": "2.0",
|
|
"method": method,
|
|
"id": random.randint(1, 1000000),
|
|
"params": {
|
|
"model": model,
|
|
"method": method,
|
|
"args": args,
|
|
"kwargs": kwargs,
|
|
},
|
|
}
|
|
|
|
return self.session.post(
|
|
f"{self.base_url}/web/dataset/call_kw",
|
|
json=body,
|
|
)
|
|
|
|
def web_search_read(
|
|
self,
|
|
model: str,
|
|
domain: List[tuple],
|
|
spec: Dict[str, Any],
|
|
*args: Any,
|
|
**kwargs: Any,
|
|
) -> List[Dict[str, Any]]:
|
|
args = [domain, spec, *args]
|
|
res = self.rpc(model, "web_search_read", *args, **kwargs)
|
|
res.raise_for_status()
|
|
json = res.json()
|
|
if json.get("error"):
|
|
raise Exception(json.get("error"))
|
|
return json.get("result", {}).get("records", [])
|