On 5/22/15 6:13 AM, Lie Ryan wrote:
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)?
we need to look at the big picture. You have code like this:
obj1 = Object('foo')
obj2 = Object('bar')
obj1.tags['cheese'] = 'inspector'
obj2.tags['cheese'] = 'inspector'
you'd like this to work such that when these string names are assigned,
a database lookup is performed. To do database lookups, we need a
database connection.
Where would a system like the above most naturally get a database
connection from, assuming we don't want a global registry? It would
have to be explicit in some way. What way that is, I don't know, but
here are things that I'd imagine would be candidates:
# candidate #1
with using_session(session):
obj1 = Object('foo')
obj2 = Object('bar')
obj1.tags['cheese'] = 'inspector'
obj2.tags['cheese'] = 'inspector'
or maybe:
# candidate #2
obj1 = Object('foo')
obj2 = Object('bar')
obj1.tags['cheese'] = via_session(session, 'inspector')
obj2.tags['cheese'] = via_session(session, 'inspector')
or perhaps we want to be really slick, and have it work this way:
# candidate #3
obj1 = Object('foo')
obj1.tags['cheese'] = 'inspector' # <-- no SQL lookup
session.add(obj1) # <-- SQL lookup
e.g. "how do i obtain", from the pov of the receiver, is not really the
question here. It's "what am I asking for?" on the part of the caller
we have to decide. It's not reasonable to expect that we would be
able to say:
obj1 = Object('foo')
obj1.tags['cheese'] = 'inspector'
that we then would have a database-persisted Tag object on our Object.
Because we didn't tell it about a database. We shouldn't *want* it to
work that way (again, if we know there's no "global" database set up).
Not sure what approach appeals to you, here is a hint for each:
candidate #1: build a Python context manager that temporarily places a
Session in a global, thread-local registry, which is then consulted
within the "creator"
candidate #2: build a "via_session" "command" object that is received by
the "creator" and is interpreted to include both the session and the key.
candidate #3: build a brand new Tag() object that isn't database
persisted; then use the before_attach event
(http://docs.sqlalchemy.org/en/rel_1_0/orm/events.html?highlight=after_attach#sqlalchemy.orm.events.SessionEvents.before_attach)
to intercept when the parent Object is going into the Session, and then
swap out the Tag object that is present on it for the real persisted one.
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 [email protected]
<mailto:[email protected]>.
To post to this group, send email to [email protected]
<mailto:[email protected]>.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
--
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 [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
