Recientemente tuve que implementar una transacción en mi aplicación que dice lo siguiente:
Tengo un:
Ahora quiero realizar esta transacción o retroceder si falla:
Mi base de datos es sqlite.
mi código:
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()Si devuelvo db_item, lo devuelve con la identificación, sin embargo, si verifico el db, ese elemento no se conservó ni las etiquetas o descripciones. ¿Qué estoy haciendo mal ya que no recibo ningún error?
Si necesita más de mi código, hágamelo saber, uso el marco FastAPI con SQLAlchemy.
Me las arreglé para hacerlo funcionar yo mismo. Resulta que db.begin(subtransactions=True) no era realmente necesario.
esta es mi función básica ahora y hace lo que necesito:
# 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()