如何让Raw SQL查询返回适配Pydantic Schema的ORM对象？

差异原因 #

• ORM查询：SQLAlchemy的ORM查询会自动把数据库返回的行数据映射成你定义的app.models.Post类实例，这些实例有明确的属性结构，和Pydantic的PostOut Schema字段一一对应，所以能直接适配。
• Raw SQL查询：直接执行原生SQL返回的是纯字段值组成的元组或列表，没有和ORM模型绑定的结构，Pydantic没法识别这些零散值对应的字段，自然适配失败。

让Raw SQL返回ORM对象的方法 #

方法1：用ORM查询包裹原生SQL #

通过from_statement()把原生SQL套进ORM查询流程里，让SQLAlchemy帮你完成映射：

from sqlalchemy import text
from app.models import Post

# 用:text()处理占位符，避免SQL注入
sql = text("SELECT * FROM posts WHERE id = :post_id")
post = db_session.query(Post).from_statement(sql).params(post_id=1).first()
# 返回的post就是Post类实例，直接能用在PostOut Schema里

方法2：手动把元组转成ORM对象 #

如果必须直接用execute()执行原生SQL，可以手动把返回的元组数据转换成ORM实例：

from app.models import Post

# 执行Raw SQL并获取结果
row = db_session.execute("SELECT id, title, content FROM posts WHERE id = %s", (1,)).fetchone()
# 按字段顺序映射成Post对象
post_obj = Post(id=row[0], title=row[1], content=row[2])
# 现在post_obj可以适配PostOut Schema

要是怕字段顺序出错，也可以先拿到字段名再用字典映射：

result = db_session.execute("SELECT * FROM posts WHERE id = %s", (1,))
row = result.fetchone()
column_names = result.keys()
post_obj = Post(**dict(zip(column_names, row)))

方法3：旧版本SQLAlchemy用from_raw_sql #

如果你的SQLAlchemy是1.3及以下版本，直接用from_raw_sql方法：

post = db_session.query(Post).from_raw_sql("SELECT * FROM posts WHERE id = %s", 1).first()

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