[sqlalchemy] Re: Automatically execute a procedure on saving/loading from Sql database with SQLAlchemy (and MeGrok)
Wow... I just saw this... http://www.sqlalchemy.org/docs/05/reference/sqlalchemy/types.html#sqlalchemy.types.TypeDecorator Maybe that's what I need! I'll confirm (for future questions) :-) 2010/10/25 Hector Blanco white.li...@gmail.com Hello everyone. First of all, thank you for reading this (no matter whether you can/want to help me or not) :-) Second, the question: I am using sqlalchemy under MeGrok (http://pypi.python.org/pypi/megrok.rdb) to have my Python/Grok classes stored over a RDBMS (MySql) database. I have a Python class in which one of the fields is a list of strings. I don't really find worthy to create a table to relate the class with the fields (is just a list that can take certain names of days of the week) so I was planning to store them in MySql as an string where the values would be separated with a comma (or semicolon). On the other hand, is very useful to have that field as a list (in Python classes) so here's my question: Is there a way to automatically execute an stored procedure (SQL preferably, but I could also do it on the Python side) so when the class is saved, that list field will be automatically joined (with whatever separator character) and when the class (or that field) is loaded, it will be automatically split-ed (so it will come back as a list)? In my brain, I have dreamed about something like an special type of * sqlalchemy.Column* in which you can specify something like *on_save = execute this()* and *on_load = execute that()*... :-) I also asked this very same question in the Grok-dev mail list. They suggested me the use of decorators, which I find an interesting idea, but I don't really know where to put a decorator to ensure that when that field is saved, it's saved as an array, and when it's loaded, it's loaded as a list. Oh, and, for the record, I am a newbie with this Grok over MySql thing so it may not make any sense what I just asked but... I had to try. Thank you in advance. -- You received this message because you are subscribed to the Google Groups sqlalchemy group. To post to this group, send email to sqlalch...@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
[sqlalchemy] Re: Automatically execute a procedure on saving/loading from Sql database with SQLAlchemy (and MeGrok)
Yuuup... It works like a charm! I have created a simple class that gets a dictionary and serializes its values. The underlying database is MySql... Just in case my code may help someone (or if someone has any suggestions...) here it goes: from sqlalchemy import types import logging log = logging.getLogger(__name__) class SimpleDict(types.TypeDecorator): impl = types.String size = -1 __separatorChar = chr(0x1D) __boolPrefix = b_ __intPrefix = i_ __floatPrefix = f_ __nullPrefix = n_ __specialPrefixes = set([__boolPrefix, __intPrefix, __floatPrefix, __nullPrefix]) __nullValues = set([null, None]) def __init__(self, length = 1024): self.size = int(length) super(ZepSimpleDict, self).__init__(self.size) def __toString(self, value): retval = None if isinstance(value, bool): retval = self.__boolPrefix + str(value) elif isinstance(value, float): retval = self.__floatPrefix + str(value) elif isinstance(value, int): retval = self.__intPrefix + str(value) elif (value is None) or (value in self.__nullValues): retval = self.__nullPrefix + str(None) else: retval = str(value) return retval def __fromString(self, value): retval = None prefix = None actualValue = None if len(value) 2: prefix = value[0:2] if (prefix in self.__specialPrefixes): actualValue = value[2:] if prefix == self.__boolPrefix: if actualValue == True: retval = True elif actualValue == False: retval = False else: retval = value elif prefix == self.__floatPrefix: try: retval = float(actualValue) except ValueError: retval = value elif prefix == self.__intPrefix: try: retval = int(actualValue) except ValueError: retval = value elif prefix == self.__nullPrefix: if actualValue == str(None): retval = None else: retval = value else: retval = value else: retval = value return retval def process_bind_param(self, value, dialect): value_tmp = None flattenedValue = list() retval = None if isinstance(value, dict): value_tmp = dict() for key, val in value.iteritems(): value_tmp[self.__toString(key)] = self.__toString(val) else: value_tmp = None if (value_tmp is not None): for key, val in value_tmp.iteritems(): flattenedValue.append(key) flattenedValue.append(val) retval = self.__separatorChar.join(flattenedValue) else: retval = None return retval def process_result_value(self, value, dialect): retval = dict() value_tmp = value.split(self.__separatorChar) if (len(value_tmp) 0): if (len(value_tmp) % 2 != 0): log.warn(process_result_value Processing an string with odd number of elements. This should not have happened.) for i in range(0, len(value_tmp), 2): retval[self.__fromString(value_tmp[i])] = self.__fromString(value_tmp[i+1]) return retval In my previous message, I said: *In my brain, I have dreamed about something like an special type of sqlalchemy.Column in which you can specify something like on_save = execute this() and on_load = execute that()... * This does exactly that! :) -- You received this message because you are subscribed to the Google Groups sqlalchemy group. To post to this group, send email to sqlalch...@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
[sqlalchemy] Re: Automatically execute a procedure on saving/loading from Sql database with SQLAlchemy (and MeGrok)
I hate when the identation gets messed up... Gonna try again: Yuuup... It works like a charm! I have created a simple class that gets a dictionary and serializes its values. The underlying database is MySql... Just in case my code may help someone (or if someone has any suggestions...) here it goes: from sqlalchemy import types import logging log = logging.getLogger(__name__) class ZepSimpleDict(types.TypeDecorator): impl = types.String size = -1 __separatorChar = chr(0x1D) __boolPrefix = b_ __intPrefix = i_ __floatPrefix = f_ __nullPrefix = n_ __specialPrefixes = set([__boolPrefix, __intPrefix, __floatPrefix, __nullPrefix]) __nullValues = set([null, None]) def __init__(self, length = 1024): self.size = int(length) super(ZepSimpleDict, self).__init__(self.size) def __toString(self, value): retval = None if isinstance(value, bool): retval = self.__boolPrefix + str(value) elif isinstance(value, float): retval = self.__floatPrefix + str(value) elif isinstance(value, int): retval = self.__intPrefix + str(value) elif (value is None) or (value in self.__nullValues): retval = self.__nullPrefix + str(None) else: retval = str(value) return retval def __fromString(self, value): retval = None prefix = None actualValue = None if len(value) 2: prefix = value[0:2] if (prefix in self.__specialPrefixes): actualValue = value[2:] if prefix == self.__boolPrefix: if actualValue == True: retval = True elif actualValue == False: retval = False else: retval = value elif prefix == self.__floatPrefix: try: retval = float(actualValue) except ValueError: retval = value elif prefix == self.__intPrefix: try: retval = int(actualValue) except ValueError: retval = value elif prefix == self.__nullPrefix: if actualValue == str(None): retval = None else: retval = value else: retval = value else: retval = value return retval def process_bind_param(self, value, dialect): value_tmp = None flattenedValue = list() retval = None if isinstance(value, dict): value_tmp = dict() for key, val in value.iteritems(): value_tmp[self.__toString(key)] = self.__toString(val) else: value_tmp = None if (value_tmp is not None): for key, val in value_tmp.iteritems(): flattenedValue.append(key) flattenedValue.append(val) retval = self.__separatorChar.join(flattenedValue) else: retval = None return retval def process_result_value(self, value, dialect): retval = dict() value_tmp = value.split(self.__separatorChar) if (len(value_tmp) 0): if (len(value_tmp) % 2 != 0): log.warn(process_result_value Processing an string with odd number of elements. This should not have happened.) for i in range(0, len(value_tmp), 2): retval[self.__fromString(value_tmp[i])] = self.__fromString(value_tmp[i+1]) return retval
