基于类实现SQLAlchemy AsyncSession生成器及FastAPI后台调用报错处理

1. 你定义的get_db_session是异步生成器方法，FastAPI的Depends工具会自动处理生成器迭代，拿到实际的AsyncSession实例注入到路由函数中。但你在后台任务中没有对应工具处理，直接使用生成器对象的话，自然会报没有execute属性的错误。
2. 你代码中手动实例化AsyncSession(self.instance_manger.db_instance.engine)的写法绕过了已经定义好的会话生命周期管理逻辑，包括异常自动回滚、会话自动关闭、async_scoped_session的作用域隔离规则，会导致连接泄漏、事务异常等潜在问题。

不需要修改DBManager的现有封装，只需要调整后台任务中获取会话的写法，手动迭代异步生成器即可：

# 修改background_thread.py中的run_main方法
async def run_main(self):
    self.instance_manger.logger_handler_instance.write_log(__name__, logging.INFO, "Background task started")
    # 手动迭代异步生成器获取会话，自动触发生命周期逻辑
    async for session in self.instance_manger.db_instance.get_db_session():
        results = await CryptoCoinService(CryptoCoinRepository(session)).get_coin()
        print(results)

• 所有场景的数据库操作都统一走DBManager封装的get_db_session获取会话，不要手动实例化AsyncSession，保证事务和连接生命周期统一管理
• 你当前用asyncio.create_task创建的是异步协程任务，跑在FastAPI的主事件循环中，不是系统级后台线程，async_scoped_session配置的scopefunc=current_task可以正常识别不同协程的作用域，会话隔离逻辑生效
• 如果后续需要跑定时循环任务，每次循环执行数据库操作都要重新获取会话，不要跨操作复用同一个会话实例，避免连接超时或泄漏

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