I am currently stuck on creating association proxy. The composite association_proxy ( http://docs.sqlalchemy.org/en/latest/orm/extensions/associationproxy.html#composite-association-proxies) pattern pointed to using UniqueObject recipe ( https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/UniqueObject) to prevent creator from creating duplicate Keyword objects.
How do I obtain the session from inside the creator function when I manage sessions explicitly (i.e. not using scoped_session)? Here is what I got to so far: from sqlalchemy import Column, Integer, String, ForeignKey, Table, create_engine, UniqueConstraint from sqlalchemy.orm import relationship, sessionmaker, backref from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm.collections import attribute_mapped_collection Base = declarative_base() DBSession = sessionmaker() # taken verbatim from https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/UniqueObject def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw): cache = getattr(session, '_unique_cache', None) if cache is None: session._unique_cache = cache = {} key = (cls, hashfunc(*arg, **kw)) if key in cache: return cache[key] else: with session.no_autoflush: q = session.query(cls) q = queryfunc(q, *arg, **kw) obj = q.first() if not obj: obj = constructor(*arg, **kw) session.add(obj) cache[key] = obj return obj def creator(key, value): #return Tag(key=key, value=value) # this creates a Unique Constraint error on Tag table return _unique( session, # NameError: global name 'session' is not defined Tag, lambda key, value: (key, value), lambda query, key, value: query.filter_by(key=key, value=value), Tag, arg=[], kw={'key': key, 'value': value}, ) class Object(Base): __tablename__ = 'obj' id = Column(Integer, primary_key=True) name = Column(String(64)) tags_obj = relationship( 'Tag', secondary='obj_tag', backref='objs', collection_class=attribute_mapped_collection('key'), ) tags = association_proxy( 'tags_obj', 'value', creator=creator, ) def __init__(self, name): self.name = name class ObjectTag(Base): __tablename__ = 'obj_tag' obj_id = Column(Integer, ForeignKey('obj.id'), primary_key=True) tag_id = Column(Integer, ForeignKey('tag.id'), primary_key=True) class Tag(Base): __tablename__ = 'tag' id = Column(Integer, primary_key=True) key = Column('key', String(64)) value = Column('value', String(64)) __table_args__ = ( UniqueConstraint('key', 'value', name=tag_uq'), ) def main(): engine = create_engine('sqlite://') session = DBSession(bind=engine) Base.metadata.drop_all(engine) Base.metadata.create_all(engine) obj1 = Object('foo') obj2 = Object('bar') obj1.tags['cheese'] = 'inspector' obj2.tags['cheese'] = 'inspector' session.add(obj1) session.add(obj2) assert session.query(Tag).count() == 1 # (IntegrityError) UNIQUE constraint failed: tag.key, tag.value session.commit() main() -- 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 http://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.