116 lines
4.4 KiB
Python
116 lines
4.4 KiB
Python
"""Odoo client abstraction.
|
|
|
|
Cetana never spreads raw Odoo calls through the codebase. Everything goes
|
|
through this client behind an interface. The concrete client is chosen at
|
|
runtime from the dynamic settings store (apps.core.settings_service), so the
|
|
integration can be configured and switched from the admin panel without a
|
|
deploy — with a mock fallback when Odoo is not configured.
|
|
"""
|
|
import xmlrpc.client
|
|
from typing import Protocol
|
|
|
|
from apps.core.settings_service import SettingsService
|
|
|
|
|
|
class OdooClientInterface(Protocol):
|
|
def search_partner(self, key: str, value) -> dict | None: ...
|
|
def create_partner(self, payload: dict) -> dict: ...
|
|
def create_invoice(self, payload: dict) -> dict: ...
|
|
|
|
|
|
class MockOdooClient:
|
|
"""Deterministic in-process stand-in used when Odoo is not configured."""
|
|
|
|
def search_partner(self, key: str, value) -> dict | None:
|
|
return None
|
|
|
|
def create_partner(self, payload: dict) -> dict:
|
|
return {"id": f"MOCK-PARTNER-{abs(hash(str(payload))) % 10_000_000}"}
|
|
|
|
def create_invoice(self, payload: dict) -> dict:
|
|
return {"id": f"MOCK-INV-{abs(hash(str(payload))) % 10_000_000}"}
|
|
|
|
|
|
class OdooXmlRpcClient:
|
|
"""Real Odoo client over the standard XML-RPC API (stdlib only)."""
|
|
|
|
def __init__(self, url, db, username, password="", api_key="", auth_method="password"):
|
|
self.url = (url or "").rstrip("/")
|
|
self.db = db
|
|
self.username = username
|
|
self.password = password
|
|
self.api_key = api_key
|
|
self.auth_method = auth_method
|
|
self._uid = None
|
|
|
|
def _authenticate(self):
|
|
common = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/common")
|
|
secret = self.api_key if self.auth_method == "api_key" else self.password
|
|
self._uid = common.authenticate(self.db, self.username, secret, {})
|
|
return self._uid
|
|
|
|
def _execute(self, model, method, method_args, kwargs=None):
|
|
if self._uid is None:
|
|
self._authenticate()
|
|
secret = self.api_key if self.auth_method == "api_key" else self.password
|
|
models = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/object")
|
|
return models.execute_kw(self.db, self._uid, secret, model, method, method_args, kwargs or {})
|
|
|
|
def search_partner(self, key: str, value) -> dict | None:
|
|
results = self._execute(
|
|
"res.partner", "search_read",
|
|
[[[key, "=", value]]],
|
|
{"limit": 1, "fields": ["id", "name"]},
|
|
)
|
|
return results[0] if results else None
|
|
|
|
def create_partner(self, payload: dict) -> dict:
|
|
partner_id = self._execute("res.partner", "create", [payload])
|
|
return {"id": partner_id}
|
|
|
|
def create_invoice(self, payload: dict) -> dict:
|
|
ref = payload.get("partner_ref") or ""
|
|
partner = None
|
|
if ref:
|
|
partner = self.search_partner("ref", ref) or self.search_partner("name", ref)
|
|
if partner is None and ref:
|
|
partner = {"id": self._execute("res.partner", "create", [{"name": ref}])}
|
|
vals = {
|
|
"move_type": "out_invoice",
|
|
"ref": payload.get("reference"),
|
|
"partner_id": partner["id"] if partner else False,
|
|
"invoice_line_ids": [(0, 0, {
|
|
"name": payload.get("reference") or "Processing",
|
|
"quantity": 1,
|
|
"price_unit": float(payload.get("amount") or 0),
|
|
})],
|
|
}
|
|
invoice_id = self._execute("account.move", "create", [vals])
|
|
return {"id": invoice_id}
|
|
|
|
def test_connection(self) -> dict:
|
|
"""Validate reachability + credentials. Returns server version and uid."""
|
|
common = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/common")
|
|
version = common.version()
|
|
uid = self._authenticate()
|
|
if not uid:
|
|
raise ValueError("Authentication failed — check database, username and secret.")
|
|
return {"version": version, "uid": uid}
|
|
|
|
|
|
def get_odoo_client() -> OdooClientInterface:
|
|
service = SettingsService()
|
|
if not service.get_bool("odoo.enabled", False):
|
|
return MockOdooClient()
|
|
url = service.get("odoo.url", "")
|
|
if not url:
|
|
return MockOdooClient()
|
|
return OdooXmlRpcClient(
|
|
url=url,
|
|
db=service.get("odoo.database", ""),
|
|
username=service.get("odoo.username", ""),
|
|
password=service.get("odoo.password", ""),
|
|
api_key=service.get("odoo.api_key", ""),
|
|
auth_method=service.get("odoo.auth_method", "password"),
|
|
)
|