"""
Interface between the mobile apps, websites,
client systems, partners and the Python modules of the Odoo system.
See :ref:`chapter_architecture-diagram` for more information
on how Twisted is integrated into the core model.
"""
from twisted.internet import ssl, reactor, endpoints
from twisted.web import xmlrpc, server, resource
import os
import configparser
import traceback
# ------- get parameter from the settings file "parameter_global.ini"
# Name and full path of the "ini" file
filename = "parameter_global.ini"
path_name = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(path_name, 'config')
fullpath = os.path.join(config_path, filename)
config = configparser.ConfigParser()
config.read(fullpath)
# get the path to the certificate file
certificate_key = config.get('get_settings_certificate', 'certificate_key')
# get the path to the private key file
private_key = config.get('get_settings_certificate', 'private_key')
[docs]
class custom_functions(xmlrpc.XMLRPC):
"""
The following functions serve the interfaces between the Odoo system and
the mobile apps, websites, client systems and the partners.
"""
def _handle_error(self, func_name, e):
"""
Helper method to format exceptions into detailed XMLRPC Fault responses.
"""
error_type = type(e).__name__
error_msg = str(e)
error_trace = traceback.format_exc()
detailed_msg = f"Error in {func_name}:\n{error_type}: {error_msg}\n\nTraceback:\n{error_trace}"
return xmlrpc.Fault(8002, detailed_msg)
[docs]
def xmlrpc_test_odoo_connection(self):
"""
:func:`odoo_xmlrpc_twisted.functions.test_odoo_connection.test_odoo_connection` tests if the connection to odoo works.
"""
try:
from functions.test_odoo_connection import test_odoo_connection
result = test_odoo_connection()
return result
except Exception as e:
raise self._handle_error('test_odoo_connection', e)
[docs]
def xmlrpc_test_odoo_xmlrpc_server(self):
"""
:func:`odoo_xmlrpc_twisted.functions.test_odoo_xmlrpc_server.test_odoo_xmlrpc_server` tests if the XML-RPC server from Odoo system is running.
"""
try:
from functions.test_odoo_xmlrpc_server import test_odoo_xmlrpc_server
result = test_odoo_xmlrpc_server()
return result
except Exception as e:
raise self._handle_error('test_odoo_xmlrpc_server', e)
# ------ Functions for the product handling -----
[docs]
def xmlrpc_get_product(self, custom_product_id):
"""
:func:`odoo_xmlrpc_twisted.functions.get_product.get_product` gets a specific product with the details.
"""
try:
from functions.get_product import get_product
result = get_product(custom_product_id)
return result
except Exception as e:
raise self._handle_error('get_product', e)
[docs]
def xmlrpc_get_product_list(self, custom_partner_id, custom_place):
"""
:func:`odoo_xmlrpc_twisted.functions.get_product_list.get_product_list` gets all products from a user.
"""
try:
from functions.get_product_list import get_product_list
result = get_product_list(custom_partner_id, custom_place)
return result
except Exception as e:
raise self._handle_error('get_product_list', e)
[docs]
def xmlrpc_count_product_place(self, custom_partner_id, custom_x_place):
"""
:func:`odoo_xmlrpc_twisted.functions.count_product_place.count_product_place` gets the number of products of a partner belonging to a dedicated place.
"""
try:
from functions.count_product_place import count_product_place
result = count_product_place(custom_partner_id, custom_x_place)
return result
except Exception as e:
raise self._handle_error('count_product_place', e)
[docs]
def xmlrpc_get_product_place_list(self, custom_partner_id, custom_x_place):
"""
:func:`odoo_xmlrpc_twisted.functions.get_product_place_list.get_product_place_list` gets a list of all products belonging to the specified user and place.
"""
try:
from functions.get_product_place_list import get_product_place_list
result = get_product_place_list(custom_partner_id, custom_x_place)
return result
except Exception as e:
raise self._handle_error('get_product_place_list', e)
[docs]
def xmlrpc_create_dummy_product(self, custom_partner_id):
"""
:func:`odoo_xmlrpc_twisted.functions.create_dummy_product.create_dummy_product` creates a new initial (= dummy) product in Odoo.
"""
try:
from functions.create_dummy_product import create_dummy_product
result = create_dummy_product(custom_partner_id)
return result
except Exception as e:
raise self._handle_error('create_dummy_product', e)
[docs]
def xmlrpc_create_product(self, custom_partner_id, object_images, object_volume, object_weight):
"""
:func:`odoo_xmlrpc_twisted.functions.create_product.create_product` creates a new product in Odoo.
"""
try:
from functions.create_product import create_product
result = create_product(custom_partner_id, object_images, object_volume, object_weight)
return result
except Exception as e:
raise self._handle_error('create_product', e)
[docs]
def xmlrpc_update_product(self, custom_product_id, custom_name, custom_description):
"""
:func:`odoo_xmlrpc_twisted.functions.update_product.update_product` updates a product in Odoo with the values "Name" and "Description".
"""
try:
from functions.update_product import update_product
result = update_product(custom_product_id, custom_name, custom_description)
return result
except Exception as e:
raise self._handle_error('update_product', e)
[docs]
def xmlrpc_get_picture_string(self, custom_product_id, custom_res_field, custom_name):
"""
:func:`odoo_xmlrpc_twisted.functions.get_picture_string.get_picture_string` gets the picture in form of a data string from a specific product.
"""
try:
from functions.get_picture_string import get_picture_string
result = get_picture_string(custom_product_id, custom_res_field, custom_name)
return result
except Exception as e:
raise self._handle_error('get_picture_string', e)
[docs]
def xmlrpc_save_picture(self, custom_product_id, custom_datas):
"""
:func:`odoo_xmlrpc_twisted.functions.save_picture.save_picture` saves the picture of a product.
"""
try:
from functions.save_picture import save_picture
result = save_picture(custom_product_id, custom_datas)
return result
except Exception as e:
raise self._handle_error('save_picture', e)
[docs]
def xmlrpc_create_order(self, custom_product_id, custom_transfer):
"""
:func:`odoo_xmlrpc_twisted.functions.create_order.create_order` creates the stock move and other records
when an object moves from one place to another.
"""
try:
from functions.create_order import create_order
result = create_order(custom_product_id, custom_transfer)
return result
except Exception as e:
raise self._handle_error('create_order', e)
# ------ Functions for the user management ------
[docs]
def xmlrpc_name_exist(self, custom_name):
"""
:func:`odoo_xmlrpc_twisted.functions.name_exist.name_exist` checks if a user with that name already exists.
"""
try:
from functions.name_exist import name_exist
result = name_exist(custom_name)
return result
except Exception as e:
raise self._handle_error('name_exist', e)
[docs]
def xmlrpc_email_exist(self, custom_email):
"""
:func:`odoo_xmlrpc_twisted.functions.email_exist.email_exist` checks if a user with that e-mail already exists.
"""
try:
from functions.email_exist import email_exist
result = email_exist(custom_email)
return result
except Exception as e:
raise self._handle_error('email_exist', e)
[docs]
def xmlrpc_login_exist(self, custom_login):
"""
:func:`odoo_xmlrpc_twisted.functions.login_exist.login_exist` checks if a user with that login already exists.
"""
try:
from functions.login_exist import login_exist
result = login_exist(custom_login)
return result
except Exception as e:
raise self._handle_error('login_exist', e)
[docs]
def xmlrpc_create_user(self, custom_name, custom_login, custom_email, custom_password):
"""
:func:`odoo_xmlrpc_twisted.functions.create_user.create_user` creates a new user in Odoo.
"""
try:
from functions.create_user import create_user
result = create_user(custom_name, custom_login, custom_email, custom_password)
return result
except Exception as e:
raise self._handle_error('create_user', e)
[docs]
def xmlrpc_create_activatetoken(self, custom_login):
"""
:func:`odoo_xmlrpc_twisted.functions.create_activatetoken.create_activatetoken` sets the user during the user activation to active.
"""
try:
from functions.create_activatetoken import create_activatetoken
result = create_activatetoken(custom_login)
return result
except Exception as e:
raise self._handle_error('create_activatetoken', e)
[docs]
def xmlrpc_test_activatetoken(self, custom_activate_token):
"""
:func:`odoo_xmlrpc_twisted.functions.test_activatetoken.test_activatetoken` tests if a user with the "x_activate_token" does exist.
"""
try:
from functions.test_activatetoken import test_activatetoken
result = test_activatetoken(custom_activate_token)
return result
except Exception as e:
raise self._handle_error('test_activatetoken', e)
[docs]
def xmlrpc_update_activatetoken(self, custom_activate_token):
"""
:func:`odoo_xmlrpc_twisted.functions.update_activatetoken.update_activatetoken` updates the res_users record
so that the token could no longer be used.
"""
try:
from functions.update_activatetoken import update_activatetoken
result = update_activatetoken(custom_activate_token)
return result
except Exception as e:
raise self._handle_error('update_activatetoken', e)
[docs]
def xmlrpc_create_resettoken(self, custom_login):
"""
:func:`odoo_xmlrpc_twisted.functions.create_resettoken.create_resettoken` sets the "activate" fields in the "res.users" table during the activation of a user.
"""
try:
from functions.create_resettoken import create_resettoken
result = create_resettoken(custom_login)
return result
except Exception as e:
raise self._handle_error('create_resettoken', e)
[docs]
def xmlrpc_test_resettoken(self, custom_reset_token):
"""
:func:`odoo_xmlrpc_twisted.functions.test_resettoken.test_resettoken` tests if a user with the "x_reset_token" does exist.
"""
try:
from functions.test_resettoken import test_resettoken
result = test_resettoken(custom_reset_token)
return result
except Exception as e:
raise self._handle_error('test_resettoken', e)
[docs]
def xmlrpc_update_resettoken(self, custom_reset_token):
"""
:func:`odoo_xmlrpc_twisted.functions.update_resettoken.update_resettoken` updates the res_users record
so that the token could no longer be used.
"""
try:
from functions.update_resettoken import update_resettoken
result = update_resettoken(custom_reset_token)
return result
except Exception as e:
raise self._handle_error('update_resettoken', e)
[docs]
def xmlrpc_check_user_login(self, custom_login, custom_password):
"""
:func:`odoo_xmlrpc_twisted.functions.check_user_login.check_user_login` validates if login and password are correct.
"""
try:
from functions.check_user_login import check_user_login
result = check_user_login(custom_login, custom_password)
return result
except Exception as e:
raise self._handle_error('check_user_login', e)
[docs]
def xmlrpc_get_credentials(self, custom_login):
"""
:func:`odoo_xmlrpc_twisted.functions.get_credentials.get_credentials` gets the user credentials of a dedicated user.
"""
try:
from functions.get_credentials import get_credentials
result = get_credentials(custom_login)
return result
except Exception as e:
raise self._handle_error('get_credentials', e)
[docs]
def xmlrpc_change_password(self, custom_login, custom_password):
"""
:func:`odoo_xmlrpc_twisted.functions.change_password.change_password` changes the password of a user.
"""
try:
from functions.change_password import change_password
result = change_password(custom_login, custom_password)
return result
except Exception as e:
raise self._handle_error('change_password', e)
[docs]
def xmlrpc_get_user_profile_values(self, custom_user_id):
"""
:func:`odoo_xmlrpc_twisted.functions.get_user_profile_values.get_user_profile_values` gets the user profile values.
"""
try:
from functions.get_user_profile_values import get_user_profile_values
result = get_user_profile_values(custom_user_id)
return result
except Exception as e:
raise self._handle_error('get_user_profile_values', e)
[docs]
def xmlrpc_get_user_address_values(self, custom_partner_id, custom_country_code, custom_user_language_code):
"""
:func:`odoo_xmlrpc_twisted.functions.get_user_address_values.get_user_address_values` gets the user address values.
"""
try:
from functions.get_user_address_values import get_user_address_values
result = get_user_address_values(custom_partner_id, custom_country_code, custom_user_language_code)
return result
except Exception as e:
raise self._handle_error('get_user_address_values', e)
[docs]
def xmlrpc_update_user_profile_values(self, custom_user_id, custom_login, custom_email, custom_name, custom_title, custom_language, custom_phone_number, custom_mobile_number):
"""
:func:`odoo_xmlrpc_twisted.functions.update_user_profile_values.update_user_profile_values` updates the user profile values.
"""
try:
from functions.update_user_profile_values import update_user_profile_values
result = update_user_profile_values(custom_user_id, custom_login, custom_email, custom_name, custom_title, custom_language, custom_phone_number, custom_mobile_number)
return result
except Exception as e:
raise self._handle_error('update_user_profile_values', e)
[docs]
def xmlrpc_update_user_address_values(self, custom_partner_id, custom_type, custom_street, custom_street_2, custom_zip, custom_city, custom_state, custom_country):
"""
:func:`odoo_xmlrpc_twisted.functions.update_user_address_values.update_user_address_values` updates the user address values.
"""
try:
from functions.update_user_address_values import update_user_address_values
result = update_user_address_values(custom_partner_id, custom_type, custom_street, custom_street_2, custom_zip, custom_city, custom_state, custom_country)
return result
except Exception as e:
raise self._handle_error('update_user_address_values', e)
[docs]
def xmlrpc_get_language(self, custom_partner_id):
"""
:func:`odoo_xmlrpc_twisted.functions.get_language.get_language` gets the language of the partner.
"""
try:
from functions.get_language import get_language
result = get_language(custom_partner_id)
return result
except Exception as e:
raise self._handle_error('get_language', e)
[docs]
def xmlrpc_get_installed_languages(self):
"""
:func:`odoo_xmlrpc_twisted.functions.get_installed_languages.get_installed_languages` gets all installed languages from Odoo.
"""
try:
from functions.get_installed_languages import get_installed_languages
result = get_installed_languages()
return result
except Exception as e:
raise self._handle_error('get_installed_languages', e)
[docs]
def xmlrpc_get_installed_user_titles(self, custom_language_code):
"""
:func:`odoo_xmlrpc_twisted.functions.get_installed_user_titles.get_installed_user_titles` gets all installed user title for a specific language from Odoo.
"""
try:
from functions.get_installed_user_titles import get_installed_user_titles
result = get_installed_user_titles(custom_language_code)
return result
except Exception as e:
raise self._handle_error('get_installed_user_titles', e)
[docs]
def xmlrpc_get_available_countries(self):
"""
:func:`odoo_xmlrpc_twisted.functions.get_available_countries.get_available_countries` gets all available countries from Odoo.
"""
try:
from functions.get_available_countries import get_available_countries
result = get_available_countries()
return result
except Exception as e:
raise self._handle_error('get_available_countries', e)
[docs]
def xmlrpc_get_available_states(self, custom_country, custom_language_code):
"""
:func:`odoo_xmlrpc_twisted.functions.get_available_states.get_available_states` gets all available countries from Odoo.
"""
try:
from functions.get_available_states import get_available_states
result = get_available_states(custom_country, custom_language_code)
return result
except Exception as e:
raise self._handle_error('get_available_states', e)
if __name__ == '__main__':
# create a site and associate the resource handler with it
site = server.Site(custom_functions())
# create an SSL context and load the private and certificate key
ssl_context = ssl.DefaultOpenSSLContextFactory(private_key, certificate_key)
# run the Twisted reactor with the SSL context and site on port 7443
reactor.listenSSL(7443, site, ssl_context)
reactor.run()
"""
# run the Twisted reactor without SSL (for testing purposes only)
r = custom_functions()
endpoint = endpoints.TCP4ServerEndpoint(reactor, 7080)
endpoint.listen(server.Site(r))
reactor.run()
"""