本文介绍如何使用 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 方法：
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。
总结
通过使用 SQL 的 ON CONFLICT 子句和唯一索引，我们可以有效地避免在数据库中插入重复记录，并报告操作结果。这种方法简洁高效，适用于大多数场景。但需要注意并发问题，并根据实际情况选择合适的并发控制机制。
