>
> 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.
