On Aug 13, 2010, at 10:24 AM, Dan wrote:
>
>
> On Aug 13, 3:17 pm, Michael Bayer <[email protected]> wrote:
>> On Aug 13, 2010, at 10:01 AM, Dan wrote:
>>
>>> I have created a custom type in order to store denormalized PKs in a
>>> TEXT field. The idea is that the text is converted back and forth from
>>> a set of integers:
>>
>>> http://paste.pocoo.org/show/249784/
>>
>> this is unrelated, but the code is incorrect there, should be
>>
>> def process(value):
>> if value is not None:
>> items = [str(item).strip() for item in value]
>> value = self.separator.join(item for item in items if item)
>>
>> otherwise, you must implement copy_value() on your type. Here, the value
>> isn't being copied so there's nothing to compare to.
>>
>
> Yes, sorry for the typo. Realized myself once I'd posted.
>
>> Usually you're supposed to mixin MutableType which will raise notimplemented
>> for copy_value(). I guess still more docs are needed since you were misled
>> by the is_mutable() method.
>>
>
> I've tried the same thing with the MutableType mixin with the same
> result, i.e:
>
> class DenormalizedText(types.TypeDecorator, types.MutableType):
MutableType would be first. But again this only just so the
NotImplementedError lets you know copy_value() is needed. I could make the
default copy_value() raise if is_mutable() is true...though it pains me to add
more method calls...
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import *
Base = declarative_base()
metadata = Base.metadata
engine = create_engine('sqlite://', echo=True)
from sqlalchemy import types
class DenormalizedText(types.TypeDecorator):
"""
Stores denormalized primary keys that can be
accessed as a set.
:param coerce: coercion function that ensures correct
type is returned
:param separator: separator character
"""
impl = types.Text
def __init__(self, coerce=int, separator=" ", **kwargs):
self.coerce = coerce
self.separator = separator
super(DenormalizedText, self).__init__(**kwargs)
def bind_processor(self, dialect):
def process(value):
if value is not None:
items = [str(item).strip() for item in value]
value = self.separator.join(item for item in items if item)
return value
return process
def result_processor(self, dialect, coltype):
def process(value):
if not value:
return set()
return set(self.coerce(item) \
for item in value.split(self.separator))
return process
def copy_value(self, value):
return set(value)
def is_mutable(self):
return True
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True)
votes = Column(DenormalizedText)
def __init__(self, *args, **kwargs):
super(Post, self).__init__(*args, **kwargs)
self.votes = self.votes or set()
Base.metadata.create_all(engine)
session = sessionmaker(engine)()
post = Post()
post.votes.add(3)
session.add(post)
session.commit()
print "-----------------------"
post.votes.add(5)
session.commit()
>
> --
> 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.
>
--
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.
