[SOLVED] How do I create a model instance using raw SQL in async SQLAlchemy?

How do I create a model instance using raw SQL in async SQLAlchemy?

I have asyncio sqlalchemy code:

import asyncio
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import text, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(f'mysql+aiomysql://root:example@127.0.0.1:3306/spy-bot')
AsyncSession = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(255))

async def main():
    async with AsyncSession() as session:
        stmt = text('SELECT * FROM `users` LIMIT 1')
        result = await session.execute(stmt)
        user = result.one()
        print(type(user), user)

asyncio.run(main())

how do I make session query return instances on User class while still using raw sql?

On sync version this would look like

result = session.query(User).from_statement(text(query))

Solution

You would do it in the same way, albeit using 2.0-style syntax as the legacy Query class is not supported in asyncio:

result = await session.scalars(select(User).from_statement(text_object))

The docs at Getting ORM Results from Textual Statements apply.

The complete function might look like this:

import sqlalchemy as sa
...
async def main():
    async with AsyncSession() as session:
        stmt = text('SELECT * FROM `users` LIMIT 1')
        result = await session.scalars(sa.select(User).from_statement(stmt))
        user = result.one()
        print(type(user), user)
    await async_engine.dispose()