Source code for examples.custom_attributes.custom_management
"""Illustrates customized class instrumentation, usingthe :mod:`sqlalchemy.ext.instrumentation` extension package.In this example, mapped classes are modified tostore their state in a dictionary attached to an attributenamed "_goofy_dict", instead of using __dict__.this example illustrates how to replace SQLAlchemy's classdescriptors with a user-defined system."""fromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportForeignKeyfromsqlalchemyimportIntegerfromsqlalchemyimportMetaDatafromsqlalchemyimportTablefromsqlalchemyimportTextfromsqlalchemy.ext.instrumentationimportInstrumentationManagerfromsqlalchemy.ormimportregistryas_regfromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportSessionfromsqlalchemy.orm.attributesimportdel_attributefromsqlalchemy.orm.attributesimportget_attributefromsqlalchemy.orm.attributesimportset_attributefromsqlalchemy.orm.instrumentationimportis_instrumentedregistry=_reg()classMyClassState(InstrumentationManager):defget_instance_dict(self,class_,instance):returninstance._goofy_dictdefinitialize_instance_dict(self,class_,instance):instance.__dict__["_goofy_dict"]={}definstall_state(self,class_,instance,state):instance.__dict__["_goofy_dict"]["state"]=statedefstate_getter(self,class_):deffind(instance):returninstance.__dict__["_goofy_dict"]["state"]returnfindclassMyClass:__sa_instrumentation_manager__=MyClassStatedef__init__(self,**kwargs):forkinkwargs:setattr(self,k,kwargs[k])def__getattr__(self,key):ifis_instrumented(self,key):returnget_attribute(self,key)else:try:returnself._goofy_dict[key]exceptKeyError:raiseAttributeError(key)def__setattr__(self,key,value):ifis_instrumented(self,key):set_attribute(self,key,value)else:self._goofy_dict[key]=valuedef__delattr__(self,key):ifis_instrumented(self,key):del_attribute(self,key)else:delself._goofy_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)classA(MyClass):passclassB(MyClass):passregistry.map_imperatively(A,table1,properties={"bs":relationship(B)})registry.map_imperatively(B,table2)a1=A(name="a1",bs=[B(name="b1"),B(name="b2")])asserta1.name=="a1"asserta1.bs[0].name=="b1"sess=Session(engine)sess.add(a1)sess.commit()a1=sess.query(A).get(a1.id)asserta1.name=="a1"asserta1.bs[0].name=="b1"a1.bs.remove(a1.bs[0])sess.commit()a1=sess.query(A).get(a1.id)assertlen(a1.bs)==1