168 lines
4.5 KiB
Python
168 lines
4.5 KiB
Python
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()
|