如何使用SQLAlchemy (asyncio)创建单个数据表？

错误原因 #

你遇到的TypeError: DDLElement.__call__() missing 1 required positional argument: 'bind'，是因为run_sync方法需要传入可调用函数，但你直接传递了CreateTable(table)这个DDL对象——它不是可调用的函数，而是需要绑定连接执行的DDL元素。

正确实现方式 #

1. 创建单个数据表 #

修正后的代码核心是通过run_sync传入一个接收同步连接的函数，在函数内部执行DDL操作：

from sqlalchemy import Table, Column, BigInteger, Float, MetaData
from sqlalchemy.schema import DropTable, CreateTable
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("你的数据库URL", echo=True)
metadata_obj = MetaData()

async def create_price_timestamp_table(table_name):
    table = Table(table_name, metadata_obj, 
        Column('timestamp', BigInteger, primary_key=True, nullable=False),
        Column('price', Float, nullable=False),
        keep_existing=True,
    )
    async with engine.begin() as conn:
        # 用lambda封装同步执行逻辑，接收同步连接参数
        await conn.run_sync(lambda sync_conn: sync_conn.execute(CreateTable(table)))
    return table

2. 删除单个数据表 #

删除表的逻辑和创建一致，只需替换为DropTable，并可添加if_exists=True避免表不存在时报错：

async def drop_price_timestamp_table(table_name):
    # 加载已存在的表结构，或直接复用之前定义的table对象
    table = Table(table_name, metadata_obj, autoload_with=engine)
    async with engine.begin() as conn:
        await conn.run_sync(lambda sync_conn: sync_conn.execute(DropTable(table, if_exists=True)))

可选：用单独函数替代lambda #

如果觉得lambda不够直观，也可以定义独立的同步函数：

def create_table_sync(sync_conn, table):
    sync_conn.execute(CreateTable(table))

async def create_price_timestamp_table(table_name):
    table = Table(table_name, metadata_obj, 
        Column('timestamp', BigInteger, primary_key=True, nullable=False),
        Column('price', Float, nullable=False),
        keep_existing=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(create_table_sync, table)
    return table

SQLAlchemy 2.0+ 简化写法 #

如果你使用的是SQLAlchemy 2.0及以上版本，异步连接可直接执行DDL元素，无需run_sync：

async def create_price_timestamp_table(table_name):
    table = Table(table_name, metadata_obj, 
        Column('timestamp', BigInteger, primary_key=True, nullable=False),
        Column('price', Float, nullable=False),
        keep_existing=True,
    )
    async with engine.begin() as conn:
        await conn.execute(CreateTable(table))
    return table

内容的提问来源于stack exchange，提问作者John Jethro Rashbold