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

Question

我有一个简单的 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)):
    ???

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

Answer 1

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

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

Answer 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

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

Raw SQL with FastApi and SqlAlchemy (get all columns)

sqlalchemy

fastapi

feature-engineering