时间戳

首页 > 解决方案 > SQLAlchemy 事务未持久化

python - SQLAlchemy 事务未持久化

问题描述

我最近不得不在我的应用程序上实现一个事务，如下所示：

我有一个：

如果失败，我现在想执行此事务或回滚：

我的数据库是sqlite。

我的代码：

db_item = models.Items(title=item.title, description=item.description,
                        quantity=item.quantity, images=item.images,
                        category_name=item.category_name, status=item.status,
                        organization_id=item.organization_id, price=item.price)


db.begin(subtransactions=True)
db.add(db_item)
db.flush()

db_tags_to_items = [models.ItemsAndTags(tag_name=tag, item_id=db_item.id) for tag in item.tags]
db_short_descriptions = [models.ItemsShortDescription(line=line, item_id=db_item.id) for line in item.descriptions]

db.add_all(db_tags_to_items)
db.add_all(db_short_descriptions)
db.commit()

如果我返回 db_item 它会返回它的 ID，但是如果我检查 db 该项目没有持久化，也没有标签或描述。我做错了什么，因为我没有错误。

如果需要我的更多代码，请告诉我，我使用带有 SQLAlchemy 的 FastAPI 框架。

标签： pythonsqlalchemyfastapi

解决方案

我设法让它自己工作。原来 db.begin(subtransactions=True) 并不是真正需要的。

这是我现在的 crud func，它可以满足我的需要：

# create item as transaction
def create_item_v2(db: Session, item: items_schemas.ItemCreateV2):
    db_item = models.Items(title=item.title, description=item.description,
                        quantity=item.quantity, images=item.images,
                        category_name=item.category_name, status=item.status,
                        organization_id=item.organization_id, price=item.price)

    try:
        db.add(db_item)
        db.flush()
        db_tags_to_items = [models.ItemsAndTags(tag_name=tag, item_id=db_item.id) for tag in item.tags]
        db_short_descriptions = [models.ItemsShortDescription(line=line, item_id=db_item.id) for line in item.descriptions]

        db.add_all(db_tags_to_items)
        db.add_all(db_short_descriptions)
        db.commit()
        db.refresh(db_item)
        return db_item
    except:
        db.rollback()

推荐阅读