使用SQLAlchemy AsyncCore无法加载表对象的问题求助

同步方式实现代码 #

connect_string = 'db_handle://user:password@db_address:port/database' 
# where db_handle is postgressql+psycopg2
engine = create_engine(connect_string)
table = Table(table, metadata, autoload=True, autoload_with=engine)

作为SQLAlchemy Core用户，尝试异步加载表对象时多次失败，无法执行stmt=select([table.c.col])这类查询。以下是已尝试的四种方案及对应错误：

基础异步配置 #

connect_string = 'db_handle://user:password@db_address:port/database' 
# where db_handle is postgressql+asyncpg
engine = create_async_engine(connect_string, echo=True)
metadata = MetaData()

尝试1 #

#try 1
table = await Table(table, metadata, autoload=True, autoload_with=engine)

错误：sqlalchemy.exc.NoInspectionAvailable: Inspection on an AsyncEngine is currently not supported. Please obtain a connection then use ``conn.run_sync`` to pass a callable where it's possible to call ``inspect`` on the passed connection.

尝试2 #

#try 2
metadata.bind(db_engine_object)
table = await Table(table, metadata, autoload=True)

错误：TypeError: 'NoneType' object is not callable

尝试3 #

#try 3
connection = db_engine_object.connect()
table = await Table(table, metadata, autoload=True, autoload_with=connection)

错误：sqlalchemy.exc.NoInspectionAvailable: Inspection on an AsyncConnection is currently not supported. Please use ``run_sync`` to pass a callable where it's possible to call ``inspect`` on the passed connection.

尝试4 #

#try 4
connection = db_engine_object.connect()
table = await Table(table, metadata, autoload=True, autoload_with=connection.run_sync())

错误：TypeError: run_sync() missing 1 required positional argument: 'fn'

正确的异步加载表方法 #

核心是利用AsyncConnection.run_sync()将异步连接转换为同步上下文，在其中执行表加载逻辑：

from sqlalchemy import MetaData, Table
from sqlalchemy.ext.asyncio import create_async_engine
import asyncio

async def load_and_query_table():
    connect_string = 'postgresql+asyncpg://user:password@db_address:port/database'
    engine = create_async_engine(connect_string, echo=True)
    metadata = MetaData()
    
    async with engine.begin() as conn:
        # 定义同步函数用于加载表结构
        def sync_load_table(sync_conn):
            return Table("your_table_name", metadata, autoload=True, autoload_with=sync_conn)
        
        # 通过run_sync执行同步加载逻辑
        table = await conn.run_sync(sync_load_table)
    
    # 执行异步查询
    stmt = table.select().where(table.c.col == "target_value")
    async with engine.connect() as conn:
        result = await conn.execute(stmt)
        rows = result.fetchall()
        print(rows)

asyncio.run(load_and_query_table())

关键说明 #

• run_sync()会将异步连接转为同步连接，满足表结构自动加载所需的同步inspection要求
• 表加载逻辑必须放在传入run_sync的同步函数中执行
• 加载完成后的表对象可直接用于后续异步查询操作

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