如何在Python包的多个函数中复用xmlrpc.client.ServerProxy连接对象(基于Odoo XML-RPC API场景)
如何在Python包的多个函数中复用xmlrpc.client.ServerProxy连接对象(基于Odoo XML-RPC API场景)
嘿,我懂你现在的需求——在Python包的多个函数里复用同一个Odoo XML-RPC的ServerProxy连接,不用每次调用函数都重新建立连接,对吧?先看看你当前的连接代码,我帮你理清楚基础情况:
import xmlrpc.client import pprint # Demo connection to Odoo info = xmlrpc.client.ServerProxy('https://demo.odoo.com/start').start() url, db, username, password = info['host'], info['database'], info['user'], info['password']
接下来给你几种开发者常用的实用方案,你可以根据自己项目的复杂度来选:
方法一:模块级全局变量(简单直接,适合小型项目)
这是最省事的办法,在你的包里单独整个连接模块,提前初始化好连接对象,其他函数直接导入用就行。比如你可以建个odoo_connection.py文件:
# odoo_connection.py import xmlrpc.client # 拉取Odoo演示环境的连接信息 info = xmlrpc.client.ServerProxy('https://demo.odoo.com/start').start() url, db, username, password = info['host'], info['database'], info['user'], info['password'] # 初始化两个核心的Proxy对象,模块级变量,整个包都能直接引用 common_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/common') object_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object') # 可选:提前验证连接,确保一开始就是有效的 uid = common_proxy.authenticate(db, username, password, {})
然后在你的业务函数模块里,直接导入这些变量使用即可:
# 比如业务操作模块 odoo_operations.py from .odoo_connection import db, uid, password, object_proxy def get_company_partner_count(): # 直接用已有的object_proxy,无需重复创建连接 return object_proxy.execute_kw(db, uid, password, 'res.partner', 'search_count', [[['is_company', '=', True]]]) def create_new_partner(partner_info): return object_proxy.execute_kw(db, uid, password, 'res.partner', 'create', [partner_info])
⚠️ 小提醒:这种方式如果遇到Odoo服务重启或者连接断开,全局对象不会自动重连,你可以加个简单的检查逻辑,比如调用前先验证连接有效性。
方法二:类封装连接(优雅灵活,适合中大型项目)
如果你的包功能较多,用类来封装连接和相关操作会更清晰,还能方便地添加连接管理逻辑,比如自动重连。可以写个OdooClient类:
# odoo_client.py import xmlrpc.client class OdooClient: def __init__(self): # 初始化时直接建立连接 self._setup_connection() def _setup_connection(self): # 私有方法,专门处理连接初始化逻辑 info = xmlrpc.client.ServerProxy('https://demo.odoo.com/start').start() self.url = info['host'] self.db = info['database'] self.username = info['user'] self.password = info['password'] self.common_proxy = xmlrpc.client.ServerProxy(f'{self.url}/xmlrpc/2/common') self.object_proxy = xmlrpc.client.ServerProxy(f'{self.url}/xmlrpc/2/object') self.uid = self.common_proxy.authenticate(self.db, self.username, self.password, {}) def _validate_connection(self): # 检查连接是否有效,失效则重新初始化 try: self.common_proxy.authenticate(self.db, self.username, self.password, {}) except Exception: self._setup_connection() # 具体业务方法 def get_company_partner_count(self): self._validate_connection() # 调用前先检查连接状态 return self.object_proxy.execute_kw(self.db, self.uid, self.password, 'res.partner', 'search_count', [[['is_company', '=', True]]]) def create_new_partner(self, partner_info): self._validate_connection() return self.object_proxy.execute_kw(self.db, self.uid, self.password, 'res.partner', 'create', [partner_info])
然后在其他模块里,只需要实例化一次这个类,所有函数共享这个实例即可:
# 业务模块里 from .odoo_client import OdooClient # 全局实例化一次,整个包共用这个对象 odoo_client = OdooClient() def print_partner_count(): count = odoo_client.get_company_partner_count() print(f"当前系统中的公司客户数量:{count}")
这种方式把所有和Odoo交互的逻辑都封装在类里,维护起来特别方便,还能轻松扩展功能,比如添加日志、重试机制等。
方法三:依赖注入(低耦合,适合需要单元测试的场景)
如果你想让函数更灵活,方便编写单元测试,那就用依赖注入的方式——把连接对象作为参数传给函数,函数本身不关心连接是如何创建的:
# 业务函数模块 operations.py def get_company_partner_count(object_proxy, db, uid, password): return object_proxy.execute_kw(db, uid, password, 'res.partner', 'search_count', [[['is_company', '=', True]]]) def create_new_partner(object_proxy, db, uid, password, partner_info): return object_proxy.execute_kw(db, uid, password, 'res.partner', 'create', [partner_info])
然后在调用这些函数的地方,先初始化好连接,再把需要的参数传进去:
# 初始化连接的入口文件 main.py import xmlrpc.client from .operations import get_company_partner_count, create_new_partner # 先建立连接 info = xmlrpc.client.ServerProxy('https://demo.odoo.com/start').start() url, db, username, password = info['host'], info['database'], info['user'], info['password'] object_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object') uid = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/common').authenticate(db, username, password, {}) # 调用函数时传入连接相关参数 count = get_company_partner_count(object_proxy, db, uid, password) new_partner_id = create_new_partner(object_proxy, db, uid, password, {'name': '新测试公司'})
这种方式的好处是函数和连接逻辑完全解耦,测试时可以很容易用mock对象代替真实的ServerProxy,无需连接真实的Odoo服务器。
备注:内容来源于stack exchange,提问作者Marco Di Gennaro
相关产品推荐
相关产品推荐

