SQLAlchemy validator event example

validate.py

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event
import datetime

Base = declarative_base()


def validate_int(instance, value, oldvalue, initiator):
    # Assigning a string to an Integer column will try to coerce it to the
    # proper type
    if isinstance(value, str):
        value = int(value)
    else:
        assert isinstance(value, Integer), "Not an integer!"
    return value


def validate_datetime(instance, value, oldvalue, initiator):
    assert isinstance(value, datetime.datetime), "Not a datetime object!"
    return value


validators = {
    Integer: validate_int,
    String: validate_string,
    DateTime: validate_datetime,
}


@event.listens_for(Base, 'attribute_instrument')
def configure_listener(class_, key, inst):
    """Listen to when an InstrumentedAttribute like a Column gets attached to
    an ORM class during defintion time, so we can attach events to each of
    those."""

    if not hasattr(inst.property, 'columns'):
        return

    # Find SQLAlchemy type of column, and then the validator registered for
    # that type
    column_type = inst.property.columns[0].type.__class__
    validator = validators.get(column_type)

    if validator:
        # If this is a known type, set the pre-defined validator
        event.listen(inst, "set", validator, retval=True)
    else:
        # Otherwise, attach a generic event listener that tries to call a
        # .serialize() method on the value and failing that, lets the user deal
        # with it.
        @event.listens_for(inst, "set", retval=True)
        def set_(instance, value, oldvalue, initiator):
            try:
                value = getattr(value, "serialize")()
            except AttributeError:
                raise ValueError("Value %s of type %s does not have a .serialize()!" % (value, type(value)))
            return value


class CustomBool(object):
    """A special python class with a .serialize() method that save the class
    state in a format we can store in the database."""

    def __init__(self, value):
        self.value = value

    def serialize(self):
        if self.value in ["correct", "affirmative"]:
            return True
        else:
            return False


class MyObject(Base):
    __tablename__ = 'mytable'

    id = Column(Integer, primary_key=True)
    ivalue = Column(Integer)
    dvalue = Column(DateTime)
    bvalue = Column(Boolean)


m = MyObject()

# Validation using event handlers

# These all succeed
m.ivalue = 45  # validate_int notices this as an int, uses as is
m.ivalue = "45"  # validate_int coerces the string "45" into the integer 45.
m.bvalue = CustomBool("correct") # The generic validator calls .serialize() and stores True

# These all fail
m.ivalue = True  # validate_int: True is not an int
m.dvalue = "not a date"  # validate_datetime: A string is not a datetime
m.bvalue = 0.0  # generic validator: float 0.0 doesn't have a .serialize() function

# Validation using TypeDecorator
# ... similar to ...