使用SQLAlchemy Core操作PostgreSQL时execute出现CompileError报错求助

问题场景 #

我正在学习使用Python的SQLAlchemy Core操作PostgreSQL数据库，运行以下CRUD脚本时出现编译错误：

原代码 #

from sqlalchemy import create_engine  
from sqlalchemy import Table, MetaData, String

engine = create_engine('postgresql://postgres:123456@localhost:5432/red30')

with engine.connect() as connection:
    meta = MetaData(engine)  
    sales_table = Table('sales', meta)

    # Create
    insert_statement = sales_table.insert().values(order_num=1105911, 
                                                order_type='Retail', 
                                                cust_name='Syman Mapstone', 
                                                prod_number='EB521', 
                                                prod_name='Understanding Artificial Intelligence', 
                                                quantity=3, 
                                                price=19.5, 
                                                discount=0, 
                                                order_total=58.5)
    connection.execute(insert_statement)

    # Read
    select_statement = sales_table.select().limit(10)
    result_set = connection.execute(select_statement)
    for r in result_set:
        print(r)

    # Update
    update_statement = sales_table.update().where(sales_table.c.order_num==1105910).values(quantity=2, order_total=39)
    connection.execute(update_statement)

    # Confirm Update: Read
    reselect_statement = sales_table.select().where(sales_table.c.order_num==1105910)
    updated_set = connection.execute(reselect_statement)
    for u in updated_set:
        print(u)

    # Delete
    delete_statement = sales_table.delete().where(sales_table.c.order_num==1105910)
    connection.execute(delete_statement)

    # Confirm Delete: Read
    not_found_set = connection.execute(reselect_statement)
    print(not_found_set.rowcount)

错误信息 #

(postgres-prac) E:\xfile\postgresql\postgres-prac>python postgres-sqlalchemy-core.py
Traceback (most recent call last):
  File "postgres-sqlalchemy-core.py", line 20, in <module>
    connection.execute(insert_statement)
  ...
sqlalchemy.exc.CompileError: Unconsumed column names: order_type, quantity, cust_name, discount, prod_number, price, order_total, order_num, prod_name

错误原因 #

核心问题是SQLAlchemy不知道sales表的列定义。你仅通过Table('sales', meta)声明了表名，但没有加载任何列信息，导致执行插入、更新等操作时，无法识别你传入的列名，从而抛出编译错误。

另外，原代码还有一个隐藏问题：with engine.connect()的上下文管理器默认不会自动提交事务，所以你的增删改操作执行后不会实际写入数据库，需要手动调用connection.commit()。

修复方案 #

方案1：反射已存在的表结构（适用于数据库中已创建sales表的情况） #

通过autoload_with=engine参数让SQLAlchemy自动从数据库读取sales表的列结构：

from sqlalchemy import create_engine  
from sqlalchemy import Table, MetaData

engine = create_engine('postgresql://postgres:123456@localhost:5432/red30')

with engine.connect() as connection:
    meta = MetaData()  
    # 自动反射数据库中已有的sales表结构
    sales_table = Table('sales', meta, autoload_with=engine)

    # Create
    insert_statement = sales_table.insert().values(order_num=1105911, 
                                                order_type='Retail', 
                                                cust_name='Syman Mapstone', 
                                                prod_number='EB521', 
                                                prod_name='Understanding Artificial Intelligence', 
                                                quantity=3, 
                                                price=19.5, 
                                                discount=0, 
                                                order_total=58.5)
    connection.execute(insert_statement)
    # 提交事务
    connection.commit()

    # Read
    select_statement = sales_table.select().limit(10)
    result_set = connection.execute(select_statement)
    for r in result_set:
        print(r)

    # Update
    update_statement = sales_table.update().where(sales_table.c.order_num==1105910).values(quantity=2, order_total=39)
    connection.execute(update_statement)
    connection.commit()

    # Confirm Update: Read
    reselect_statement = sales_table.select().where(sales_table.c.order_num==1105910)
    updated_set = connection.execute(reselect_statement)
    for u in updated_set:
        print(u)

    # Delete
    delete_statement = sales_table.delete().where(sales_table.c.order_num==1105910)
    connection.execute(delete_statement)
    connection.commit()

    # Confirm Delete: Read
    not_found_set = connection.execute(reselect_statement)
    print(not_found_set.rowcount)

方案2：手动定义表结构（适用于数据库中尚未创建sales表的情况） #

如果数据库里还没有sales表，需要先手动定义所有列，再创建表：

from sqlalchemy import create_engine, Table, MetaData, Column, Integer, String, Float

engine = create_engine('postgresql://postgres:123456@localhost:5432/red30')

meta = MetaData()
# 手动定义sales表的列结构
sales_table = Table(
    'sales', meta,
    Column('order_num', Integer, primary_key=True),
    Column('order_type', String),
    Column('cust_name', String),
    Column('prod_number', String),
    Column('prod_name', String),
    Column('quantity', Integer),
    Column('price', Float),
    Column('discount', Float),
    Column('order_total', Float)
)

# 创建表（仅在表不存在时执行）
meta.create_all(engine)

with engine.connect() as connection:
    # Create
    insert_statement = sales_table.insert().values(order_num=1105911, 
                                                order_type='Retail', 
                                                cust_name='Syman Mapstone', 
                                                prod_number='EB521', 
                                                prod_name='Understanding Artificial Intelligence', 
                                                quantity=3, 
                                                price=19.5, 
                                                discount=0, 
                                                order_total=58.5)
    connection.execute(insert_statement)
    connection.commit()

    # 后续的Read/Update/Delete操作同方案1，记得每次操作后commit
    # ...

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