Release v4928.1.4.2 stable

app/odoo_client.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import ssl
+import xmlrpc.client
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from app.config import settings
+
+
+class OdooIntegrationError(RuntimeError):
+    pass
+
+
+@dataclass
+class OdooConfig:
+    base_url: str
+    db: str
+    username: str
+    api_key: str
+
+
+def get_odoo_config() -> OdooConfig:
+    base_url = (settings.odoo_base_url or "").rstrip("/")
+    db = settings.odoo_db or ""
+    username = settings.odoo_username or ""
+    api_key = settings.odoo_api_key or ""
+
+    if not base_url:
+        raise OdooIntegrationError("ODOO_BASE_URL não está configurado.")
+    if not db:
+        raise OdooIntegrationError("ODOO_DB não está configurado.")
+    if not username:
+        raise OdooIntegrationError("ODOO_USERNAME não está configurado.")
+    if not api_key:
+        raise OdooIntegrationError("ODOO_API_KEY não está configurado.")
+
+    return OdooConfig(base_url=base_url, db=db, username=username, api_key=api_key)
+
+
+class OdooClient:
+    def __init__(self, config: Optional[OdooConfig] = None):
+        self.config = config or get_odoo_config()
+        self._uid: Optional[int] = None
+        context = ssl._create_unverified_context()
+
+        self.common = xmlrpc.client.ServerProxy(
+            f"{self.config.base_url}/xmlrpc/2/common",
+            allow_none=True,
+            context=context,
+        )
+        self.models = xmlrpc.client.ServerProxy(
+            f"{self.config.base_url}/xmlrpc/2/object",
+            allow_none=True,
+            context=context,
+        )
+
+    def version(self) -> Dict[str, Any]:
+        try:
+            return dict(self.common.version())
+        except Exception as exc:
+            raise OdooIntegrationError(f"Falha ao consultar versão Odoo: {exc}") from exc
+
+    def authenticate(self) -> int:
+        if self._uid:
+            return self._uid
+        try:
+            uid = self.common.authenticate(
+                self.config.db,
+                self.config.username,
+                self.config.api_key,
+                {},
+            )
+        except Exception as exc:
+            raise OdooIntegrationError(f"Falha de autenticação Odoo: {exc}") from exc
+
+        if not uid:
+            raise OdooIntegrationError("Autenticação Odoo falhou. Verifica ODOO_USERNAME e ODOO_API_KEY/password.")
+
+        self._uid = int(uid)
+        return self._uid
+
+    def execute_kw(
+        self,
+        model: str,
+        method: str,
+        args: Optional[List[Any]] = None,
+        kwargs: Optional[Dict[str, Any]] = None,
+    ) -> Any:
+        uid = self.authenticate()
+        try:
+            return self.models.execute_kw(
+                self.config.db,
+                uid,
+                self.config.api_key,
+                model,
+                method,
+                args or [],
+                kwargs or {},
+            )
+        except Exception as exc:
+            raise OdooIntegrationError(f"Odoo {model}.{method} falhou: {exc}") from exc
+
+    def search_read(
+        self,
+        model: str,
+        domain: Optional[List[Any]] = None,
+        fields: Optional[List[str]] = None,
+        *,
+        limit: int = 500,
+        offset: int = 0,
+        order: str = "id asc",
+        context: Optional[Dict[str, Any]] = None,
+    ) -> List[Dict[str, Any]]:
+        kwargs: Dict[str, Any] = {
+            "fields": fields or [],
+            "limit": int(limit),
+            "offset": int(offset),
+            "order": order,
+        }
+        if context is not None:
+            kwargs["context"] = context
+        return list(self.execute_kw(model, "search_read", [domain or []], kwargs) or [])
+
+    def count(self, model: str, domain: Optional[List[Any]] = None) -> int:
+        return int(self.execute_kw(model, "search_count", [domain or []]) or 0)