FastApi结合SqlAlchemy执行原生SQL查询未知列数据表咨询

你注入的db: Session已经由get_db依赖自动管理生命周期，路由函数执行完毕后无论成功或异常，都会自动触发db.close()释放资源，无需你手动处理connection、cursor的创建和关闭，SqlAlchemy底层会自动管理这些资源的生命周期，不会出现泄漏问题。

路由内执行原生SQL的示例代码如下：

from sqlalchemy import text

@app.get("/get_data")
def get_data(db: Session = Depends(get_db)):
    # 构造原生SQL，参数使用:name格式占位，避免SQL注入
    query_sql = text("SELECT * FROM ml_feature_table WHERE feature_group = :group_id")
    # 执行SQL并传入参数
    result = db.execute(query_sql, {"group_id": 1})
    # mappings()方法自动将行数据按<列名:字段值>的格式映射
    rows = result.mappings().all()
    # 直接转为字典列表返回，FastAPI会自动完成序列化
    return [dict(row) for row in rows]

不需要定义适配所有字段的Base模型，两种方案均可满足需求：

• 优先选择上述原生SQL+mappings()的方案，无论表有多少字段、字段名是什么，都会自动读取全部列并按列名映射为字典结构，完全不需要提前声明表字段
• 如果需要使用轻量ORM操作，可以用SqlAlchemy的自动反射机制，动态加载表结构，示例如下：

from sqlalchemy import MetaData, Table

# 反射操作仅需执行一次，无需每次请求都调用
metadata = MetaData()
# 动态反射指定表的结构，不需要手动定义字段
ml_feature_table = Table("ml_feature_table", metadata, autoload_with=engine)

@app.get("/get_data_reflect")
def get_data_reflect(db: Session = Depends(get_db)):
    result = db.query(ml_feature_table).filter(ml_feature_table.c.feature_group == 1).all()
    # 同样转为字典列表返回即可
    return [dict(row._asdict()) for row in result]

• 执行INSERT/UPDATE/DELETE等写操作时，执行完SQL后需要手动调用db.commit()提交事务，读操作不需要提交
• 原生SQL参数必须用text()包裹+占位符传参，禁止直接拼接字符串构造SQL，避免SQL注入风险
• 不需要手动获取connection、cursor对象，也无需手动关闭这些资源，使用get_db注入的Session即可保证资源安全释放

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