如何为非唯一主键更新 Sqlalchemy ORM 中的特定行
How to update specific row in Sqlalchemy ORM for a non unique primary key
我正在尝试使用 SQLAlchemy ORM 来更新数据库中的特定行,但是因为我想保留该行发生的所有事情的历史记录,所以我想取消记录版本并保留相同的主键,即主键不是唯一的。因此,当我尝试对旧记录进行版本化并创建新记录时,出现以下错误"
sqlalchemy.orm.exc.StaleDataError: UPDATE statement on table 'category' expected to update 1 row(s); 2 were matched.
正在尝试使用以下功能:
def update_category(db: Session, category: CategoryUpdate, category_code: str) -> CategoryTable:
"""update the category by ID where current_flag = 1"""
cat = db.query(CategoryTable).filter(CategoryTable.current_flag == "1", CategoryTable.category_code == category_code).first()
if cat is None:
raise Exception("Category not found")
cat.effective_to = datetime.now()
cat.current_flag = "0"
db.commit()
db.refresh(cat)
c = CategoryIn(**cat.__dict__)
c.category_name = category.category_name
# c.category_key = cat.category_key
c.category_class = category.category_class if category.category_class else c.category_class
c.effective_from = c.effective_to
c.effective_to = None
c.current_flag = "1"
c_tab = CategoryTable(**c.dict())
c_tab.category_key = cat.category_key
db.add(c_tab)
db.commit()
db.refresh(c_tab)
return c_tab
有什么想法吗?
明白了,如果你有一个比使用主键更复杂的更新,你可以从 sqlalchemy 导入更新函数
stmt = update(CategoryTable).where(CategoryTable.category_key == cat.category_key,
CategoryTable.current_flag == '1', CategoryTable.effective_from == cat.effective_from).values(
effective_to=effective_date, current_flag='0').execution_options(synchronize_session="fetch")
db.execute(stmt)
我正在尝试使用 SQLAlchemy ORM 来更新数据库中的特定行,但是因为我想保留该行发生的所有事情的历史记录,所以我想取消记录版本并保留相同的主键,即主键不是唯一的。因此,当我尝试对旧记录进行版本化并创建新记录时,出现以下错误"
sqlalchemy.orm.exc.StaleDataError: UPDATE statement on table 'category' expected to update 1 row(s); 2 were matched.
正在尝试使用以下功能:
def update_category(db: Session, category: CategoryUpdate, category_code: str) -> CategoryTable:
"""update the category by ID where current_flag = 1"""
cat = db.query(CategoryTable).filter(CategoryTable.current_flag == "1", CategoryTable.category_code == category_code).first()
if cat is None:
raise Exception("Category not found")
cat.effective_to = datetime.now()
cat.current_flag = "0"
db.commit()
db.refresh(cat)
c = CategoryIn(**cat.__dict__)
c.category_name = category.category_name
# c.category_key = cat.category_key
c.category_class = category.category_class if category.category_class else c.category_class
c.effective_from = c.effective_to
c.effective_to = None
c.current_flag = "1"
c_tab = CategoryTable(**c.dict())
c_tab.category_key = cat.category_key
db.add(c_tab)
db.commit()
db.refresh(c_tab)
return c_tab
有什么想法吗?
明白了,如果你有一个比使用主键更复杂的更新,你可以从 sqlalchemy 导入更新函数
stmt = update(CategoryTable).where(CategoryTable.category_key == cat.category_key,
CategoryTable.current_flag == '1', CategoryTable.effective_from == cat.effective_from).values(
effective_to=effective_date, current_flag='0').execution_options(synchronize_session="fetch")
db.execute(stmt)