SQLAlchemy异步ORM使用async engine无法获取数据库表问题求解

首先修正基础代码错误 #

你现有代码存在两个核心错误：

1. 会话工厂初始化错误：sessionmaker返回的是AsyncSession的工厂对象，你错误地直接加()实例化后存入self._session，导致后续调用self._session()时触发「AsyncSession对象不可调用」的报错
2. run_sync用法错误：run_sync要求传入可调用对象，会自动将同步连接作为第一个参数传入该对象，不能直接对异步连接执行inspect操作

修正后的PostgresDb类 #

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import inspect, MetaData

class PostgresDb:
    def __init__(self):
        self.session_factory = None
        self._engine = None

    def __getattr__(self, name):
        return getattr(self.session_factory, name)

    def init(self):
        # 初始化只执行一次，不要每次调用方法都执行
        self._engine = create_async_engine(
            ENGINE,
            echo=True,
            future=True
        )
        # 存会话工厂，不要直接实例化
        self.session_factory = sessionmaker(
            self._engine, 
            expire_on_commit=False, 
            class_=AsyncSession
        )

    async def create_all(self):
        async with self._engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

实现异步获取所有表的方法 #

提供两种常用实现，按需选择：

方式1：仅获取表名列表 #

async def get_table_names(self, schema=SCHEMA):
    def sync_ops(sync_conn):
        inspector = inspect(sync_conn)
        # 同时获取表和视图可以加views=True参数
        return inspector.get_table_names(schema=schema)
    
    async with self._engine.begin() as conn:
        table_names = await conn.run_sync(sync_ops)
        return table_names

方式2：获取完整的表对象字典 #

async def get_all_tables(self, schema=SCHEMA):
    def sync_ops(sync_conn):
        meta = MetaData(schema=schema)
        meta.reflect(bind=sync_conn, views=True)
        return meta.tables
    
    async with self._engine.begin() as conn:
        table_dict = await conn.run_sync(sync_ops)
        return table_dict

调用示例 #

协程方法必须被await执行，直接调用会触发「coroutine was never awaited」的警告：

import asyncio

async def main():
    db = PostgresDb()
    db.init() # 全局只执行一次
    # 获取表名
    tables = await db.get_table_names()
    print(tables)
    # 获取表对象
    table_objs = await db.get_all_tables()
    print(table_objs.keys())

asyncio.run(main())

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