如何在异步SQLAlchemy中用原生SQL创建并返回模型类实例？

问题背景 #

用户现有异步SQLAlchemy代码，希望在使用原生SQL的前提下，让查询返回User类实例（同步版本可通过session.query(User).from_statement(text(query))实现），同时想了解如何用原生SQL创建模型实例。

原异步代码如下：

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())

一、让原生SQL查询返回User类实例 #

异步SQLAlchemy中，可通过select(User).from_statement()方法实现，替代同步版本的query(User).from_statement()。具体修改如下：

async def main():
    async with AsyncSession() as session:
        # 定义原生SQL语句
        raw_sql = text('SELECT * FROM `users` LIMIT 1')
        # 关联User模型，执行查询
        result = await session.execute(select(User).from_statement(raw_sql))
        # 获取单个实例（对应同步的one()）
        user = result.scalar_one()
        print(type(user), user)  # 输出类型为 <class '__main__.User'>

补充说明 #

• 如果原生SQL只查询部分字段，需要给text语句明确指定对应模型列，避免字段映射错误：

  raw_sql = text('SELECT id, username FROM `users` LIMIT 1').columns(User.id, User.username)
  result = await session.execute(select(User).from_statement(raw_sql))
  

• 若需返回多个实例，使用result.scalars().all()替代scalar_one()。

二、用原生SQL创建模型实例 #

有两种常用方式，根据需求选择：

方式1：执行INSERT后查询返回实例 #

先执行原生INSERT语句提交，再通过查询获取对应的User实例：

async def create_user(session, username):
    # 原生INSERT语句
    insert_sql = text("INSERT INTO users (username) VALUES (:username)")
    await session.execute(insert_sql, {"username": username})
    await session.commit()
    
    # 查询刚创建的用户实例
    query_result = await session.execute(select(User).where(User.username == username))
    return query_result.scalar_one()

# 在main中调用
async def main():
    async with AsyncSession() as session:
        new_user = await create_user(session, "test_user")
        print(type(new_user), new_user)

方式2：获取自增主键后手动实例化 #

利用数据库的自增主键返回功能（如MySQL的1073200），拿到主键后直接创建User实例并关联到session：

async def create_user(session, username):
    # 执行INSERT并返回自增ID
    insert_sql = text("INSERT INTO users (username) VALUES (:username); SELECT 1073200 as id")
    result = await session.execute(insert_sql, {"username": username})
    user_id = result.scalar()
    await session.commit()
    
    # 创建User实例并添加到session，使其处于ORM管理状态
    new_user = User(id=user_id, username=username)
    session.add(new_user)
    return new_user

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