SQLAlchemy Runtime Inspection API doesn't support `AsyncEngine`

we can likely add this at some point and it can be sqlalchemy.ext.asyncio.reflection. it would cover the Inspector API using proxies and for per-dialect inspectors would use a __getattr__ scheme.

for now using run_sync() is the workaround.

Hello @whg517,

As @zzzeek mentioned above the best solution would be to use the run_sync() method. Consider the following example:

@pytest.mark.asyncio
async def test_migrate(session):
    """
    Test migrate.
    :param session:
    :return:
    """
    def get_table_names(conn):
        inspector = inspect(conn)
        return inspector.get_table_names()

    async with session.bind.connect() as connection:
        table_names = await connection.run_sync(get_table_names)
        assert table_names
        assert 'audit' in table_names

Learn more about the run_sync method here: https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.AsyncConnection.run_sync

Update:

I got the workaround…

engine = create_async_engine(url)

def load_table(conn):
        metadata = MetaData(bind=conn)
        metadata.reflect(only=["harshit_table"])
        harshit_table = Table("harshit_table", metadata, autoload_with=engine)
        return harshit_table

async def async_main(self):
        async with engine.connect() as conn:
            table = await conn.run_sync(load_table)
            print("table: ", table, type(table))

or you can use it like this

metadata = MetaData()

async def async_main(self):
        async with engine.connect() as conn:
            await conn.run_sync(metadata.reflect, only=["harshit_table"])
            harshit_table = Table("harshit_table", metadata, autoload_with=engine)
            print("tables: ", harshit_table, type(harshit_table))

Don’t create metadata and use it like.

engine = create_async_engine(url)
metadata = MetaData(bind=engine)

my_table = Table("harshit_table", metadata, autoload_with=engine)

This won’t work.