使用 FastApi 和 SqlAlchemy 的原始 SQL（获取所有列）答案

【问题标题】：Raw SQL with FastApi and SqlAlchemy (get all columns)使用 FastApi 和 SqlAlchemy 的原始 SQL（获取所有列）
【发布时间】：2021-10-10 23:52:17
【问题描述】：

我有一个简单的 FastApi 端点，它使用 SqlAlchemy 连接到 MySQL 数据库（基于教程：https://fastapi.tiangolo.com/tutorial/sql-databases/）

我使用以下方法创建会话：

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

我创建了依赖：

def get_db():
   db = SessionLocal()
   try:
      yield db
   finally:
      db.close()

在我的路线中，我想执行任意 SQL 语句，但我不确定如何正确处理会话、连接、游标等（包括关闭），我通过艰难的方式了解到对于正确的性能非常重要

@app.get("/get_data")
def get_data(db: Session = Depends(get_db)):
    ???

最终的原因是我的表包含机器学习功能，其列是事先未确定的。如果有一种方法可以使用“所有列”定义 Base 模型也可以，但我也找不到。

【问题讨论】：

标签： sqlalchemy fastapi feature-engineering

【解决方案1】：

我改用https://www.encode.io/databases/ 包解决了这个问题。它在后台处理所有连接/会话等。简化的sn-p：

database = databases.Database(DATABASE_URL)
    
@app.get("/read_db")
async def read_db():
    data = await database.fetch_all("SELECT * FROM USER_TABLE")
    return data

【讨论】：

【解决方案2】：

import pymysql, pandas as pd


engine = create_engine('mysql+pymysql://'+uname+':'+password+'@'+server+':'+port+'/'+db) 
con = engine.connect()
df = pd.read_sql('SELECT schema_name FROM information_schema.schemata', con)
return df

【讨论】：