SQLAlchemy Introspect API - Introspecting models using async connection

introspect_model_async.py

import logging

from sqlalchemy import MetaData, Table
from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker

logger = logging.getLogger(__name__)

meta = MetaData()

engine = create_async_engine(f"mysql+aiomysql://...")
AsyncSession = sessionmaker(engine, expire_on_commit=False, class_=_AsyncSession)


async def get_introspected_model(table_name):
    """
    TODO: Introspected models should be memoized
    """
    async with engine.connect() as conn:
        await conn.run_sync(meta.reflect, only=[table_name])
        return Table(table_name, meta, autoload_with=engine)