FastAPI+SQLAlchemy Core+Postgres中使用on_conflict_do_nothing报错

在FastAPI结合SQLAlchemy Core与Postgres的项目中实现带on_conflict_do_nothing的插入语句时触发上述错误，相关信息如下：

错误日志 #

2023-12-05 13:01:46 AttributeError: 'Insert' object has no attribute 'on_conflict_do_nothing'

相关代码 #

crud.py #

from app.db import client, osr, bids, database

async def join(payload):
    insert_values = {'name': payload.name, 'code': payload.code}
    insert_stmt = client.insert().values(
        insert_values
    )
    on_conflict_stmt = insert_stmt.on_conflict_do_nothing(
        index_elements=['leiCode']
    )
    return await database.execute(query=on_conflict_stmt)

app/db.py #

import os
from sqlalchemy import (
    Column,
    UniqueConstraint,
    DateTime,
    Date,
    Integer,
    MetaData,
    String,
    Table,
    ForeignKey,
    create_engine,
    text
)
from sqlalchemy.sql import func
from databases import Database

DATABASE_URL = os.getenv("DATABASE_URL")

# SQLAlchemy
engine = create_engine(DATABASE_URL)
metadata = MetaData()


client = Table(
    "bids",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String(50), nullable=False),
    Column("code", Integer),

# databases query builder
database = Database(DATABASE_URL)

用户参考SQLAlchemy官方文档写法尝试，即使导入from sqlalchemy import insert仍报错。

on_conflict_do_nothing是PostgreSQL专属语法，需使用SQLAlchemy针对PostgreSQL的扩展Insert类，而非通用Insert对象，具体修复步骤：

• 导入PostgreSQL专属构造器
  在crud.py中替换通用insert导入，改为从PostgreSQL方言导入：

  from sqlalchemy.dialects.postgresql import insert
  

• 修改插入语句创建逻辑
  使用PostgreSQL的insert构造器创建语句，而非直接调用Table对象的insert()方法：

  async def join(payload):
      insert_values = {'name': payload.name, 'code': payload.code}
      insert_stmt = insert(client).values(insert_values)
      # 注意：index_elements需指定表中实际存在的唯一约束字段，原代码的leiCode不存在
      on_conflict_stmt = insert_stmt.on_conflict_do_nothing(
          index_elements=['code']
      )
      return await database.execute(query=on_conflict_stmt)
  

• 补充表的唯一约束
  on_conflict_do_nothing依赖唯一索引/约束，需在db.py的Table定义中添加对应字段的唯一约束：

  client = Table(
      "bids",
      metadata,
      Column("id", Integer, primary_key=True, autoincrement=True),
      Column("name", String(50), nullable=False),
      Column("code", Integer),
      UniqueConstraint('code', name='uq_client_code')
  )
  

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