sqlalchemy.orm.exc.UnmappedInstanceError错误原因及修复方案求助

错误原因分析 #

1. response_to_sql函数逻辑缺陷 #

• 错误遍历ORM模型类table（for t in table），模型类并非可迭代对象，该循环无意义且会引发隐性错误
• 匹配到交易所表名后，仅创建第一个交易对实例就return，丢弃其余数据；若未匹配到表名，函数无返回值，默认返回None，导致session.add_all()传入None，触发SQLAlchemy未映射实例错误
• 使用全局变量data存储数据，多个异步任务同时修改会导致数据覆盖，仅最后一个交易所的数据被保留

2. SQLAlchemy会话使用错误 #

• 使用同步Session却调用异步的await session.close()，同步会话不支持异步关闭方法；同时全局会话在异步环境下存在线程安全问题

3. 客户端会话未正确关闭 #

• asyncio.gather设置return_exceptions=True，若某交易所请求抛出异常，可能跳过await client.close()执行；部分交易所（如bybit）需要严格的资源释放流程

修复方案 #

核心修正点 #

1. 移除全局变量：将交易所数据作为参数传递，避免异步任务间的数据覆盖
2. 重构response_to_sql：正确创建所有交易对实例，确保返回值为实例列表（无数据时返回空列表）
3. 改用异步会话：使用SQLAlchemy异步引擎和会话，适配异步运行环境
4. 可靠关闭客户端：用try/finally包裹客户端操作，保证无论是否报错都能释放资源

修正后的完整代码 #

import asyncio
import time
import ccxta
from sqlalchemy.ext.asyncio import async_create_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker

# 基础ORM模型（需根据实际数据库表结构补充字段）
Base = declarative_base()

class Binance(Base):
    __tablename__ = 'binance'
    # 示例字段，需自行补充id、symbol、ask、bid等定义

class Bitget(Base):
    __tablename__ = 'bitget'
    # 字段定义

class Bitmart(Base):
    __tablename__ = 'bitmart'
    # 字段定义

class Bitvavo(Base):
    __tablename__ = 'bitvavo'
    # 字段定义

class Bybit(Base):
    __tablename__ = 'bybit'
    # 字段定义

class Gate(Base):
    __tablename__ = 'gate'
    # 字段定义

class Huobi(Base):
    __tablename__ = 'huobi'
    # 字段定义

class Kucoin(Base):
    __tablename__ = 'kucoin'
    # 字段定义

class Mexc(Base):
    __tablename__ = 'mexc'
    # 字段定义

class Okx(Base):
    __tablename__ = 'okx'
    # 字段定义

# 创建异步数据库引擎与会话工厂
engine = async_create_engine('mysql+aiomysql://root:root@127.0.0.1/exchanges')
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

async def response_to_sql(table, exchange_data):
    instances = []
    if table.__tablename__ in exchange_data:
        tickers = exchange_data[table.__tablename__]
        for symbol, ticker in tickers.items():
            # 容错处理：跳过字段缺失的交易对
            if 'ask' in ticker and 'bid' in ticker:
                ins = table(
                    symbol=symbol,
                    ask=ticker['ask'],
                    bid=ticker['bid']
                )
                instances.append(ins)
    return instances

async def handler(tables, exchange_data):
    async with AsyncSessionLocal() as session:
        all_instances = []
        for table in tables:
            instances = await response_to_sql(table, exchange_data)
            all_instances.extend(instances)
        
        if all_instances:
            session.add_all(all_instances)
            await session.commit()

async def async_client(exchange):
    client = getattr(ccxta, exchange)()
    try:
        tickers = await client.fetch_tickers()
        return {exchange: tickers}
    finally:
        # 确保客户端资源必被释放
        await client.close()

async def multi_tickers(exchanges, tables):
    input_coroutines = [async_client(exchange) for exchange in exchanges]
    exchange_data_list = await asyncio.gather(*input_coroutines)
    
    # 合并所有交易所数据
    exchange_data = {}
    for data in exchange_data_list:
        exchange_data.update(data)
    
    print("async call spend:", time.time() - tic)
    await handler(tables, exchange_data)
    return exchange_data

if __name__ == '__main__':
    exchanges = ['binance', 'bitget', 'bitmart', 'bitvavo', 'bybit', 'gate', 'huobi', 'kucoin', 'mexc', 'okx']
    tables = [Binance, Bitget, Bitmart, Bitvavo, Bybit, Gate, Huobi, Kucoin, Mexc, Okx]
    tic = time.time()
    asyncio.run(multi_tickers(exchanges, tables))

额外注意事项 #

• 需安装异步MySQL驱动：pip install aiomysql
• 需根据实际数据库表结构补充ORM模型的字段定义
• 若需容错单个交易所的请求错误，可在async_client内部添加异常捕获，返回空字典即可

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