我的代码中有一个重复的模式,我觉得这是一个反模式.我有一个sqlalchemy发动机设置为这样:

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

当我想使用原始SQL创建一个新条目时,我有,例如:

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"

对我来说,应该可以用一个查询来完成这项工作and是否成功(请注意,在这两种情况下,return条语句都给了我一个要回馈前端的元组).我try 过CTE,并沿着类似ON CONFLICT DO NOTHING RETURNING EXCLUDED.name的路径前进,但主键不是name,它是自动生成的id,我迷路了.我可以做相反的事情,没有问题,因为我从前台得到了id分:

@staticmethod
def delete(id_: int):
​
    with autoengine.connect() as conn:
        
        q = text(
            """
            UPDATE
                unit_measures
            SET
                is_deprecated = true
            WHERE
                id = :id
            RETURNING
                name
            """)
        
        deleted = list(conn.execute(q, {'id': id_}))
        if not deleted:
            return False, "Unit measure ID not found"
        
        return True, f"Unit measure successfully deleted: {deleted[0][0]}"

有没有办法将这两个查询合并为一个?这是Postgres,但我很高兴任何方言,如果这种方法可以翻译.


我要澄清的是:我很清楚比赛的情况,我知道在目前的设置下它们可能是如何发生的.如果创建一个UNIQUE约束或一个索引是necessary来合并我的查询,这是一个可以接受的答案.但是,如果有一个使用现有数据库设置(name上没有索引)的解决方案,那就太好了.

推荐答案

正如 comments 中所讨论的,您需要在您的name列上放置一个唯一索引

CREATE UNIQUE INDEX ux_product_categories_name ON product_categories (name)

所以你可以做这样的事情

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})"

你可以这样使用它

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

Python相关问答推荐

将numpy数组与空数组相加

pandas DataFrame中类型转换混乱

替换字符串中的点/逗号,以便可以将其转换为浮动

在Python中为变量的缺失值创建虚拟值

将HTML输出转换为表格中的问题

GL pygame无法让缓冲区与vertextPointer和colorPointer一起可靠地工作

在Python中对分层父/子列表进行排序

如何自动抓取以下CSV

@Property方法上的inspect.getmembers出现意外行为,引发异常

TARete错误:类型对象任务没有属性模型'

如何从具有不同len的列表字典中创建摘要表?

将输入管道传输到正在运行的Python脚本中

我如何使法国在 map 中完全透明的代码?

如何根据一列的值有条件地 Select 前N个组,然后按两列分组?

连接一个rabrame和另一个1d rabrame不是问题,但当使用[...]'运算符会产生不同的结果

用渐近模计算含符号的矩阵乘法

需要帮助重新调整python fill_between与数据点

Polars asof在下一个可用日期加入

未调用自定义JSON编码器

使用字典或列表的值组合