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
|
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
import atexit
import os
import readline
import sys
from code import InteractiveConsole
from rlcompleter import Completer
from trytond import __version__
from trytond.pool import Pool
from trytond.transaction import Transaction
from trytond.worker import run_task
class Console(InteractiveConsole):
def __init__(self, locals=None, filename="<console>", histsize=-1,
histfile=os.path.expanduser("~/.trytond_console_history")):
super().__init__(locals, filename)
self.init_completer(locals)
self.init_history(histfile, histsize)
def init_completer(selfi, locals):
completer = Completer(locals)
readline.set_completer(completer.complete)
readline.parse_and_bind("tab: complete")
def init_history(self, histfile, histsize):
readline.parse_and_bind("tab: complete")
if hasattr(readline, 'read_history_file'):
try:
readline.read_history_file(histfile)
except FileNotFoundError:
pass
atexit.register(self.save_history, histfile, histsize)
def save_history(self, histfile, histsize):
readline.set_history_length(histsize)
readline.write_history_file(histfile)
def run(options):
db_name = options.database_name
pool = Pool(db_name)
with Transaction().start(db_name, 0, readonly=True):
pool.init()
with Transaction().start(
db_name, 0, readonly=options.readonly,
_lock_tables=options.lock_tables,
_lock_records=dict(options.lock_records)) as transaction:
local = {
'pool': pool,
'transaction': transaction,
}
if sys.stdin.isatty():
console = Console(local, histsize=options.histsize)
banner = "Tryton %s, Python %s on %s" % (
__version__, sys.version, sys.platform)
console.interact(banner=banner, exitmsg='')
else:
console = InteractiveConsole(local)
console.runcode(sys.stdin.read())
transaction.rollback()
while transaction.tasks:
task_id = transaction.tasks.pop()
run_task(pool, task_id)
|
