1. Michael Bayer
  2. sqlalchemy

Wiki

Clone wiki

sqlalchemy / UsageRecipes / SessionAwareMapper

Session-Aware Classes

This recipe is a userland workaround for most of the functionality provided by the now-deprecated Session.mapper function.

The most rudimental version is as follows:

from sqlalchemy.orm import mapper as sqla_mapper

def session_mapper(scoped_session):
    def mapper(cls, *arg, **kw):
        cls.query = scoped_session.query_property()
        return sqla_mapper(cls, *arg, **kw)
    return mapper

This version provides the MyClass.query attribute on all mapped classes. Usage is as follows:

from sqlalchemy.orm import sessionmaker, scoped_session

Session = scoped_session(sessionmaker())
mapper = session_mapper(Session)

mapper(MyClass, sometable)

print MyClass.query.all()

Another feature Session.mapper provides is the convenience __init__ method. Keep in mind the most straightforward way to do this is by defining a base class from which your classes descend:

class Base(object):
    def __init__(self, **kw):
        for k, v in kw.iteritems():
            setattr(self, k, v)

class MyClass(Base):
    # ...

Using the declarative extension also provides a method like the above. But let's illustrate how to add this to our session_mapper() function:

from sqlalchemy.orm import mapper as sqla_mapper

def session_mapper(scoped_session):
    def mapper(cls, *arg, **kw):
        if cls.__init__ is object.__init__:
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
            cls.__init__ = __init__
        cls.query = scoped_session.query_property()
        return sqla_mapper(cls, *arg, **kw)
    return mapper

Now objects can be constructed like:

x = MyClass(x=1, y=2)

Session.mapper also included a validation feature that checks that the given keyword arguments are mapped attribute names. Not all business objects would want this feature. But we illustrate how to implement that as follows:

from sqlalchemy.orm import mapper as sqla_mapper

def session_mapper(scoped_session):
    def mapper(cls, *arg, **kw):
        validate = kw.pop('validate', False)

        if cls.__init__ is object.__init__:
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    if validate:
                        if not cls_mapper.has_property(key):
                            raise TypeError(
                                "Invalid __init__ argument: '%s'" % key)
                    setattr(self, key, value)
            cls.__init__ = __init__
        cls.query = scoped_session.query_property()
        cls_mapper = sqla_mapper(cls, *arg, **kw)
        return cls_mapper
    return mapper

Where above, adding the validate=True argument to your mapper will cause the __init__ method to check the names of the arguments passed.

The key difference between this recipe and what Session.mapper did is that it does not auto-add() objects to the Session. This behavior can easily be added by adding scoped_session.add line to the init function above:

def __init__(self, **kwargs):
     for key, value in kwargs.items():
          setattr(self, key, value)
     scoped_session.add(self)

So putting it all together looks thus:

Session = scoped_session(sessionmaker())
mapper = session_mapper(Session)

mapper(MyClass, sometable, validate=True)
obj = MyClass(x=1, y=2)
Session.add(obj)
Session.commit()

assert MyClass.query.filter(MyClass.x == 2).first() is obj

Updated