0

我有一个使用 asyncpg 连接数据库的异步 python 应用程序，我想避免手动输入每个查询。我发现 sqlalchemy 核心对此非常有帮助，但我在一件事上苦苦挣扎。我希望我的应用程序能够在启动时创建数据库模式，而不是依赖于已经创建的数据库模式，而且我还没有找到这样做的自动方法。我想要的是这样的：

import asyncio

from sqlalchemy import Table, Column, Integer, String, MetaData
import asyncpg

metadata = MetaData()

people = Table(
    "people",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String)
)

class Database:
    def __init__(self, pool: asyncpg.pool.Pool):
        self.pool = pool

    async def ensure_schema(self):
        async with self.pool.acquire() as con:
            await con.execute(
                # this method doesn't really exist,
                # I'm just looking for something that would 
                # do the same - return following query:
                # CREATE TABLE IF NOT EXISTS people(id INTEGER PRIMARY KEY, name VARCHAR)
                people.get_create_query(checkfirst=True)
            )

async def main()
    pool = await asyncpg.create_pool(host="127.0.0.1", port=5432, user="postgres", password="postgres", database="test")
    db = Database(pool)
    await db.ensure_schema()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

这里的问题是，sqlalchemy 期望metadata绑定到一个引擎，但它不支持异步引擎。所以我只想从声明的表中获取一个 CREATE 查询字符串并将其传递给我的连接以执行。

与此类似的可能是people.select()它不会立即执行，它只是创建一个对象，该对象也有一个字符串表示，即要执行的查询。

我可以将 sql 模式放在一个单独的文件中并在启动时加载它，但是每次更改模式时感觉都不对，我需要在代码中的两个位置进行更改。有什么建议么？

4

2 回答 2

1

实现 sqlalchemy 可以打印出它在设置时执行的所有语句create_engine("connection-string", echo=True)，并在创建新表时打印出 people

CREATE TABLE people (
    id INTEGER NOT NULL,
    name VARCHAR,
    PRIMARY KEY (id)
)

我运行调试器来找出这个字符串来自哪里，结果证明它可以这样获得

>>> from sqlalchemy.sql.ddl import CreateTable
>>> print(CreateTable(people))

我无法在文档中找到它，所以我至少把它留在这里。

于 2019-11-30T11:35:02.453 回答
0

SQLAlchemy 创建表

self.Base.metadata.create_all(engine, tables=[people.__table__])
于 2019-11-29T10:28:01.030 回答