[sqlalchemy] Preventing duplicate entries for an association proxy

[Lie Ryan]

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.