SQLAlchemy Continuum manual revisions

gistfile1.py

from itertools import chain

from sqlalchemy_continuum import (
    VersioningManager, Operation, UnitOfWork,
    make_versioned)

from sqlalchemy_continuum.utils import(
    is_versioned, versioned_objects, option, is_modified)

from sqlalchemy.orm import object_session


def is_modified_or_deleted(obj):
    """The same as sqlalchemy_continuum.utils.is_modified_or_deleted, but
    objects which are manually versioned do not not count as modified."""
    session = object_session(obj)
    manual = option(obj, 'manual')
    return is_versioned(obj) and (
        (not manual and is_modified(obj)) or
        obj in chain(session.deleted, session.new))


def is_session_modified(session):
    """The same as sqlalchemy_continuum.utils.is_session_modified, but uses the
    altered version of is_modified_or_deleted"""
    return any(is_modified_or_deleted(obj)
               for obj in versioned_objects(session))


class ManualUnitOfWork(UnitOfWork):
    """A subclass of UnitOfWork that determines if the session has been
    modified by excluding the edits on manual objects."""

    def is_modified(self, session):
        """This is the same as the default, but instead uses the altered
        version is_session_modified."""
        return (
            is_session_modified(session) or
            any(self.manager.plugins.is_session_modified(session))
        )


class ManualVersioningManager(VersioningManager):
    """A subclass of VersioningManager which allows some models to not
    automatically create a version when edited. A model can be made manual
    by adding a `manual=True` parameter to its `__versioned__` object"""

    def __init__(self, *args, **kwargs):
        """Add a default `manual` parameter to the options dictionary. By
        default, objects are not manual."""
        kwargs.setdefault('options', {})
        kwargs['options'].setdefault('manual', False)
        super(ManualVersioningManager, self).__init__(*args, **kwargs)

    def track_updates(self, mapper, connection, target):
        """Overrides the sqlalchemy on update hook. If the target is not a
        manually versioned object, the default implementation is called. If the
        target is manually versioned then this function does nothing, resulting
        in the changes for the model not being added to the unit of work and
        therefore no version is created"""
        if is_versioned(target) and not self.option(target, 'manual'):
            return super(ManualVersioningManager, self).track_updates(
                mapper, connection, target)

    def create_version_for(self, *models):
        """Manually create a version for the given models."""

        # We need a single session, but multiple models can be provided. If all
        # models dont have the same session, raise an execption.
        sessions = set(object_session(model) for model in models)
        if len(sessions) != 1:
            raise Exception("Cannot transact on multiple sessions")

        # Create or get the unit of work associated with the session.
        session = sessions.pop()
        uow = self.unit_of_work(session)

        # Add an update operation for each model.
        for model in models:
            uow.operations.add(Operation(model, Operation.UPDATE))

        # Manually create the transaction record and the version records.
        # Without this, the transaction and versions will only be created if
        # There are other changes than the manual ones.
        uow.create_transaction(session)
        uow.make_versions(session)


versioning_manager = ManualVersioningManager(unit_of_work_cls=ManualUnitOfWork)
make_versioned(manager=versioning_manager)