使用 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)):
    ???

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

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