使用 SQL ON CONFLICT 避免重复记录并报告操作结果-Python教程-PHP中文网

使用 SQL ON CONFLICT 避免重复记录并报告操作结果

碧海醫心

发布： 2025-11-18 11:49:50

原创

588人浏览过

使用 sql on conflict 避免重复记录并报告操作结果

本文介绍如何使用 SQL 的 `ON CONFLICT` 子句来避免在数据库中插入重复记录，并提供一种方法来报告记录是新创建的还是已存在的。通过在 `name` 列上添加唯一索引，我们可以利用 `INSERT ... ON CONFLICT DO NOTHING` 语句，结合后续查询，实现高效的重复记录检查和创建逻辑，并返回操作结果。

在处理数据库操作时，经常需要检查要插入的数据是否已存在。如果存在，则不执行插入操作；如果不存在，则创建新记录。同时，还需要向用户报告操作结果，即记录是新创建的还是已存在的。本文将介绍如何使用 SQL 的 ON CONFLICT 子句，结合唯一索引，来实现这一需求，并提供 Python 代码示例。

前提条件

熟悉 SQL 语法，特别是 INSERT 语句和 ON CONFLICT 子句。
了解数据库唯一索引的概念和作用。
使用支持 ON CONFLICT 子句的数据库，如 PostgreSQL。

解决方案

核心思想是利用 INSERT ... ON CONFLICT DO NOTHING 语句尝试插入新记录。如果记录已存在（违反唯一约束），则不执行任何操作。然后，通过查询数据库来确定记录是否存在，从而判断是新创建的还是已存在的。

1. 创建唯一索引

首先，需要在 name 列上创建唯一索引。这将确保 name 列的值在表中是唯一的。

CREATE UNIQUE INDEX ux_product_categories_name ON product_categories (name);

登录后复制

2. 实现 get_or_create 方法

以下是一个 Python 代码示例，使用 SQLAlchemy 和 PostgreSQL 实现 get_or_create 方法：

降重鸟

要想效果好，就用降重鸟。AI改写智能降低AIGC率和重复率。

113

查看详情

from sqlalchemy import create_engine, Column, Integer, String, text, inspect, select
from sqlalchemy.orm import Session, Mapped, mapped_column
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class ProductCategory(Base):
    __tablename__ = "product_category"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)

    @staticmethod
    def get_or_create(name: str, session: Session) -> "ProductCategory":
        with session.bind.begin() as conn:
            result = conn.execute(
                text(
                    "INSERT INTO product_category (name) "
                    "VALUES (:name) ON CONFLICT (name) DO NOTHING"
                ),
                dict(name=name),
            )
            print(
                f">>> INFO: {'returning existing category' if result.rowcount == 0 else 'category created'}"
            )
        return session.scalars(
            select(ProductCategory).where(ProductCategory.name == name)
        ).one()

    def __repr__(self):
        return f"ProductCategory(id={self.id!r}, name={self.name!r})"

# 示例用法
engine = create_engine("postgresql://user:password@host:port/database") # 替换为你的数据库连接字符串
Base.metadata.create_all(engine)

with Session(engine) as sess:
    pc = ProductCategory.get_or_create("shoes", sess)
    # >>> INFO: category created
    insp = inspect(pc)
    print(f"{pc}, persistent={insp.persistent}")
    # ProductCategory(id=1, name='shoes'), persistent=True
    another_pc = ProductCategory.get_or_create("shoes", sess)
    # >>> INFO: returning existing category
    print(another_pc)
    # ProductCategory(id=1, name='shoes')
    print(pc == another_pc)
    # True

登录后复制

代码解释:

CREATE TABLE 和 CREATE UNIQUE INDEX: 首先，确保你的数据库表已经创建，并且在 name 列上有一个唯一索引。
INSERT ... ON CONFLICT DO NOTHING: 这个 SQL 语句尝试插入一个新的 ProductCategory。 ON CONFLICT (name) 指定了当 name 列发生冲突（即已存在相同的值）时，执行 DO NOTHING，即什么也不做。
result.rowcount: conn.execute() 返回一个结果对象，其 rowcount 属性表示受影响的行数。如果 rowcount 为 0，说明插入操作没有执行，即记录已存在。
session.scalars(select(...)).one(): 这个 SQLAlchemy 代码用于查询数据库，获取具有给定 name 的 ProductCategory 对象。 one() 方法确保返回一个且仅返回一个结果。

3. 使用示例

在上面的代码示例中，首先创建了一个名为 "shoes" 的 ProductCategory。然后，再次尝试创建名为 "shoes" 的 ProductCategory。由于 name 列上存在唯一索引，第二次插入操作会因为冲突而失败，但 get_or_create 方法会返回已存在的 ProductCategory 对象。

注意事项

唯一索引是关键: 此方法依赖于 name 列上的唯一索引。如果没有唯一索引，ON CONFLICT 子句将不起作用，可能导致插入重复记录。
并发问题: 在高并发环境下，可能会出现竞态条件。虽然唯一索引可以防止插入重复记录，但在两个线程同时检查记录是否存在时，可能会都认为记录不存在，从而都尝试插入新记录。可以使用更高级的并发控制机制来解决这个问题，例如悲观锁或乐观锁。
数据库支持: ON CONFLICT 子句是 PostgreSQL 特有的。其他数据库可能使用不同的语法来实现类似的功能。例如，MySQL 可以使用 INSERT IGNORE 或 INSERT ... ON DUPLICATE KEY UPDATE。