如何将PostgreSQL中查重+插入的SQL操作合并为单查询？

问题描述 #

我在代码里发现重复的查询模式，觉得这是反模式。当前SQLAlchemy引擎配置如下：

engine = create_engine(Config.CONN_STRING)
autoengine = engine.execution_options(isolation_level="AUTOCOMMIT")

以ProductCategory类为例，新增记录时需要先查询是否存在同名记录，再执行插入，代码如下：

class ProductCategory(Base):
    
    __tablename__ = 'product_categories'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name = Column(String)
    
    @staticmethod
    def create_new(
        name: str
    ):
        with autoengine.connect() as conn:
            
            q = text(
                """
                SELECT 
                    name
                FROM
                    product_categories
                WHERE
                    name = :name
                """)
            
            data = list(conn.execute(q, {'name': name}))
            if data:
                return False, f"Product category: {name} already exists"
            
            q = text(
                """
                INSERT INTO product_categories (
                    name
                )
                VALUES (
                    :name
                )
                """)
            
            conn.execute(q, {'name': name})
            
            return True, f"Product category: {name} successfully created"

我想把这两个查询合并成一个，同时返回执行结果（返回元组给前端）。试过CTE和ON CONFLICT DO NOTHING RETURNING EXCLUDED.name的写法，但主键是自动生成的id而非name，没成功。另外我已经能实现基于ID的删除操作（用UPDATE+RETURNING判断结果），想借鉴这种方式。

环境是PostgreSQL，也接受可移植到其他方言的方案。补充：我知道当前代码有竞态条件，若必须新增UNIQUE约束或索引才能合并查询可以接受，但最好能保留现有数据库结构（name字段无索引）。

解决方案 #

方案1：无需新增唯一约束（利用CTE+INSERT+RETURNING） #

可以用PostgreSQL的CTE先查询目标名称是否存在，再在INSERT语句中通过NOT EXISTS判断是否执行插入，最后用RETURNING返回插入的名称，以此判断操作是否成功。

修改后的create_new方法如下：

@staticmethod
def create_new(name: str):
    with autoengine.connect() as conn:
        q = text("""
            WITH existing AS (
                SELECT name FROM product_categories WHERE name = :name
            )
            INSERT INTO product_categories (name)
            SELECT :name
            WHERE NOT EXISTS (SELECT 1 FROM existing)
            RETURNING name
        """)
        result = list(conn.execute(q, {'name': name}))
        if result:
            return True, f"Product category: {name} successfully created"
        else:
            return False, f"Product category: {name} already exists"

这个方案不需要修改数据库结构，但要注意：因为name没有索引，高并发场景下性能会受影响，且竞态条件依然存在（两个请求同时查询都没找到，然后都执行插入）。

方案2：新增name字段唯一约束（高效且避免竞态） #

如果可以给name字段添加唯一约束，推荐用ON CONFLICT语法，这是PostgreSQL处理这类场景的标准方式，不仅能合并查询，还能从数据库层面避免重复数据，消除竞态条件。

第一步：添加唯一约束

可以通过SQLAlchemy模型修改，或者直接执行SQL：

ALTER TABLE product_categories ADD CONSTRAINT unique_product_category_name UNIQUE (name);

或者在模型中给name字段添加unique=True：

name = Column(String, unique=True)

第二步：修改create_new方法

@staticmethod
def create_new(name: str):
    with autoengine.connect() as conn:
        q = text("""
            INSERT INTO product_categories (name)
            VALUES (:name)
            ON CONFLICT (name) DO NOTHING
            RETURNING name
        """)
        result = list(conn.execute(q, {'name': name}))
        if result:
            return True, f"Product category: {name} successfully created"
        else:
            return False, f"Product category: {name} already exists"

这个方案性能更好，因为唯一约束会自动创建索引，查询和插入的效率都更高，同时彻底解决竞态问题。

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