使用SQLAlchemy实现Upsert时遇两类错误，求解决方案

问题背景 #

使用SQLAlchemy编写PostgreSQL的Upsert（插入更新）查询时，先后遇到两个错误：

1. sqlalchemy.exc.ArgumentError: subject table for an INSERT, UPDATE or DELETE expected, got 'comparison_bill_data'
2. 修改为Base.metadata.tables[ComparisonBillData.__tablename__]后，出现'ComparisonBillData' object has no attribute 'items'

错误原因分析 #

• 第一个错误：insert()函数必须接收SQLAlchemy的表对象（模型类或Base.metadata.tables中的表实例），不能直接传入字符串类型的表名。
• 第二个错误：values()方法要求传入字典、键值对参数或字典列表，无法直接处理模型实例对象——模型实例没有字典的items()方法。

修正后的代码 #

from uuid import uuid4
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID
from base import Base
from src.app.database.database import SessionLocal
from sqlalchemy.dialects.postgresql import insert
import datetime


class ComparisonBillData(Base):
    __tablename__ = 'comparison_bill_data'

    comparison_bill_data_id = sa.Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    bill_id = sa.Column(sa.String(), nullable=False)
    account_id = sa.Column(sa.String(), nullable=False)
    service_location_id = sa.Column(sa.String(), nullable=False)
    usage = sa.Column(sa.Numeric(), nullable=False)
    cost = sa.Column(sa.Numeric(), nullable=False)
    business_type = sa.Column(sa.String(), nullable=False)
    tenant_id = sa.Column(sa.String(), nullable=False)
    commodity_type = sa.Column(sa.String(), nullable=False)
    bill_date = sa.Column(sa.Date(), nullable=False)
    usage_unit = sa.Column(sa.String(), nullable=False)
    created_at = sa.Column(sa.TIMESTAMP(), nullable=False)
    updated_at = sa.Column(sa.TIMESTAMP(), nullable=False, server_default=sa.text('now()'))

    def __init__(self, comparison_bill_data_id,bill_id,account_id,service_location_id,usage,cost,business_type,tenant_id,commodity_type,bill_date,usage_unit,created_at):
        self.comparison_bill_data_id = comparison_bill_data_id
        self.bill_id = bill_id
        self.account_id = account_id
        self.service_location_id = service_location_id
        self.usage = usage
        self.cost = cost
        self.business_type = business_type
        self.tenant_id = tenant_id
        self.commodity_type = commodity_type
        self.bill_date = bill_date
        self.usage_unit = usage_unit
        self.created_at = created_at


def upsert_comparison_bill_data(bill_data: ComparisonBillData):
    # 传入模型类作为insert的目标表
    statement = insert(ComparisonBillData).values(
        # 将模型实例转为属性字典，过滤SQLAlchemy内部私有属性
        {key: value for key, value in sa.inspect(bill_data).attrs.items() if not key.startswith('_')}
    )
    query = statement.on_conflict_do_update(
        constraint="ui_comparison_bill_data_service_location_id", 
        set_={
            "usage": statement.excluded.usage,
            "cost": statement.excluded.cost
        }
    )
    with SessionLocal() as session:
        print("Executing query")
        session.execute(query)
        # 提交会话，确保数据持久化到数据库
        session.commit()


if __name__ == "__main__":
    print("Starting")
    data = ComparisonBillData(
        "aebebd26-4f3e-44c5-83ef-cf722f0d81e2", 
        "bill_id_1", 
        "account_id_1", 
        "service_location_id_1", 
        26.3,
        30.2,
        "Office", 
        "tenant-1", 
        "electric", 
        datetime.datetime(2023, 1, 1),
        "kWh",
        datetime.datetime(2023, 1, 1)
    )
    upsert_comparison_bill_data(data)

关键修改点说明 #

1. 表对象传入：直接使用ComparisonBillData模型类作为insert()的参数，符合SQLAlchemy的API要求，避免表名字符串导致的错误。
2. 模型转字典：通过sa.inspect(bill_data).attrs获取模型的属性字典，过滤内部私有属性，让values()能正确解析数据。
3. 会话提交：添加session.commit()，否则Upsert操作只会在会话中暂存，不会写入数据库。
4. 清理无效代码：删除了原代码中无意义的print("List="+ str(list))语句（list是Python内置函数，此处调用会引发错误）。

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