Skip to content

Instantly share code, notes, and snippets.

@adharris
Created November 10, 2014 15:18
Show Gist options
  • Save adharris/23f1fc4617e6864f30e6 to your computer and use it in GitHub Desktop.
Save adharris/23f1fc4617e6864f30e6 to your computer and use it in GitHub Desktop.
SQLAlchemy Continuum manual revisions
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment