ValidateAllOccurrencesOfType

The use case here is when a mapper level validation function, i.e. that described at http://www.sqlalchemy.org/docs/orm/mapper_config.html#simple-validators, is desired on all occurrences of an attribute that meets a specific type. This can be accomplished using the Event Package, listening first for newly mapped classes, then establishing listeners on all attributes which represent the target type.

The "mapper listener to attribute listener" approach here is something we are considering making into more of a built in feature. For now the recipe is fairly simple. The same idea here is also used by the Mutable Scalars extension to apply instrumentation to attributes of a particular type.

#!python
from sqlalchemy import TypeDecorator, String, event
from sqlalchemy.orm import mapper

class MySpecialType(str):
    """A str subclass that adds some functionality to strings."""

    def pretty_print(self):
        return "---~--%% %15.15s %%--~--- :) :) :)" % self

class StoresMySpecialType(TypeDecorator):
    """Intercept strings passed to SQL statements and received 
    from result sets.    Assert values are already MySpecialType
    on the bind param side, coerce to MySpecialType on the 
    result side."""

    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            assert isinstance(value, MySpecialType), \
                    "Only MySpecialType accepted !"
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = MySpecialType(value)
        return value

def _coerce_to_my_special_type(target, value, oldvalue, initiator):
    """An attribute listener function that receives values as they
    are set on a mapped class.  Coerces values to MySpecialType."""

    if value is not None and \
        not isinstance(value, MySpecialType):
        value = MySpecialType(value)
    return value

@event.listens_for(mapper, "mapper_configured")
def _setup_my_special_listeners(mapper, class_):
    """A mapper-configured listener that establishes the 
    _coerce_to_my_special_type listener on all mapped columns that
    include StoresMySpecialType.

    """
    for prop in mapper.iterate_properties:
        if hasattr(prop, 'columns'):
            if isinstance(prop.columns[0].type, StoresMySpecialType):
                event.listen(
                    getattr(class_, prop.key), 
                    "set", 
                    _coerce_to_my_special_type,
                    retval=True)


if __name__ == '__main__':
    """demonstration case starts."""

    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, create_engine
    from sqlalchemy.orm import Session

    Base = declarative_base()

    class Something(Base):
        __tablename__ = 'some_table'
        id = Column(Integer, primary_key=True)
        data = Column(StoresMySpecialType)

    somethings = [
        Something(data="who"),
        Something(data="is"),
        Something(data="the prettiest!")
    ]

    # the data is coerced to the special type
    # upon set.
    for x in somethings:
        print x.data.pretty_print()


    # demonstrate persistence.
    e = create_engine('sqlite://')
    Base.metadata.create_all(e)

    session = Session(e)
    session.add_all(somethings)
    session.commit()

    for x, in session.query(Something.data):
        print x.pretty_print()

Uh oh!

ValidateAllOccurrencesOfType