如何使用SQLAlchemy Asyncio在运行时创建单个数据表

程序中定义了多个类，仅在类实例化时需要创建对应的数据表，不想用meta.create_all一次性创建所有表。尝试传入单个表列表调用meta.create_all时，报错AttributeError: 'AsyncEngine' object has no attribute '_run_ddl_visitor'。

错误信息 #

AttributeError: 'AsyncEngine' object has no attribute '_run_ddl_visitor'

代码示例 #

import asyncio
import logging
from sqlalchemy import MetaData, Table, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine
from functools import partial
import logging

logger = logging.getLogger()

class DB(): pass
db = DB()

class Reg():
    def __init__(self, name) -> None:
        self.name = name
        self.table = self.getTable()

    def getTable(self) -> Table:
        return Table(
            'items',
            db.meta,
            Column('id', Integer, primary_key=True),
            Column('name', String(30)),
            Column('type', String(15))
        )
    
    async def createTable(self, table:Table):
        async with db.engine.begin() as conn:
            await conn.run_sync(db.meta.create_all)
            # await conn.run_sync(partial(db.meta.create_all,db.engine,[table])) #   <--- 此行报错

async def main():
    logging.basicConfig(level=logging.INFO)

    db.engine = create_async_engine(f'sqlite+aiosqlite:///:memory:', echo=True)
    db.meta = MetaData()

    reg = Reg('Reg')
    await reg.createTable(reg.table)

    await db.engine.dispose()

if __name__ == '__main__':
    asyncio.run(main())

环境 #

python 3.12
sqlalchemy 2.0.35

解决方案 #

报错源于两个问题：

1. run_sync的回调函数会自动传入同步连接对象，不需要手动传递异步引擎db.engine给create_all；
2. create_all的tables参数需要以关键字参数形式传递，不能直接传位置参数。

修改createTable方法为以下两种写法之一即可：

写法1：用lambda传递参数 #

async def createTable(self, table:Table):
    async with db.engine.begin() as conn:
        await conn.run_sync(lambda sync_conn: db.meta.create_all(sync_conn, tables=[table]))

写法2：用partial指定关键字参数 #

async def createTable(self, table:Table):
    async with db.engine.begin() as conn:
        await conn.run_sync(partial(db.meta.create_all, tables=[table]))

修改后即可仅创建指定的单个数据表，不会生成所有未实例化类对应的空表。

内容的提问来源于stack exchange，提问作者iamthebull