如何让普通装饰器兼容异步函数，解决uplink.returns.json异步适配问题

核心问题原因 #

你之前的实现错误在于装饰器执行顺序：把不支持异步的returns.json套在了异步wrapper外层，导致它拿到的仍然是协程对象，完全没有解决核心矛盾。正确逻辑是先获取到函数的实际返回值（异步函数需await拿到结果），再将返回值交给原同步装饰器处理。

通用适配方案 #

可以写一个通用的包装工具，将任意仅支持同步函数的第三方装饰器，转为同时兼容同步/异步函数的版本：

import asyncio
from functools import wraps

def make_async_compatible(sync_decorator):
    def decorator(fn):
        if asyncio.iscoroutinefunction(fn):
            @wraps(fn)
            async def async_wrapper(*args, **kwargs):
                # 先await拿到异步函数的实际返回结果
                raw_result = await fn(*args, **kwargs)
                # 构造临时同步函数，套入原装饰器处理结果
                @sync_decorator
                def temp_sync_func(_):
                    return raw_result
                return temp_sync_func(None)
            return async_wrapper
        else:
            # 同步函数直接套用原装饰器
            return sync_decorator(fn)
    return decorator

针对uplink场景的具体实现 #

# 先将uplink原生的json返回装饰器转为异步兼容版本
async_friendly_json = make_async_compatible(uplink.returns.json(key="token"))

class MyAPI(Consumer):
    def __init__(
        self,
        password: str,
        sync: bool = True,
        auto_auth: bool = True,
        **kwargs
    ):
        self.sync = sync
        self.client = None

        if not self.sync:
            self.client = AiohttpClient()

        super().__init__(client=self.client, **kwargs)

        if auto_auth:
            if self.sync:
                self._login(password)
            else:
                # 注意：异步模式下不能在同步的__init__中直接调用异步方法，需要上层调用者手动await _login，或单独实现异步初始化逻辑
                pass

    # 替换原生装饰器即可
    @async_friendly_json
    @uplink.post("router/login")
    def _login(self, pwd: uplink.Field):
        """获取token"""

额外注意事项 #

如果不想用通用适配方案，也可以直接读uplink源码找到json解析的底层方法，跳过临时函数的构造直接调用解析逻辑，性能会更好。

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