Python-asyncio：将同步模块转为异步并支持双模式调用

一、把pyconnman同步API转为异步调用 #

pyconnman本身基于同步DBus库实现，要适配你的异步代码栈，有两种实用方案可选：

方案1：用asyncio线程池快速包装（低成本） #

不需要修改pyconnman源码，直接把同步调用放到线程池里，用asyncio.to_thread（Python3.9+）或loop.run_in_executor异步等待结果。这种方式适合快速适配，缺点是本质是同步逻辑跑在后台线程，不是纯异步DBus调用，但大多数场景下足够满足需求。

示例代码：

import asyncio
from pyconnman import ConnMan

class AsyncConnMan:
    def __init__(self):
        self.connman = ConnMan()

    # 包装同步的get_services方法为异步
    async def get_services(self):
        return await asyncio.to_thread(self.connman.get_services)

    # 同理包装其他核心方法
    async def get_technologies(self):
        return await asyncio.to_thread(self.connman.get_technologies)

    async def connect_service(self, service_path):
        return await asyncio.to_thread(self.connman.connect_service, service_path)

方案2：基于异步DBus库重写（纯异步） #

如果追求真正的无阻塞异步，可以用支持asyncio的DBus库（比如dbus-next），参考pyconnman的逻辑重新实现ConnMan的DBus接口封装。这种方式没有线程开销，但需要重新编写pyconnman的核心逻辑，工作量更大。

示例思路（简化版）：

from dbus_next.aio import MessageBus
from dbus_next import BusType

class AsyncConnMan:
    def __init__(self):
        self.bus = None
        self.manager = None

    async def connect(self):
        self.bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
        # 获取ConnMan的Manager接口
        manager_proxy = await self.bus.get_proxy(
            'net.connman', '/', 'net.connman.Manager'
        )
        self.manager = manager_proxy

    async def get_services(self):
        if not self.manager:
            await self.connect()
        # 异步调用DBus方法
        return await self.manager.call_get_services()

二、同时支持同步和异步的库设计模式 #

推荐两种成熟的设计模式，可根据你的需求选择：

模式1：分离同步/异步类 + 工厂模式 #

这是最清晰、易维护的方案：分别实现同步和异步的核心类，再用一个工厂函数根据参数返回对应实例。

示例代码：

# 定义同步基类（或接口）
class BaseConnector:
    def my_method(self):
        raise NotImplementedError

# 定义异步基类（或接口）
class AsyncBaseConnector:
    async def my_method(self):
        raise NotImplementedError

# 同步实现类
class SyncConnector(BaseConnector):
    def my_method(self):
        # 同步业务逻辑
        return "同步调用结果"

# 异步实现类
class AsyncConnector(AsyncBaseConnector):
    async def my_method(self):
        # 异步业务逻辑
        return "异步调用结果"

# 工厂函数，根据参数返回对应实例
def create_connector(async_mode: bool = False):
    if async_mode:
        return AsyncConnector()
    return SyncConnector()

# 使用方式
sync_conn = create_connector(async_mode=False)
print(sync_conn.my_method())

async def run_async():
    async_conn = create_connector(async_mode=True)
    result = await async_conn.my_method()
    print(result)

asyncio.run(run_async())

模式2：组合模式（同一类支持两种模式） #

如果希望对外暴露同一个类，通过初始化参数切换同步/异步模式，可以用组合方式：内部持有同步或异步的实现实例，对外提供统一的调用入口。

示例代码：

class Connector:
    def __init__(self, async_mode: bool = False):
        if async_mode:
            self._impl = AsyncConnector()
        else:
            self._impl = SyncConnector()
        self._is_async = async_mode

    def my_method(self):
        if self._is_async:
            # 异步模式下返回协程对象，需要调用方用await
            return self._impl.my_method()
        return self._impl.my_method()

# 使用方式
sync_conn = Connector(async_mode=False)
print(sync_conn.my_method())

async def run_async():
    async_conn = Connector(async_mode=True)
    result = await async_conn.my_method()
    print(result)

asyncio.run(run_async())

三、实例化时选择同步/异步版本的派生类 #

如果想让同一个派生类在实例化时切换同步/异步父类，类工厂函数是最实用的方案：动态生成派生类，根据参数选择继承同步或异步基类。

示例代码：

# 你原有的同步和异步基类
class myExampleConnectorClass:
    def my_method(self):
        return "基类同步方法"

class myExampleAsyncConnectorClass:
    async def my_method(self):
        return "基类异步方法"

# 类工厂函数，动态生成派生类
def create_my_custom_obj(async_mode: bool = False):
    # 根据模式选择父类
    parent_cls = myExampleAsyncConnectorClass if async_mode else myExampleConnectorClass

    # 定义派生类
    class myCustomObj(parent_cls):
        def my_method(self):
            if async_mode:
                # 异步方法需要重新定义为async def
                async def wrapped():
                    return await super().my_method()
                return wrapped()
            else:
                return super().my_method()

    # 返回实例
    return myCustomObj()

# 使用示例
sync_obj = create_my_custom_obj(async_mode=False)
print(sync_obj.my_method())  # 输出：基类同步方法

async def run_async():
    async_obj = create_my_custom_obj(async_mode=True)
    result = await async_obj.my_method()
    print(result)  # 输出：基类异步方法

asyncio.run(run_async())

这种方式既保留了派生类的逻辑，又能灵活切换同步/异步模式，同时避免了复杂的继承层级问题。

内容的提问来源于stack exchange，提问作者Nicola Lunghi