1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
|
from os import path
from time import sleep
import re
import csv
from xmlrpclib import Fault
openerplib = __import__('openerplib')
import config
try:
import ipdb as pdb
except:
import pdb
class db(object):
def __init__(self, server, db_name, user=None, password=None, server_password=None):
self.server = server
try:
self.server_password = config.db_password
except:
self.server_password = None
finally:
if server_password: self.server_password = server_password
self.db_name = db_name
self.service = server.get_service('db')
if user: self.connect(user, password)
def exec_workflow(self, model, method, ids):
return self.server.get_service('object').exec_workflow(
self.server.database, self.server.user_id, self.server.password,
model, method, ids)
def connect(self, login, password):
self.server = openerplib.get_connection(hostname=self.server.connector.hostname, port=self.server.connector.port, database=self.db_name, login=login, password=password)
return self
def create_db(self, password, demo=False, lang='en_US', wait=False):
if not self.server_password: raise Exception, "The server password is needed for this operation"
self.id = self.service.create(self.server_password, self.db_name, demo, 'en_US', password)
if wait: self.wait()
return self
create = create_db
def wait(self):
if not self.server_password: raise Exception, "The server password is needed for this operation"
while True:
try:
running = self.service.get_progress(self.server_password, self.id)
if not running: break
sleep(1)
except:
break
return self
def drop(self):
if not self.server_password: raise Exception, "The server password is needed for this operation"
## TODO does'n handle database current access prevent dropping
#self.service.drop(self.server_password, self.db_name)
try:
self.service.drop(self.server_password, self.db_name)
except Fault, e:
if not re.search(r"database ([\"']).+\1 does not exist", e.faultCode): raise e
finally:
return self
def get(self, model):
return self.server.get_model(model)
def wizard(self, model, data):
return wizard(self, model, data)
def ref(self, xml_id, index=0):
return ref(xml_id).get(self)
#def ref(self, model, xml_id, index=0):
# return ref(model, xml_id).get(self)
def module(self, name, index=0):
return module(self, name, index)
def action_done(self, action_id):
proxy = self.get('ir.actions.todo')
ids = proxy.search([('action_id', '=', action_id)])
if ids: proxy.write(ids, {'state' : 'done'})
def user(self, login):
return user(self, login)
def group(self, name):
return group(self, name)
def search_data(self, model, data, domain=None, action=None, active='active', pdb=False):
if isinstance(data, dict):
searchDomain = [(field,'=',value) for (field,value) in data.items() if value]
elif isinstance(data, (list,tuple)):
searchDomain = list(data)
else:
raise NotImplementedError("Cannot transform %s (of type %s) to a domain (list)" % (data, type(data).__name__))
if domain:
searchDomain += list(domain)
if active == 'unactive': searchDomain.append(('active','=',0))
elif active == 'both': searchDomain.extend(['|',('active','=',0),('active','=',1)])
elif not active == 'active': raise Exception, "'active' prameter must be one of these: active, unactive or both"
proxy = self.get(model)
ids = proxy.search(searchDomain)
if not ids and pdb:
pdb.set_trace()
if action == 'unlink': return proxy.unlink(ids) if ids else True
elif action == 'test': return bool(ids)
else: return ids
test = lambda self, model, data, **o: self.search_data(model, data, action='test', **o)
clean = lambda self, model, data, **o: self.search_data(model, data, action='unlink', **o)
def write(self, model, ids_or_domain, data):
if all([isinstance(x, (tuple, list)) for x in ids_or_domain]):
ids = self.get(model).search(ids_or_domain)
elif isinstance(ids_or_domain, dict):
ids = self.search_data(model, ids_or_domain)
else:
ids = ids_or_domain
if ids and not self.get(model).write(ids, data):
return 0
return len(ids)
def activate(self, model, domain, state=1):
proxy = self.get(model)
search_domain = [('active','=',1-state)] + domain
ids = proxy.search(search_domain)
proxy.write(ids, {'active':state})
return len(ids)
def desactivate(self, model, domain):
return self.activate(model, domain, 0)
def load_datas(self, filepath):
datas = []
if path.isfile(filepath):
if filepath[-4:] == '.csv':
ch = csv.reader(open(filepath, 'rb'), quotechar='"', delimiter=',')
fields = ch.next()
datetime_re = re.compile(r"^\d\d\d\d-\d\d-\d\d( \d\d:\d\d:\d\d)?$")
for values in ch:
evaluated_values = []
for v in values:
try:
if datetime_re.match(v): raise Exception
evaluated_values += [eval(v)]
except:
evaluated_values += [eval('"""'+v+'"""')]
datas.append(dict(zip(fields, evaluated_values)))
elif filepath[-4:] == '.xml':
raise Exception, 'While importing XML file: Not yet implemented'
else:
raise Exception, 'Cannot import file "%s": unrecognized file type!' % (filepath,)
else:
raise IOError, 'Can\'t access to file "%s", check read permission!' % (filepath,)
return datas
def Import(self, model, datas, mode='init'):
if type(datas) == dict:
fields = datas.keys()
datas = [datas.values()]
elif not type(datas) == list:
raise Exception, 'Cannot import data: datas must be a list of dict!'
else:
fields = []
for i in datas:
fields = set(list(fields) + i.keys())
fields = list(fields)
old_datas = datas
datas = []
for i in old_datas:
datas.append([i.get(field, None) for field in fields])
for data in datas:
for i, v in enumerate(data):
data[i] = str(data[i])
#if v == None: data[i] = False
#elif type(v) == bool: data[i] = 1 if v else 0
if 'reconcile note' in fields: ipdb.set_trace()
result, rows, warning_msg, dummy = self.get(model).import_data(fields, datas, mode)
if result == -1:
raise Exception, "Unable to import data: "+str(warning_msg)
return self
class ref(object):
def __init__(self, xml_id, index=0):
self.xml_id = xml_id
self.index = 0
def get(self, db, refetch=False):
try:
if refetch: raise AttributeError
return self.value
except AttributeError:
module, xml_id = self.xml_id.split('.')
model, i = db.get('ir.model.data').get_object_reference(module, xml_id)
self.value = db.get(model).read([i], ['id'])[0]
return self.value['id']
class wizard(object):
def __init__(self, db, model, data):
for k, v in data.items():
if isinstance(v, ref):
data[k] = v.get(db)
self.proxy = db.get(model)
self.id = self.proxy.create(data)
def __getattr__(self, attr):
real_attr = getattr(self.proxy, attr)
def foo(*args):
return real_attr([self.id], *args)
return foo
def data(self, *attrs):
data = self.proxy.read([self.id], attrs)
if len(attrs) == 1: return data[0][attrs[0]]
else: return data[0]
class module(object):
def __init__(self, db, name, index=0):
self.db = db
self.module_proxy = db.get('ir.module.module')
if name == 'all':
name = 'base'
#self.ids = self.module_proxy.search([('state','=','installed')])
#if not self.ids:
# raise Exception, "No module installed!"
#else:
self.ids = self.module_proxy.search([('name','=',name)])
if not self.ids:
raise Exception, "Unable to find module %s!" % (name,)
def remove(self):
self.expect = 'uninstalled'
self.module_proxy.button_uninstall(self.ids)
return self
uninstall = remove
def install(self):
self.expect = 'installed'
self.module_proxy.button_install(self.ids)
return self
def upgrade(self):
self.expect = 'installed'
self.module_proxy.button_upgrade(self.ids)
return self
def do(self):
self.db.get('base.module.upgrade').upgrade_module([])
if self.module_proxy.search([('id','in',self.ids),('state','=',self.expect)], 0, False, False, True) == len(self.ids):
raise Exception, "Modules modifications not applied"
return self
class user(object):
def __init__(self, db, login):
self.db = db
self.users = db.get('res.users')
self.login = login
try: self.id = self.users.search([('login','=',login)])[0]
except: self.id = None
def delGroups(self, *groups):
if not self.id: raise Exception, 'Cannot remove groups to unknown user: '+self.login
group_ids = [i.id if type(i) == group else group(self.db, i).id for i in groups]
if None in group_ids: raise Exception, 'Cannot find some groups in this list: '+', '.join(groups)
group_changes = [(3,i,) for i in group_ids]
self.users.write([self.id],{'groups_id':group_changes})
return self
def addGroups(self, *groups):
if not self.id: raise Exception, 'Cannot add groups to unknown user: '+self.login
group_ids = [i.id if type(i) == group else group(self.db, i).id for i in groups]
if None in group_ids: raise Exception, 'Cannot find some groups in this list: '+', '.join(groups)
group_changes = [(4,i,) for i in group_ids]
self.users.write([self.id],{'groups_id':group_changes})
return self
def add(self, password, name=None):
if not self.id:
self.id = self.users.create({'login':self.login,'name':(name or self.login),'password':password,})
return self
def exists(self):
return bool(self.id)
class group(object):
def __init__(self, db, name):
self.db = db
self.groups = db.get('res.groups')
self.name = name
try: self.id = self.groups.search([('name','=',name)])[0]
except: self.id = None
def delUsers(self, *users):
if not self.id: self.add()
user_ids = [i.id if type(i) == user else user(self.db, i).id for i in users]
if None in user_ids: raise Exception, 'Cannot find some users in this list: '+', '.join(users)
user_changes = [(3,i,) for i in user_ids]
self.groups.write([self.id],{'users':user_changes})
return self
def addUsers(self, *users):
if not self.id: self.add()
user_ids = [i.id if type(i) == user else user(self.db, i).id for i in users]
if None in user_ids: raise Exception, 'Cannot find some users in this list: '+', '.join(users)
user_changes = [(4,i,) for i in user_ids]
self.groups.write([self.id],{'users':user_changes})
return self
def add(self):
if not self.id:
self.id = self.groups.create({'name':self.name,})
return self
def exists(self):
return bool(self.id)
|