86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
# odoo_rpc_client.py
|
|
import ssl
|
|
import xmlrpc.client
|
|
|
|
import certifi
|
|
|
|
context = ssl._create_unverified_context()
|
|
|
|
class OdooRPCClient:
|
|
def __init__(self, url, db, username, password):
|
|
self.url = url.rstrip("/")
|
|
self.db = db
|
|
self.username = username
|
|
self.password = password
|
|
self.uid = None
|
|
self.common = None
|
|
self.models = None
|
|
self._connect()
|
|
|
|
def _connect(self):
|
|
self.common = xmlrpc.client.ServerProxy('{}/xmlrpc/2/common'.format(self.url),context=context)
|
|
self.uid = self.common.authenticate(self.db, self.username, self.password, {})
|
|
if not self.uid:
|
|
raise Exception("Authentication failed.")
|
|
self.models = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/object")
|
|
|
|
def search(self, model, domain, offset=0, limit=None, order=None):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'search',
|
|
[domain],
|
|
{'offset': offset, 'limit': limit, 'order': order}
|
|
)
|
|
|
|
def read(self, model, ids, fields=None):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'read',
|
|
[ids], {'fields': fields} if fields else {}
|
|
)
|
|
|
|
def search_read(self, model, domain, fields=None, offset=0, limit=None, order=None):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'search_read',
|
|
[domain],
|
|
{
|
|
'fields': fields or [],
|
|
'offset': offset,
|
|
'limit': limit,
|
|
'order': order
|
|
}
|
|
)
|
|
|
|
def create(self, model, data):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'create',
|
|
[data]
|
|
)
|
|
|
|
def write(self, model, ids, data):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'write',
|
|
[ids, data]
|
|
)
|
|
|
|
def unlink(self, model, ids):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, 'unlink',
|
|
[ids]
|
|
)
|
|
|
|
def execute(self, model, method, *args, **kwargs):
|
|
return self.models.execute_kw(
|
|
self.db, self.uid, self.password,
|
|
model, method,
|
|
list(args),
|
|
kwargs
|
|
)
|
|
|
|
def get_uid(self):
|
|
return self.uid
|