Python 3.10+：异步函数中访问含异步逻辑属性的无报错方案

我开发了一个包含ORM的库，同步版本的使用模式如下：

class Model(SqlModel):
    columns = ('id', 'details', 'parent_id')
    parent: RelatedModel

Model.parent = belongs_to(Model, Model, 'parent_id')
...
def example(some_id, some_other_id):
    record = Model.find(some_id)
    if not record.parent:
        record.parent = Model.find(some_other_id)
        record.parent().save()

最近给同步版新增特性：首次读取关联属性时自动加载模型，无需用户提前调用model.parent().reload()，例如可直接写parent_details = Model.find(some_id).parent.details。

之后我复刻了整套SQL模型、查询构建器和ORM以支持异步，异步版本使用模式如下：

class AsyncModel(AsyncSqlModel):
    columns = ('id', 'details', 'parent_id')
    parent: AsyncRelatedModel

AsyncModel.parent = async_belongs_to(AsyncModel, AsyncModel, 'parent_id')
...
async def example(some_id, some_other_id):
    record = await AsyncModel.find(some_id)
    if not record.parent:
        record.parent = await AsyncModel.find(some_other_id)
        await record.parent().save()

但在异步函数中读取关联属性（比如示例里的if not record.parent）时，会触发RuntimeError: asyncio.run() cannot be called from a running event loop。我试过替换成asyncio.get_running_loop().run_until_complete，又出现RuntimeError: This event loop is already running；尝试启动新事件循环也报错RuntimeError: Cannot run the event loop while another loop is running。用nest-asyncio能解决错误，但会产生未关闭事件循环的ResourceWarning，而且这个包已经归档，所以需要更优方案。

相关建议说要把函数改成async，但属性做不到这点。请问有没有无需重构整个库的使用模式、也不依赖废弃包的解决方案？

1. 实现异步属性描述符，让关联属性支持await #

核心思路是把异步关联的属性 getter 改造成返回可await的代理对象，而非直接在同步getter里执行异步操作。用户仅需在访问关联属性时添加await，无需大幅修改原有使用模式。

简化实现代码：

class AsyncBelongsToDescriptor:
    def __init__(self, model, related_model, foreign_key):
        self.model = model
        self.related_model = related_model
        self.foreign_key = foreign_key
        self._cache_key = f"_cached_{foreign_key}"

    def __get__(self, instance, owner):
        if instance is None:
            return self
        
        # 优先返回缓存的关联数据
        if hasattr(instance, self._cache_key):
            return getattr(instance, self._cache_key)
        
        # 返回可await的代理对象，延迟执行异步加载
        return AsyncRelationProxy(instance, self)

class AsyncRelationProxy:
    def __init__(self, instance, descriptor):
        self.instance = instance
        self.descriptor = descriptor

    async def __await__(self):
        # 执行异步加载逻辑
        foreign_id = getattr(self.instance, self.descriptor.foreign_key)
        related_record = await self.descriptor.related_model.find(foreign_id)
        
        # 缓存加载结果，避免重复查询
        setattr(self.instance, self.descriptor._cache_key, related_record)
        return related_record

# 替换原有async_belongs_to工厂函数
def async_belongs_to(model, related_model, foreign_key):
    return AsyncBelongsToDescriptor(model, related_model, foreign_key)

用户使用时仅需对关联属性添加await：

async def example(some_id, some_other_id):
    record = await AsyncModel.find(some_id)
    # 加载关联并访问属性
    parent_details = (await record.parent).details
    # 检查关联是否存在
    if not await record.parent:
        record.parent = await AsyncModel.find(some_other_id)
        await record.parent().save()

2. 利用__getattr__延迟处理异步关联 #

在AsyncSqlModel中重写__getattr__方法，当用户访问未加载的关联属性时，自动返回可await的代理对象。这种方式让用户代码更贴近同步版写法，仅需少量改动。

简化实现代码：

class AsyncSqlModel:
    # 原有核心逻辑...
    _relations = {}  # 存储类定义的关联属性

    def __getattr__(self, name):
        # 判断当前访问的是否是关联属性
        if name in self._relations:
            relation = self._relations[name]
            cache_key = f"_cached_{name}"
            # 已缓存则直接返回
            if hasattr(self, cache_key):
                return getattr(self, cache_key)
            # 未缓存返回可await的代理
            return AsyncRelationProxy(self, relation)
        # 非关联属性按原有逻辑处理
        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")

3. 新增预加载选项（可选优化） #

为查询方法添加预加载参数，允许用户在查询主模型时就指定加载关联，后续访问属性无需await，进一步贴近同步版体验：

# 扩展AsyncModel的find方法
@classmethod
async def find(cls, id_, preload=None):
    record = await cls._query().where("id", id_).first()
    if preload and record:
        for relation_name in preload:
            # 提前加载关联并缓存
            setattr(record, f"_cached_{relation_name}", await getattr(record, relation_name))
    return record

# 用户使用示例
async def example(some_id):
    # 查询时预加载parent关联
    record = await AsyncModel.find(some_id, preload=["parent"])
    # 直接访问关联属性，无需await
    parent_details = record.parent.details

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