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
------------------------------------------------------------------------------------------------
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 [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
