pythonsqlalchemyforeign-key-relationship

Python SQLAlchemy - Update time stamp on parent model when child is updated


Ive created two models, Basket and BasketItem, shown below. A basket can have many basket items so I'm using a foreign key to create this relationship.

You can see that the Basket model has a updated_at field which holds a timestamp of when the basket was last updated. I would like this field to be updated when a BasketItem is added, removed or updated.

I've added an event listener but my solution doesn't seem very elegant. Is there a better way of doing this?

class Basket(Base):
    __tablename__ = "baskets"

    id = Column(String(45), primary_key=True, default=uuid4)
    shop = Column(String(45), nullable=False)
    currency_code = Column(String(3), nullable=False)
    total = Column(String(15), nullable=False, default=0)
    status = Column(Enum("open", "closed"), nullable=False, default="open")
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.utcnow)

    items = relationship("BasketItem", backref="basket")


class BasketItem(Base):
    __tablename__ = "basket_items"

    basket_id = Column(String(45), ForeignKey('baskets.id'), primary_key=True)
    product_url = Column(String(90), nullable=False, primary_key=True)
    quantity = Column(Integer, nullable=False)

@event.listens_for(BasketItem, 'after_insert')
@event.listens_for(BasketItem, 'after_update')
@event.listens_for(BasketItem, 'after_delete')
def basket_item_change(mapper, connection, target):
    db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))
    basket = db_session.query(Basket).get(target.basket_id)
    basket.updated_at = datetime.datetime.utcnow()
    db_session.commit()

Solution

  • I think it's good to use event listener but you can do something like below. When you add a new basketitem I am sure you are query parent first and then add this child with that parent. If yes then:

    basket = Basket.query.get(id)
    if basket:
        basket.update()
        # you can perform delete or update too
        item = BasketItem(basket=basket, ...)
        db.session.add(item)
        db.session.commit()
    
    class Basket(db.Model):
        ...
        def update(self):
            self.updated_at  = datetime.utcnow()
            db.session.add(self)
            db.session.commit()