我的代码中有一个重复的模式,我觉得这是一个反模式.我有一个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
上没有索引)的解决方案,那就太好了.