在事件上创建 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()

如何在我的事件函数中访问 get_db()，以便我可以使用 Session？

我试过：

@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 manual and use a contextmanager 所示，以打开和关闭此会话：

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

Create SQLAlchemy session on event