feat(integrations): add Odoo XML-RPC client

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

services/integrations/app/odoo/client.py

@@ -0,0 +1,167 @@
+import xmlrpc.client
+from typing import Any, Optional
+from functools import lru_cache
+
+from app.config import get_settings
+from app.odoo.exceptions import (
+    OdooConnectionError,
+    OdooAuthError,
+    OdooNotFoundError,
+    OdooValidationError,
+    OdooError,
+)
+
+settings = get_settings()
+
+
+class OdooClient:
+    """XML-RPC client for Odoo"""
+
+    def __init__(
+        self,
+        url: str = None,
+        db: str = None,
+        user: str = None,
+        api_key: str = None,
+    ):
+        self.url = url or settings.ODOO_URL
+        self.db = db or settings.ODOO_DB
+        self.user = user or settings.ODOO_USER
+        self.api_key = api_key or settings.ODOO_API_KEY
+        self._uid: Optional[int] = None
+        self._common = None
+        self._models = None
+
+    def _get_common(self):
+        if not self._common:
+            try:
+                self._common = xmlrpc.client.ServerProxy(
+                    f"{self.url}/xmlrpc/2/common",
+                    allow_none=True,
+                )
+            except Exception as e:
+                raise OdooConnectionError(f"Failed to connect: {e}")
+        return self._common
+
+    def _get_models(self):
+        if not self._models:
+            try:
+                self._models = xmlrpc.client.ServerProxy(
+                    f"{self.url}/xmlrpc/2/object",
+                    allow_none=True,
+                )
+            except Exception as e:
+                raise OdooConnectionError(f"Failed to connect: {e}")
+        return self._models
+
+    def authenticate(self) -> int:
+        """Authenticate and return user ID"""
+        if self._uid:
+            return self._uid
+
+        if not all([self.url, self.db, self.user, self.api_key]):
+            raise OdooAuthError("Missing Odoo credentials")
+
+        try:
+            common = self._get_common()
+            uid = common.authenticate(self.db, self.user, self.api_key, {})
+            if not uid:
+                raise OdooAuthError("Invalid credentials")
+            self._uid = uid
+            return uid
+        except OdooAuthError:
+            raise
+        except Exception as e:
+            raise OdooConnectionError(f"Authentication failed: {e}")
+
+    def execute(
+        self,
+        model: str,
+        method: str,
+        *args,
+        **kwargs,
+    ) -> Any:
+        """Execute Odoo method"""
+        uid = self.authenticate()
+        models = self._get_models()
+
+        try:
+            return models.execute_kw(
+                self.db,
+                uid,
+                self.api_key,
+                model,
+                method,
+                list(args),
+                kwargs if kwargs else {},
+            )
+        except xmlrpc.client.Fault as e:
+            if "not found" in str(e).lower():
+                raise OdooNotFoundError(str(e))
+            if "validation" in str(e).lower():
+                raise OdooValidationError(str(e))
+            raise OdooError(str(e))
+
+    def search(
+        self,
+        model: str,
+        domain: list,
+        limit: int = None,
+        offset: int = 0,
+        order: str = None,
+    ) -> list:
+        """Search records"""
+        kwargs = {"offset": offset}
+        if limit:
+            kwargs["limit"] = limit
+        if order:
+            kwargs["order"] = order
+        return self.execute(model, "search", domain, **kwargs)
+
+    def read(
+        self,
+        model: str,
+        ids: list,
+        fields: list = None,
+    ) -> list:
+        """Read records by IDs"""
+        kwargs = {}
+        if fields:
+            kwargs["fields"] = fields
+        return self.execute(model, "read", ids, **kwargs)
+
+    def search_read(
+        self,
+        model: str,
+        domain: list,
+        fields: list = None,
+        limit: int = None,
+        offset: int = 0,
+        order: str = None,
+    ) -> list:
+        """Search and read in one call"""
+        kwargs = {"offset": offset}
+        if fields:
+            kwargs["fields"] = fields
+        if limit:
+            kwargs["limit"] = limit
+        if order:
+            kwargs["order"] = order
+        return self.execute(model, "search_read", domain, **kwargs)
+
+    def create(self, model: str, values: dict) -> int:
+        """Create a record"""
+        return self.execute(model, "create", [values])
+
+    def write(self, model: str, ids: list, values: dict) -> bool:
+        """Update records"""
+        return self.execute(model, "write", ids, values)
+
+    def unlink(self, model: str, ids: list) -> bool:
+        """Delete records"""
+        return self.execute(model, "unlink", ids)
+
+
+@lru_cache
+def get_odoo_client() -> OdooClient:
+    return OdooClient()