Jon Hayden's questions

我已经研究了一段时间，但没有任何收获。如果我从命令行运行“alembic upgrade head”，效果很好。我想在 FastAPI 中执行此操作，这样当它启动时，它会升级数据库架构。所以经过大量搜索，我想出了这个方法：

async def run_migrations():
  alembic_cfg = Config("alembic.ini")
  url = str(engine.url)
  async with engine.begin() as conn:
    await conn.run_sync(alembic_cfg.set_main_option, "sqlalchemy.url", url)
    await conn.run_sync(command.upgrade, alembic_cfg, "head")

我在 FastAPI 生命周期启动期间调用它。它给出了这个错误：

TypeError: Config.set_main_option() takes 3 positional arguments but 4 were given

我也尝试过：

await conn.run_sync(alembic_cfg.set_main_option("sqlalchemy.url", url))

但这让我：

TypeError: 'NoneType' object is not callable

run_sync 不喜欢我这样做吗？我没有向 set_main_option 传递 4 个参数。url 的值是正确的，因为我将其打印出来以确保无误。

或者有没有更简单/更好的方法从 fastapi 内部升级到 head？

谢谢

• SQLAlchemy 2.0.32
• uvicorn 0.30.6
• fastapi 0.112.1
• alembic 1.13.2

我正在尝试在 Async Sqlalchemy 2.0 中构建动态查询。选项以列表形式显示，如下所示：

select_options=[selectinload(Table.entitity1).selectinload(Table2.entity2)]

然后我这样做：

query = select(model).filter(model.id == item_id).options(select_options[0])
result = await async_session.execute(query)

效果很好，我得到了正确的响应。但我希望能够继续将选项附加到该语句。我尝试了这个作为示例，并认为如果这有效，那么我可以循环遍历 select_options 并添加每个选项：

query = select(model)
query.filter(model.id == item_id)
query.options(select_options[0])

我没有收到任何错误，但响应忽略了选项。我如何才能继续添加 .options().options() 等？

谢谢