gistfile1.py

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, Text,\
    ForeignKey
from sqlalchemy.orm import mapper, relationship, Session

from sqlalchemy.orm.attributes import set_attribute, get_attribute, \
    del_attribute
from sqlalchemy.orm.instrumentation import is_instrumented
from sqlalchemy.orm import instrumentation
from sqlalchemy.ext.instrumentation import InstrumentationManager

class MyClassManager(InstrumentationManager):
    def install_descriptor(self, class_, key, inst):
        # dont install any descriptors.
        pass

    def uninstall_descriptor(self, class_, key):
        pass

    def install_member(self, class_, key, implementation):
        # in fact, don't install anything on the class.
        pass

    def uninstall_member(self, class_, key):
        pass

    def get_instance_dict(self, class_, instance):
        # don't want SQLAlchemy reading/writing from instance.__dict__?
        # use any dictionary you want here - store it in a weak registry,
        # attach it to the instance, whatever.  It's SQLAlchemy's view
        # of the state of your object.
        return instance.__dict__

    def install_state(self, class_, instance, state):
        # SQLAlchemy wants to store an InstanceState with the object.
        # define here how you'd like it do to that.
        setattr(instance, '_default_state', state)

    def remove_state(self, class_, instance):
        # and remove it...
        delattr(instance, '_default_state')

    def state_getter(self, class_):
        # and how to get it.
        return lambda instance: getattr(instance, '_default_state')

class MyClass(object):

    # one of several methods of establishing alternate implementation
    __sa_instrumentation_manager__ = MyClassManager

    # now hand-roll what SQLAlchemy needs in order to track state.
    # no descriptors or wrapped methods in use here!
    def __init__(self, **kwargs):
        manager = instrumentation.manager_of_class(self.__class__)

        # this creates the InstanceState object and calls our install_state()
        # method.
        manager.setup_instance(self)

        for k in kwargs:
            setattr(self, k, kwargs[k])

    def __getattr__(self, key):
        # return SQLAlchemy getattr, reads from dict, does lazy loading,
        # etc.  if attribute is locally present in __dict__ then this isn't
        # called.

        if is_instrumented(self, key):
            return get_attribute(self, key)
        else:
            raise AttributeError(key)

    def __setattr__(self, key, value):
        # set attribute with history, etc.
        if is_instrumented(self, key):
            set_attribute(self, key, value)
        else:
            self.__dict__[key] = value

    def __delattr__(self, key):
        # delete attribute with history, etc.
        if is_instrumented(self, key):
            del_attribute(self, key)
        else:
            del self.__dict__[key]


if __name__ == '__main__':
    engine = create_engine('sqlite://')
    meta = MetaData()

    table1 = Table('table1', meta,
                    Column('id', Integer, primary_key=True),
                    Column('name', Text))
    table2 = Table('table2', meta,
                    Column('id', Integer, primary_key=True),
                    Column('name', Text),
                    Column('t1id', Integer, ForeignKey('table1.id')))
    meta.create_all(engine)

    class A(MyClass):
        pass

    class B(MyClass):
        pass

    mapper(A, table1, properties={
        'bs': relationship(B)
    })

    mapper(B, table2)

    a1 = A(name='a1', bs=[B(name='b1'), B(name='b2')])

    assert a1.name == 'a1'
    assert a1.bs[0].name == 'b1'

    sess = Session(engine)
    sess.add(a1)

    sess.commit()

    a1 = sess.query(A).get(a1.id)

    assert a1.name == 'a1'
    assert a1.bs[0].name == 'b1'

    a1.bs.remove(a1.bs[0])

    sess.commit()

    a1 = sess.query(A).get(a1.id)
    assert len(a1.bs) == 1