Source code for examples.vertical.dictlike-polymorphic
"""Mapping a polymorphic-valued vertical table as a dictionary.Builds upon the dictlike.py example to also add differently typedcolumns to the "fact" table, e.g.:: Table('properties', metadata Column('owner_id', Integer, ForeignKey('owner.id'), primary_key=True), Column('key', UnicodeText), Column('type', Unicode(16)), Column('int_value', Integer), Column('char_value', UnicodeText), Column('bool_value', Boolean), Column('decimal_value', Numeric(10,2)))For any given properties row, the value of the 'type' column will point to the'_value' column active for that row.This example approach uses exactly the same dict mapping approach as the'dictlike' example. It only differs in the mapping for vertical rows. Here,we'll use a @hybrid_property to build a smart '.value' attribute that wraps upreading and writing those various '_value' columns and keeps the '.type' up todate."""fromsqlalchemyimporteventfromsqlalchemyimportliteral_columnfromsqlalchemy.ext.hybridimporthybrid_propertyfromsqlalchemy.orm.interfacesimportPropComparatorfrom.dictlikeimportProxiedDictMixinclassPolymorphicVerticalProperty(object):"""A key/value pair with polymorphic value storage. The class which is mapped should indicate typing information within the "info" dictionary of mapped Column objects; see the AnimalFact mapping below for an example. """def__init__(self,key,value=None):self.key=keyself.value=value@hybrid_propertydefvalue(self):fieldname,discriminator=self.type_map[self.type]iffieldnameisNone:returnNoneelse:returngetattr(self,fieldname)@value.setterdefvalue(self,value):py_type=type(value)fieldname,discriminator=self.type_map[py_type]self.type=discriminatoriffieldnameisnotNone:setattr(self,fieldname,value)@value.deleterdefvalue(self):self._set_value(None)@value.comparatorclassvalue(PropComparator):"""A comparator for .value, builds a polymorphic comparison via CASE."""def__init__(self,cls):self.cls=clsdef_case(self):pairs=set(self.cls.type_map.values())whens=[(literal_column("'%s'"%discriminator),cast(getattr(self.cls,attribute),String),)forattribute,discriminatorinpairsifattributeisnotNone]returncase(whens,value=self.cls.type,else_=null())def__eq__(self,other):returnself._case()==cast(other,String)def__ne__(self,other):returnself._case()!=cast(other,String)def__repr__(self):return"<%s%r=%r>"%(self.__class__.__name__,self.key,self.value)@event.listens_for(PolymorphicVerticalProperty,"mapper_configured",propagate=True)defon_new_class(mapper,cls_):"""Look for Column objects with type info in them, and work up a lookup table."""info_dict={}info_dict[type(None)]=(None,"none")info_dict["none"]=(None,"none")forkinmapper.c.keys():col=mapper.c[k]if"type"incol.info:python_type,discriminator=col.info["type"]info_dict[python_type]=(k,discriminator)info_dict[discriminator]=(k,discriminator)cls_.type_map=info_dictif__name__=="__main__":fromsqlalchemyimport(Column,Integer,Unicode,ForeignKey,UnicodeText,and_,or_,String,Boolean,cast,null,case,create_engine,)fromsqlalchemy.ormimportrelationship,Sessionfromsqlalchemy.orm.collectionsimportattribute_mapped_collectionfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ext.associationproxyimportassociation_proxyBase=declarative_base()classAnimalFact(PolymorphicVerticalProperty,Base):"""A fact about an animal."""__tablename__="animal_fact"animal_id=Column(ForeignKey("animal.id"),primary_key=True)key=Column(Unicode(64),primary_key=True)type=Column(Unicode(16))# add information about storage for different types# in the info dictionary of Columnsint_value=Column(Integer,info={"type":(int,"integer")})char_value=Column(UnicodeText,info={"type":(str,"string")})boolean_value=Column(Boolean,info={"type":(bool,"boolean")})classAnimal(ProxiedDictMixin,Base):"""an Animal"""__tablename__="animal"id=Column(Integer,primary_key=True)name=Column(Unicode(100))facts=relationship("AnimalFact",collection_class=attribute_mapped_collection("key"))_proxied=association_proxy("facts","value",creator=lambdakey,value:AnimalFact(key=key,value=value),)def__init__(self,name):self.name=namedef__repr__(self):return"Animal(%r)"%self.name@classmethoddefwith_characteristic(self,key,value):returnself.facts.any(key=key,value=value)engine=create_engine("sqlite://",echo=True)Base.metadata.create_all(engine)session=Session(engine)stoat=Animal("stoat")stoat["color"]="red"stoat["cuteness"]=7stoat["weasel-like"]=Truesession.add(stoat)session.commit()critter=session.query(Animal).filter(Animal.name=="stoat").one()print(critter["color"])print(critter["cuteness"])print("changing cuteness value and type:")critter["cuteness"]="very cute"session.commit()marten=Animal("marten")marten["cuteness"]=5marten["weasel-like"]=Truemarten["poisonous"]=Falsesession.add(marten)shrew=Animal("shrew")shrew["cuteness"]=5shrew["weasel-like"]=Falseshrew["poisonous"]=Truesession.add(shrew)session.commit()q=session.query(Animal).filter(Animal.facts.any(and_(AnimalFact.key=="weasel-like",AnimalFact.value==True)))print("weasel-like animals",q.all())q=session.query(Animal).filter(Animal.with_characteristic("weasel-like",True))print("weasel-like animals again",q.all())q=session.query(Animal).filter(Animal.with_characteristic("poisonous",False))print("animals with poisonous=False",q.all())q=session.query(Animal).filter(or_(Animal.with_characteristic("poisonous",False),~Animal.facts.any(AnimalFact.key=="poisonous"),))print("non-poisonous animals",q.all())q=session.query(Animal).filter(Animal.facts.any(AnimalFact.value==5))print("any animal with a .value of 5",q.all())