11

我正在查看使用 sqlalchemy 的 aiopg 使用示例，这些行让我害怕：

async def create_table(conn):
    await conn.execute('DROP TABLE IF EXISTS tbl')
    await conn.execute(CreateTable(tbl))

我不想在使用 sqlalchemy 时执行原始 sql 查询。但是我找不到任何其他方法来实现相同的逻辑。我的尝试是：

1)

await conn.execute(tbl.drop(checkfirst=True))

这提出了：

sqlalchemy.exc.UnboundExecutionError：表对象“tbl”未绑定到引擎或连接。如果没有要执行的数据库，则无法继续执行。

我也找不到将表绑定到引擎的方法，因为aiopg 不支持 metadata.create_all

2)

await conn.execute(DropTable(tbl))

这提出了：

psycopg2.errors.UndefinedTable：表“tbl”不存在

似乎DropTable构造不IF EXISTS以任何方式支持部分。

So, the question is, is there any way to rewrite await conn.execute('DROP TABLE IF EXISTS tbl') statement into something without raw sql when using aiopg + sqlalchemy?

4

1 回答 1

1

This question was posted when the latest version was SQLAlchemy 1.3.11.

As of SQLAlchemy 1.4.0, DropTable supports if_exists=True.

await conn.execute(DropTable(tbl, if_exists=True))

Reference: https://docs.sqlalchemy.org/en/14/core/ddl.html#sqlalchemy.schema.DropTable

于 2021-10-16T14:02:09.037 回答