> > Hi, i have some issue. I tried to replace namedtuple to dict and list > items (with adding hast function to them). But SQLAlchemy types returns > list ordereddict, when i change items, session didn't see changes. Do you > found solution? > def make_hash(o): if isinstance(o,(set,tuple,list)): for i in o: i= tuple([make_hash(e) for e in o]) return hash(i) elif isinstance(o,dict): pass i=make_hash([make_hash(j)for j in o.items()]) return hash(i)
elif isinstance(o,(str,int)): return hash(o) else: try: return hash(o) except: return make_hash(o.__dict__) return hash(o) # sys.exit() class hashabledict(dict): def __key(self): return tuple((k,self[k]) for k in sorted(self)) def __hash__(self): print 'try to make_hash'+str(self)+' result = '+ str(make_hash(self)) return make_hash(self) def __eq__(self, other): return make_hash(self) == make_hash(other) class hashablelist(list): def __hash__(self): return make_hash(self) def __eq__(self, other): return make_hash(self) == make_hash(other) class MyCompositeType(CompositeType): pass def __init__(self, name, columns): if psycopg2 is None: raise ImproperlyConfigured("'psycopg2' package is required in order to use CompositeType.") sqlalchemy.types.SchemaType.__init__(self) self.name = name self.columns = columns if name in registered_composites: self.type_cls = registered_composites[name].type_cls else: temp=hashablelist() for c in columns: temp.append(c.name) self.__dict__.update({c.name:None}) self.type_cls=hashabledict({self.name:temp}) registered_composites[name] = self class Caster(psycopg2.extras.CompositeCaster): def make(obj, values): my_dict=hashabledict() for index,data in enumerate(self.type_cls.items()[0][1]): if isinstance(values[index], unicode): i= values[index].encode("utf-8") values[index]=i my_dict.update({data:values[index]}) j=hashabledict() j.update({self.type_cls.items()[0][0]:my_dict}) return j self.caster = Caster attach_composite_listeners() def bind_processor(self, dialect): print 'bind_processor' def process(value): if value is None: return None out='' out=out+ ' '.join('{name}'.format(name=i) for i in self.type_cls.items()[0][1]) mylist=[] for j in value.items(): mylist.append(j[1]) i=collections.namedtuple(self.name,out) return i(*mylist) return process def result_processor(self, dialect, coltype): print 'result_processor' # print self.__dict__ def process(value): print 'my values is'+str(value) return value self.type_cls=value i=self.__class__(name=value.items()[0][0],columns=value.items()[0][1]) # i.set_values(value.items()[0][1],name=value.items()[0][0]) return i return process def __hash__(self): return make_hash(self) def __eq__(self, other): return make_hash(self) == make_hash(other) def __ne__(self, other): return make_hash(self) != make_hash(other) Composite_type = MyCompositeType( 'composite_type', [ sqlalchemy.Column('chislo', sqlalchemy.Integer), sqlalchemy.Column('slovo', sqlalchemy.String) ] ) -- SQLAlchemy - The Python SQL Toolkit and Object Relational Mapper http://www.sqlalchemy.org/ To post example code, please provide an MCVE: Minimal, Complete, and Verifiable Example. See http://stackoverflow.com/help/mcve for a full description. --- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.