Created
July 23, 2016 20:11
-
-
Save ianjosephwilson/d2ed7d0ef774228feb759f023dfe1917 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
from collections import namedtuple | |
#tryton_jsonrpc is a softlink to tryton.jsonrpc. | |
from tryton_jsonrpc import ServerProxy, Fault | |
from pyramid.httpexceptions import HTTPUnauthorized | |
log = logging.getLogger(__name__) | |
def get_server_proxy(settings): | |
db = settings['laspilitas.tryton_db'] | |
host = settings['laspilitas.tryton_host'] | |
port = settings['laspilitas.tryton_port'] | |
return ServerProxy(host, port, db) | |
def login_to_tryton(request, username, password): | |
server = get_server_proxy(request.registry.settings) | |
return server.common.db.login(username, password) | |
class TrytonServerWrapper(object): | |
""" | |
Wrap the tryton json rpc interface to implicitly pass arguments. | |
""" | |
def __init__(self, target, manager): | |
self.target = target | |
self.manager = manager | |
def __call__(self, *args): | |
""" Include the implicit arguments when we are are finally called. | |
This supports the final call: | |
server.model.inventory.inventory.search(user_id, session_id, *args, | |
context) | |
""" | |
new_args = [self.manager.user_id, self.manager.session_id] | |
new_args.extend(args) | |
new_args.append(self.manager.context) | |
try: | |
return self.target.__call__(*new_args) | |
except Fault, e: | |
if e.faultCode == 'NotLogged': | |
_raise_logout(self.manager.request) | |
else: | |
raise | |
def __getattribute__(self, name): | |
""" Keep wrapping attributes until we are called. | |
This supports attribute lookup: | |
server.model.inventory.* | |
""" | |
if name in ('manager', 'target'): | |
# Do not wrap attrs used by this class. | |
return object.__getattribute__(self, name) | |
else: | |
# Otherwise the attribute on target becomes the new target. | |
new_target = getattr(self.target, name) | |
return TrytonServerWrapper(new_target, self.manager) | |
class TrytonManager(object): | |
def __init__(self, user_id, session_id, context, server, request): | |
self.server = TrytonServerWrapper(server, self) | |
self.user_id = user_id | |
self.session_id = session_id | |
self.context = context | |
self.request = request | |
def _raise_logout(request): | |
if 'user_id' in request.session: | |
request.session.clear() | |
raise HTTPUnauthorized() | |
def get_tryton(request): | |
if request.user and request.user.encrypted_tryton_credentials: | |
if 'tryton_user_id' in request.session and 'tryton_session_id' in request.session: | |
tryton_user_id = request.session['tryton_user_id'] | |
tryton_session_id = request.session['tryton_session_id'] | |
server = get_server_proxy(request.registry.settings) | |
# Populate the context from the server. | |
starting_context = {} | |
try: | |
context = server.model.res.user.get_preferences( | |
tryton_user_id, tryton_session_id, True, starting_context) | |
except Fault, e: | |
if e.faultCode == 'NotLogged': | |
_raise_logout(request) | |
else: | |
raise | |
manager = TrytonManager( | |
tryton_user_id, tryton_session_id, context, server, request) | |
def cleanup(request): | |
del manager.request | |
del manager.server | |
request.add_finished_callback(cleanup) | |
return manager | |
else: | |
# Logout. | |
_raise_logout(request) | |
else: | |
return None |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment