Last active
August 29, 2015 13:57
-
-
Save zzzeek/9667904 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment