在事件上创建 SQLAlchemy 会话

Question

如果我想在处理请求时使用数据库，我会进行如下依赖注入：

@app.post("/sample_test")
async def sample_test(db: Session = Depends(get_db)):
    return db.query(models.User.height).all()

但我不能用这样的事件来做到这一点：

@app.on_event("startup")
async def sample_test(db: Session = Depends(get_db)):
    return db.query(models.User.height).all()

因为starlette事件不支持 Depends。

这是我的get_db()功能：

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

就像 FastAPI 手册（ https://fastapi.tiangolo.com/tutorial/sql-databases/ ）中一样。

如何访问get_db()我的事件函数内部，以便我可以使用会话？

我试过了：

@app.on_event("startup")
async def sample_test(db: Session = Depends(get_db)):
    db = next(get_db())
    return db.query(models.User.height).all()

但它不起作用。

如果重要的话我会使用 MSSQL。

Answer 1

您可以导入SessionLocal您创建的依赖项（如FastAPI 手册中所示） ，然后使用上下文管理器打开和关闭此会话，而不是使用依赖项：

@app.on_event("startup")
async def sample_test():
    with SessionLocal() as db:
        return db.query(models.User.height).all()