tagged_articles.py

from sqlalchemy_continuum import get_versioning_manager

class VersionBase(object):
    """Base class for SQL Alchemy continuum objects that supports tagging"""
  
    @classmethod
    def tag_current_transaction(cls, **kwargs):
        """Add keyword arguments that will be stored as a `TransactionMeta`
        entry on next successful commit that includes versioned objects.
        """
        existing_tx_meta = get_versioning_manager(cls).uow.tx_meta
        for k in kwargs.keys():
            if k in existing_tx_meta:
                raise KeyError(
                   "Key '{}' already exists in transaction meta ".format(k) +
                   "key-value pairs: '{}'".format(existing_tx_meta))
        existing_tx_meta.update(**kwargs)
        
    def remove_tags(self, **kwargs):
        """Delete all found key-value tags from any transaction"""
        TransactionMeta = get_versioning_manager(self).transaction_meta_cls
        session = object_session(self)
        transaction_meta_filter = []
        for key, value in kwargs.items():
            transaction_meta_filter.extend([TransactionMeta.key == key,
                                            TransactionMeta.value == str(value)])
        #just delete them
        session.query(TransactionMeta).filter(*transaction_meta_filter).delete()
        
    def get_tagged_version(self, **kwargs):
        """Return a query on the version objects that are associated with the
        transactions tagged with the `kwargs`. The `kwargs` values will be
        converted to strings to construct the key/value query.
        """
        TransactionMeta = self.versioning_manager.transaction_meta_cls
        TransactionLog = self.transaction_log
        session = object_session(self)
        
        transaction_meta_filter = []
        for key, value in kwargs.items():
            #convert value to string, TransactionMeta stores strings
            transaction_meta_filter.extend([TransactionMeta.key == key,
                                            TransactionMeta.value == str(value)])
        
        object_query = session.query(self.history_class
            ).join(TransactionLog,
                   TransactionLog.id == self.history_class.transaction_id
            ).join(TransactionMeta,
                   TransactionLog.id == TransactionMeta.transaction_id
            ).filter(*transaction_meta_filter)
        return object_query

#the class
class Article(Base):
  __versioned__ = {}

  def get_published():
      return self.get_tagged_version(article_published=self.id).scalar()
      
  def mark_as_published():
      #remove older tags if you want to support only 1 published version
      self.remove_tags(article_published=self.id)
      self.tag_current_transaction(article_published=self.id)