You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.13 17:14:53