Am Samstag, 10. September 2016 09:02:32 UTC+2 schrieb Dominik George:
>
> Hi Mike,
>
> here it is. See attached mwe.py.
>
Which is here ;).
-nik
--
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 https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
from sqlalchemy import create_engine, Column, ForeignKey, Integer, Unicode
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import backref, relationship, sessionmaker
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm import sessionmaker, scoped_session
def monkey_patch_sqlalchemy():
from sqlalchemy.ext.associationproxy import AssociationProxy
from sqlalchemy.util import memoized_property
# Monkey patch support for chained association proxy queries into SQLAlchemy
# https://bitbucket.org/zzzeek/sqlalchemy/issues/3769/chained-any-has-with-association-proxy
if not hasattr(AssociationProxy, "_unwrap_target_assoc_proxy"):
def _unwrap_target_assoc_proxy(self):
attr = getattr(self.target_class, self.value_attr)
if isinstance(attr, AssociationProxy):
return attr, getattr(self.target_class, attr.target_collection)
return None, None
AssociationProxy._unwrap_target_assoc_proxy = memoized_property(_unwrap_target_assoc_proxy)
orig_any = AssociationProxy.any
def any_(self, criterion=None, **kwargs):
target_assoc, inner = self._unwrap_target_assoc_proxy
if target_assoc is not None:
if target_assoc._target_is_object and target_assoc._uselist:
inner = inner.any(criterion=criterion, **kwargs)
else:
inner = inner.has(criterion=criterion, **kwargs)
return self._comparator.any(inner)
orig_any(self, criterion, **kwargs)
AssociationProxy.any = any_
orig_has = AssociationProxy.has
def has(self, criterion=None, **kwargs):
target_assoc, inner = self._unwrap_target_assoc_proxy
if target_assoc is not None:
if target_assoc._target_is_object and target_assoc._uselist:
inner = inner.any(criterion=criterion, **kwargs)
else:
inner = inner.has(criterion=criterion, **kwargs)
return self._comparator.has(inner)
orig_has(self, criterion, **kwargs)
AssociationProxy.has = has
monkey_patch_sqlalchemy()
engine = create_engine("sqlite:///:memory:")
base = declarative_base(bind=engine)
session = scoped_session(sessionmaker(bind=engine))
class Tag(base):
__tablename__ = "tags"
tag_id = Column(Integer, primary_key=True)
key = Column(Unicode(256))
value = Column(Unicode(256))
def __init__(self, key="", value="", **kwargs):
self.key = key
self.value = value
base.__init__(self, **kwargs)
class Element(base):
__tablename__ = "elements"
element_id = Column(Integer, primary_key=True)
tags = association_proxy("elements_tags", "tag_value",
creator=lambda k, v: ElementsTags(tag_key=k, tag_value=v))
class ElementsTags(base):
__tablename__ = "elements_tags"
map_id = Column(Integer, primary_key=True)
element_id = Column(Integer, ForeignKey('elements.element_id'))
tag_id = Column(Integer, ForeignKey('tags.tag_id'))
element = relationship(Element, foreign_keys=[element_id],
backref=backref("elements_tags",
collection_class=attribute_mapped_collection("tag_key"),
cascade="all, delete-orphan"))
tag = relationship(Tag, foreign_keys=[tag_id])
tag_key = association_proxy("tag", "key")
tag_value = association_proxy("tag", "value")
base.metadata.create_all()
e = Element()
e.tags[u"foo"] = u"bar"
e.tags[u"bang"] = u"baz"
session.add(e)
session.commit()
e = session.query(Element).one()
foo = e.tags[u"foo"]
bang = e.tags[u"bang"]
print("foo = %s , bang = %s" % (foo, bang))
