On Aug 13, 2010, at 10:24 AM, Dan wrote: > > > On Aug 13, 3:17 pm, Michael Bayer <mike...@zzzcomputing.com> 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 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. > -- 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.