-
Notifications
You must be signed in to change notification settings - Fork 12
/
session.py
392 lines (319 loc) · 13.3 KB
/
session.py
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
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
import numbers
import os.path
import sys
import pprint
from getpass import getpass
from extend_me import Extensible
# project imports
from .core import Client
from .utils import (json_read,
json_write,
xinput,
DirMixIn)
__all__ = ('Session',)
class SessionClientExt(Client):
""" Simple Client extension to add attribute '_no_save'
used in session
"""
def __init__(self, *args, **kwargs):
super(SessionClientExt, self).__init__(*args, **kwargs)
self._no_save = False
class Session(Extensible, DirMixIn):
""" Simple session manager which allows to manage databases easier
This class stores information about databases You used in home
directory, and on init it loads history and allows simply connect
to database by url or index. No more hosts, usernames, ports, etc...
required to be memorized.
Just on session start call::
>>> print(session)
And You will get all databases You worked with listed as (index, url)
pairs. To connect to one of thouse databases
just call session[index|url] and required
Client object will be returned.
:param data_file: path to session file
:type data_file: string
"""
def __init__(self, data_file='~/.openerp_proxy.json'):
"""
"""
self.data_file = os.path.expanduser(data_file)
# key: url; value: instance of DB or dict with init args
self._databases = {}
self._db_aliases = {} # key: aliase name; value: url
self._options = {}
self._db_index = {} # key: index; value: url
self._db_index_rev = {} # key: url; value: index
self._db_index_counter = 0 # max index used
if os.path.exists(self.data_file):
data = json_read(self.data_file)
self._databases = data.get('databases', {})
self._db_aliases = data.get('aliases', {})
self._options = data.get('options', {})
if 'index' in data:
for url, index in data['index'].items():
if url in self._databases:
self._db_index[index] = url
self._db_index_rev[url] = index
self._db_index_counter = max(self._db_index_counter, index)
for path in self.extra_paths:
self.add_path(path)
for module in self.start_up_imports: # pragma: no cover
try:
__import__(module)
except ImportError:
# TODO: implement some logging
pass
@property
def extra_paths(self):
""" List of extra pyhton paths, used by this session
"""
return self.option('extra_paths', default=[])
@property
def start_up_imports(self):
""" List of start-up imports
If You want some module to be automaticaly imported on
when session starts, that just add it to this list::
session.start_up_imports.append('openerp_proxy.ext.sugar')
"""
return self.option('start_up_imports', default=[])
def add_path(self, path):
""" Adds extra path to python import path.
:param path: Paths to be added
:type path: string
:return: None
Note: this way path will be saved in session
"""
# TODO: rewrite extrapaths logic with custom importers. It will be more
# pythonic
if path not in sys.path:
sys.path.append(path)
if path not in self.extra_paths:
self.extra_paths.append(path)
def option(self, opt, val=None, default=None):
""" Get or set option.
if *val* is passed, *val* will be set as value for option,
else just option value will be returned
:param str opt: option to get or set value for
:param val: value to be set for option *opt*
:return: value of option *opt*
Currently available options:
- store_passwords (bool) If set to True then all used
passwords will be stored on session.save. But be careful,
because of encription used for stored passwords is very week.
"""
if val is not None:
self._options[opt] = val
elif opt not in self._options and default is not None:
self._options[opt] = default
return self._options.get(opt, default)
@property
def aliases(self):
""" List of database aliases
To add new database aliase, use method *aliase*::
session.aliase('mdb', db) # db is instance of Client
"""
return self._db_aliases.copy()
def aliase(self, name, val):
""" Sets up aliase 'name' for *val*
:param name: new aliase
:type name: string
:param val: database to create aliase for
:type val: int|string|Client instance
*val* could be index, url or Client object::
session.aliase('tdb', 1)
session.aliase('mdb', 'xml-rpc://me@my.example.com:8069/my_db')
session.aliase('xdb', db)
And now You can use this aliase like::
session.tdb
session.mdb
session.xdb
:return: unchanged val
"""
if isinstance(val, Client):
url = val.get_url()
elif isinstance(val, numbers.Integral) and val in self.index:
url = self.index[val]
else:
url = val
if url in self._databases:
self._db_aliases[name] = url
else:
raise ValueError("Bad value type: %s" % val)
return val
@property
def index(self):
""" Property which returns dict with {index: url}
"""
if not self._db_index:
for url in self._databases.keys():
self._index_url(url)
return dict(self._db_index)
def _index_url(self, url):
""" Returns index of specified URL, or adds it to
store assigning new index
"""
if self._db_index_rev.get(url, False):
return self._db_index_rev[url]
self._db_index_counter += 1
self._db_index[self._db_index_counter] = url
self._db_index_rev[url] = self._db_index_counter
return self._db_index_counter
def add_db(self, db):
""" Add db to session.
param db: database (client instance) to be added to session
type db: Client instance
"""
url = db.get_url()
self._databases[url] = db
self._index_url(url)
def get_db(self, url_or_index, **kwargs):
""" Returns instance of Client object, that represents single
Odoo database it connected to, specified by passed index (integer)
or url (string) of database, previously saved in session.
:param url_or_index: must be integer (if index) or string (if url).
this parametr specifies database
to get from session
:type url_or_index: int|string
:param kwargs: can contain aditional arguments
to be passed on init of Client
:return: Client instance
:raises ValueError: if cannot find database by specified args
Examples::
session.get_db(1) # using index
session.get_db('xml-rpc://katyukha@erp.jbm.int:8069/jbm0')
session.get_db('my_db') # using aliase
"""
if isinstance(url_or_index, numbers.Integral):
url = self.index[url_or_index]
else:
url = self._db_aliases.get(url_or_index, url_or_index)
db = self._databases.get(url, False)
if not db:
raise ValueError("Bad url %s. not found in history or databases"
"" % url)
if isinstance(db, Client):
return db
ep_args = db.copy() # DB here is instance of dict
ep_args.update(**kwargs)
if 'pwd' not in ep_args:
if self.option('store_passwords') and 'password' in ep_args:
import base64
encoded_pwd = ep_args.pop('password').encode('utf8')
crypter, password = base64.b64decode(encoded_pwd).split(b':')
if crypter == b'simplecrypt': # pragma: no cover
import simplecrypt
ep_args['pwd'] = simplecrypt.decrypt(
Client.to_url(ep_args), base64.b64decode(password))
elif crypter == b'plain':
ep_args['pwd'] = password.decode('utf-8')
else: # pragma: no cover
raise Exception("Unknown crypter (%s) used in session"
"" % repr(crypter))
else:
ep_args['pwd'] = getpass('Password: ') # pragma: no cover
db = Client(**ep_args)
self.add_db(db)
return db
@property
def db_list(self):
""" Returns list of URLs of databases available in current session
:return: list of urls of databases from session
:rtype: list of strings
"""
return self._databases.keys()
def connect(self, host=None, dbname=None, user=None, pwd=None, port=8069,
protocol='xml-rpc', interactive=True, no_save=False):
""" Wraper around Client constructor class
to simplify connect from shell.
:param str host: host name to connect to
(will be asked interactvely if not provided)
:param str dbname: name of database to connect to
(will be asked interactvely if not provided)
:param str user: user name to connect as
(will be asked interactvely if not provided)
:param str pwd: password for selected user
(will be asked interactvely if not provided)
:param int port: port to connect to. (default: 8069)
:param bool interactive: ask for connection parameters
if not provided. (default: True)
:param bool no_save: if set to True database
will not be saved to session
:return: Client object
"""
if interactive: # pragma: no cover
# ask user for connection data if not provided, if interactive set
# to True
host = host or xinput('Server Host: ')
dbname = dbname or xinput('Database name: ')
user = user or xinput('Login: ')
pwd = pwd or getpass("Password: ")
url = Client.to_url(inst=None,
user=user,
host=host,
dbname=dbname,
port=port,
protocol=protocol)
db = self._databases.get(url, False)
if isinstance(db, Client):
return db
db = Client(host=host,
dbname=dbname,
user=user,
pwd=pwd,
port=port,
protocol=protocol)
self.add_db(db)
# if set to True, disalows saving database connection in session
db._no_save = no_save
return db
def _get_db_init_args(self, database):
if isinstance(database, Client):
res = database.get_init_args()
if self.option('store_passwords') and database._pwd:
import base64
password = base64.b64encode(
b'plain:' + database._pwd.encode('utf-8')).decode('utf-8')
res.update({'password': password})
return res
elif isinstance(database, dict):
return database
else: # pragma: no cover
raise ValueError("Bad database instance. "
"It should be dict or Client object")
def save(self):
""" Saves session on disc
"""
databases = {}
for url, database in self._databases.items():
if not getattr(database, '_no_save', False):
init_args = self._get_db_init_args(database)
databases[url] = init_args
data = {
'databases': databases,
'aliases': self._db_aliases,
'options': self._options,
'index': self._db_index_rev,
}
json_write(self.data_file, data, indent=4)
# Overridden to be able to access database like
# session[url_or_index]
def __getitem__(self, url_or_index):
try:
res = self.get_db(url_or_index)
except ValueError as e:
raise KeyError(str(e))
return res
# Overriden to be able to access database like
# session.my_db
def __getattr__(self, name):
try:
res = self.get_db(name)
except ValueError as e:
raise AttributeError(str(e))
return res
def __str__(self):
return pprint.pformat(self.index)
def __dir__(self):
res = super(Session, self).__dir__()
res += self.aliases.keys()
return res