Created
May 25, 2022 02:14
-
-
Save melvinkcx/c17d937c567f067485c816e8d1a4547a to your computer and use it in GitHub Desktop.
SQLAlchemy Introspect API - Introspecting models using async connection
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment