117 lines
3.1 KiB
Python
117 lines
3.1 KiB
Python
import requests
|
|
from datetime import date
|
|
import simplejson as json
|
|
|
|
from app.models import MODELS
|
|
|
|
|
|
def encoder(obj):
|
|
# FIXME: add datetime, buffer, bytes
|
|
if isinstance(obj, date):
|
|
return {
|
|
'__class__': 'date',
|
|
'year': obj.year,
|
|
'month': obj.month,
|
|
'day': obj.day,
|
|
}
|
|
raise TypeError(repr(obj) + " is not JSON serializable")
|
|
|
|
|
|
class FastModel(object):
|
|
|
|
def __init__(self, model, ctx, fields=None):
|
|
self.model = model
|
|
self.ctx = ctx
|
|
api_url = ctx['params']['api_url']
|
|
db = ctx['params']['database']
|
|
self.api = '/'.join(['http:/', api_url, db])
|
|
_model = MODELS.get(model)
|
|
self.fields = None
|
|
if fields:
|
|
self.fields = fields
|
|
elif _model.get('fields'):
|
|
self.fields = _model['fields']
|
|
else:
|
|
print('No hay model ', model)
|
|
|
|
|
|
def __getattr__(self, name, *args):
|
|
'Return attribute value'
|
|
self.method = name
|
|
return self
|
|
|
|
def find(self, domain, order=None, limit=1000, ctx=None, fields=None):
|
|
if ctx:
|
|
self.ctx.update(ctx)
|
|
route = self.get_route('search')
|
|
if not fields:
|
|
fields = self.fields
|
|
args_ = {
|
|
'model': self.model,
|
|
'domain': domain,
|
|
'order': order,
|
|
'limit': limit,
|
|
'fields': fields,
|
|
'context': self.ctx,
|
|
}
|
|
data = json.dumps(args_, default=encoder)
|
|
res = requests.post(route, data=data)
|
|
return res.json()
|
|
|
|
def write(self, ids, values):
|
|
route = self.get_route('save')
|
|
args_ = {
|
|
'model': self.model,
|
|
'id': ids[0],
|
|
'record_data': values,
|
|
'context': self.ctx,
|
|
}
|
|
data = json.dumps(args_, default=encoder)
|
|
res = requests.put(route, data=data)
|
|
return res.json()
|
|
|
|
def delete(self, ids):
|
|
route = self.get_route('delete')
|
|
args_ = {
|
|
'model': self.model,
|
|
'ids': ids,
|
|
'context': self.ctx,
|
|
}
|
|
data = json.dumps(args_, default=encoder)
|
|
res = requests.delete(route, data=data)
|
|
return res.json()
|
|
|
|
def get_route(self, target):
|
|
route = self.api + '/' + target
|
|
return route
|
|
|
|
def __call__(self, args=None):
|
|
args_ = {
|
|
'model': self.model,
|
|
'method': self.method,
|
|
'args': args,
|
|
'context': self.ctx,
|
|
}
|
|
route = self.get_route('model_method')
|
|
data = json.dumps(args_, default=encoder)
|
|
res = requests.post(route, data=data)
|
|
response = None
|
|
try:
|
|
response = res.json()
|
|
except ValueError:
|
|
print('Error converting json: ', res.text)
|
|
return response
|
|
|
|
|
|
if __name__ == "__main__":
|
|
params = {'api_url': 'localhost:5070', 'database': 'DEMO50'}
|
|
model = {'model': 'sale.sale'}
|
|
test_model = FastModel(params, model)
|
|
id = 180
|
|
data = {
|
|
'reference': 'OC-02874'
|
|
}
|
|
# res = test_model.find(dom)
|
|
res = test_model.write([id], data)
|
|
print(res)
|