--- /dev/null
+.. change::
+ :tags: usecase, orm
+ :tickets: 5171
+
+ Enhanced logic that tracks if relationships will be conflicting with each
+ other when they write to the same column to include simple cases of two
+ relationships that should have a "backref" between them. This means that
+ if two relationships are not viewonly, are not linked with back_populates
+ and are not otherwise in an inheriting sibling/overriding arrangement, and
+ will populate the same foreign key column, a warning is emitted at mapper
+ configuration time warning that a conflict may arise. A new parameter
+ :paramref:`.relationship.overlaps` is added to suit those very rare cases
+ where such an overlapping persistence arrangement may be unavoidable.
+
return self.base_mapper is other.base_mapper
+ def is_sibling(self, other):
+ """return true if the other mapper is an inheriting sibling to this
+ one. common parent but different branch
+
+ """
+ return (
+ self.base_mapper is other.base_mapper
+ and not self.isa(other)
+ and not other.isa(self)
+ )
+
def _canload(self, state, allow_subtypes):
s = self.primary_mapper()
if self.polymorphic_on is not None or allow_subtypes:
from __future__ import absolute_import
import collections
+import re
import weakref
from . import attributes
order_by=False,
backref=None,
back_populates=None,
+ overlaps=None,
post_update=False,
cascade=False,
viewonly=False,
:paramref:`~.relationship.backref` - alternative form
of backref specification.
+ :param overlaps:
+ A string name or comma-delimited set of names of other relationships
+ on either this mapper, a descendant mapper, or a target mapper with
+ which this relationship may write to the same foreign keys upon
+ persistence. The only effect this has is to eliminate the
+ warning that this relationship will conflict with another upon
+ persistence. This is used for such relationships that are truly
+ capable of conflicting with each other on write, but the application
+ will ensure that no such conflicts occur.
+
+ .. versionadded:: 1.4
+
:param bake_queries=True:
Use the :class:`.BakedQuery` cache to cache the construction of SQL
used in lazy loads. True by default. Set to False if the
self.strategy_key = (("lazy", self.lazy),)
self._reverse_property = set()
+ if overlaps:
+ self._overlaps = set(re.split(r"\s*,\s*", overlaps))
+ else:
+ self._overlaps = ()
if cascade is not False:
self.cascade = cascade
# if multiple relationships overlap foreign() directly, but
# we're going to assume it's typically a ForeignKeyConstraint-
# level configuration that benefits from this warning.
- if len(to_.foreign_keys) < 2:
- continue
if to_ not in self._track_overlapping_sync_targets:
self._track_overlapping_sync_targets[
for pr, fr_ in prop_to_from.items():
if (
pr.mapper in mapperlib._mapper_registry
+ and pr not in self.prop._reverse_property
+ and pr.key not in self.prop._overlaps
+ and self.prop.key not in pr._overlaps
+ and not self.prop.parent.is_sibling(pr.parent)
+ and not self.prop.mapper.is_sibling(pr.mapper)
and (
- self.prop._persists_for(pr.parent)
- or pr._persists_for(self.prop.parent)
+ self.prop.key != pr.key
+ or not self.prop.parent.common_parent(pr.parent)
)
- and fr_ is not from_
- and pr not in self.prop._reverse_property
):
other_props.append((pr, fr_))
util.warn(
"relationship '%s' will copy column %s to column %s, "
"which conflicts with relationship(s): %s. "
- "Consider applying "
- "viewonly=True to read-only relationships, or provide "
- "a primaryjoin condition marking writable columns "
- "with the foreign() annotation."
+ "If this is not the intention, consider if these "
+ "relationships should be linked with "
+ "back_populates, or if viewonly=True should be "
+ "applied to one or more if they are read-only. "
+ "For the less common case that foreign key "
+ "constraints are partially overlapping, the "
+ "orm.foreign() "
+ "annotation can be used to isolate the columns that "
+ "should be written towards. The 'overlaps' "
+ "parameter may be used to remove this warning."
% (
self.prop,
from_,
@declared_attr
def something_else(cls):
counter(cls, "something_else")
- return relationship("Something")
+ return relationship("Something", viewonly=True)
class ConcreteConcreteAbstraction(AbstractConcreteAbstraction):
__tablename__ = "cca"
Column("name", String(50)),
)
+ def teardown(self):
+ clear_mappers()
+
def _fixture(self, collection_class, is_dict=False):
class Parent(object):
collection = association_proxy("_collection", "child")
Keyword,
keywords,
properties={
- "user_keyword": relationship(UserKeyword, uselist=False),
- "user_keywords": relationship(UserKeyword),
+ "user_keyword": relationship(
+ UserKeyword, uselist=False, back_populates="keyword"
+ ),
+ "user_keywords": relationship(UserKeyword, viewonly=True),
},
)
userkeywords,
properties={
"user": relationship(User, backref="user_keywords"),
- "keyword": relationship(Keyword),
+ "keyword": relationship(
+ Keyword, back_populates="user_keyword"
+ ),
},
)
mapper(
data = Column(String(50))
bs = relationship("B")
- b_dyn = relationship("B", lazy="dynamic")
+ b_dyn = relationship("B", lazy="dynamic", viewonly=True)
b_data = association_proxy("bs", "data")
foos = mapper(Foo, foo)
bars = mapper(Bar, bar, inherits=foos)
bars.add_property("lazy", relationship(foos, bar_foo, lazy="select"))
- bars.add_property("eager", relationship(foos, bar_foo, lazy="joined"))
+ bars.add_property(
+ "eager", relationship(foos, bar_foo, lazy="joined", viewonly=True)
+ )
foo.insert().execute(data="foo1")
bar.insert().execute(id=1, data="bar1")
mapper(
Company,
companies,
- properties={"engineers": relationship(Engineer)},
+ properties={
+ "engineers": relationship(Engineer, back_populates="company")
+ },
)
mapper(
Employee,
C1,
t1,
properties={
- "c1s": relationship(C1, cascade="all"),
+ "c1s": relationship(
+ C1, cascade="all", back_populates="parent"
+ ),
"parent": relationship(
C1,
primaryjoin=t1.c.parent_c1 == t1.c.c1,
remote_side=t1.c.c1,
lazy="select",
uselist=False,
+ back_populates="c1s",
),
},
)
super(InheritanceTest, cls).setup_mappers()
from sqlalchemy import inspect
- inspect(Company).add_property("managers", relationship(Manager))
+ inspect(Company).add_property(
+ "managers", relationship(Manager, viewonly=True)
+ )
def test_load_only_subclass(self):
s = Session()
),
lazy="joined",
order_by=open_mapper.id,
+ viewonly=True,
),
closed_orders=relationship(
closed_mapper,
),
lazy="joined",
order_by=closed_mapper.id,
+ viewonly=True,
),
),
)
),
lazy="joined",
order_by=orders.c.id,
+ viewonly=True,
),
closed_orders=relationship(
Order,
),
lazy="joined",
order_by=orders.c.id,
+ viewonly=True,
),
),
)
+
self._run_double_test()
def _run_double_test(self, no_items=False):
b_np = aliased(B, weird_selectable, flat=True)
a_mapper = inspect(A)
- a_mapper.add_property("bs_np", relationship(b_np))
+ a_mapper.add_property("bs_np", relationship(b_np, viewonly=True))
s = Session()
orders.c.isopen == 1, users.c.id == orders.c.user_id
),
lazy="select",
+ viewonly=True,
),
closed_orders=relationship(
Order,
orders.c.isopen == 0, users.c.id == orders.c.user_id
),
lazy="select",
+ viewonly=True,
),
),
)
from sqlalchemy import util
from sqlalchemy.orm import attributes
from sqlalchemy.orm import class_mapper
+from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import create_session
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import mapper
session.add(b)
assert a in session, "base is %s" % base
+ clear_mappers()
+
def test_compileonattr_rel_backref_b(self):
m = MetaData()
t1 = Table(
session = create_session()
session.add(a)
assert b in session, "base: %s" % base
+ clear_mappers()
users.c.id == open_mapper.user_id,
),
lazy="select",
+ overlaps="closed_orders",
),
closed_orders=relationship(
closed_mapper,
users.c.id == closed_mapper.user_id,
),
lazy="select",
+ overlaps="open_orders",
),
),
)
inherits=Address,
properties={
"sub_attr": column_property(address_table.c.email_address),
- "dings": relationship(Dingaling),
+ "dings": relationship(Dingaling, viewonly=True),
},
)
mapper(
SubAddr,
inherits=Address,
- properties={"flub": relationship(Dingaling)},
+ properties={"flub": relationship(Dingaling, viewonly=True)},
)
q = sess.query(Address)
mapper(
SubAddr,
inherits=Address,
- properties={"flub": relationship(Dingaling)},
+ properties={"flub": relationship(Dingaling, viewonly=True)},
)
q = sess.query(SubAddr)
mapper(
SubAddr,
inherits=Address,
- properties={"flub": relationship(Dingaling)},
+ properties={"flub": relationship(Dingaling, viewonly=True)},
)
q = sess.query(Address)
mapper(
SubAddr,
inherits=Address,
- properties={"flub": relationship(Dingaling)},
+ properties={"flub": relationship(Dingaling, viewonly=True)},
)
q = sess.query(User)
users.c.id == addresses.c.user_id,
users.c.name == addresses.c.email_address,
),
+ viewonly=True,
),
},
)
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import aliased
from sqlalchemy.orm import attributes
from sqlalchemy.orm import backref
from sqlalchemy.orm import clear_mappers
add_b_amember=False,
add_bsub1_a=False,
add_bsub2_a_viewonly=False,
+ add_b_a_overlaps=None,
):
Base = declarative_base(metadata=self.metadata)
# writes to B.a_id, which conflicts with BSub2.a_member,
# so should warn
if add_b_a:
- a = relationship("A", viewonly=add_b_a_viewonly)
+ a = relationship(
+ "A", viewonly=add_b_a_viewonly, overlaps=add_b_a_overlaps
+ )
# if added, this relationship writes to B.a_id, which conflicts
# with BSub1.a
return A, AMember, B, BSub1, BSub2
+ def _fixture_two(self, setup_backrefs=False, setup_overlaps=False):
+
+ Base = declarative_base(metadata=self.metadata)
+
+ # purposely using the comma to make sure parsing the comma works
+
+ class Parent(Base):
+ __tablename__ = "parent"
+ id = Column(Integer, primary_key=True)
+ children = relationship(
+ "Child",
+ back_populates=("parent" if setup_backrefs else None),
+ overlaps="foo, bar, parent" if setup_overlaps else None,
+ )
+
+ class Child(Base):
+ __tablename__ = "child"
+ id = Column(Integer, primary_key=True)
+ num = Column(Integer)
+ parent_id = Column(
+ Integer, ForeignKey("parent.id"), nullable=False
+ )
+ parent = relationship(
+ "Parent",
+ back_populates=("children" if setup_backrefs else None),
+ overlaps="bar, bat, children" if setup_overlaps else None,
+ )
+
+ configure_mappers()
+
+ def _fixture_three(self, use_same_mappers, setup_overlaps):
+ Base = declarative_base(metadata=self.metadata)
+
+ class Child(Base):
+ __tablename__ = "child"
+ id = Column(Integer, primary_key=True)
+ num = Column(Integer)
+ parent_id = Column(
+ Integer, ForeignKey("parent.id"), nullable=False
+ )
+
+ if not use_same_mappers:
+ c1 = aliased(Child)
+ c2 = aliased(Child)
+
+ class Parent(Base):
+ __tablename__ = "parent"
+ id = Column(Integer, primary_key=True)
+ if use_same_mappers:
+ child1 = relationship(
+ Child,
+ primaryjoin=lambda: and_(
+ Child.parent_id == Parent.id, Child.num == 1
+ ),
+ overlaps="child2" if setup_overlaps else None,
+ )
+ child2 = relationship(
+ Child,
+ primaryjoin=lambda: and_(
+ Child.parent_id == Parent.id, Child.num == 2
+ ),
+ overlaps="child1" if setup_overlaps else None,
+ )
+ else:
+ child1 = relationship(
+ c1,
+ primaryjoin=lambda: and_(
+ c1.parent_id == Parent.id, c1.num == 1
+ ),
+ overlaps="child2" if setup_overlaps else None,
+ )
+
+ child2 = relationship(
+ c2,
+ primaryjoin=lambda: and_(
+ c2.parent_id == Parent.id, c2.num == 1
+ ),
+ overlaps="child1" if setup_overlaps else None,
+ )
+
+ configure_mappers()
+
@testing.provide_metadata
def _test_fixture_one_run(self, **kw):
A, AMember, B, BSub1, BSub2 = self._fixture_one(**kw)
session.commit()
assert bsub1.a is a2 # because bsub1.a_member is not a relationship
+
+ assert BSub2.__mapper__.attrs.a.viewonly
assert bsub2.a is a1 # because bsub2.a is viewonly=True
# everyone has a B.a relationship
[(bsub1, a2), (bsub2, a1)],
)
+ @testing.provide_metadata
+ def test_simple_warn(self):
+ assert_raises_message(
+ exc.SAWarning,
+ r"relationship '(?:Child.parent|Parent.children)' will copy "
+ r"column parent.id to column child.parent_id, which conflicts "
+ r"with relationship\(s\): '(?:Parent.children|Child.parent)' "
+ r"\(copies parent.id to child.parent_id\).",
+ self._fixture_two,
+ setup_backrefs=False,
+ )
+
+ @testing.provide_metadata
+ def test_simple_backrefs_works(self):
+ self._fixture_two(setup_backrefs=True)
+
+ @testing.provide_metadata
+ def test_simple_overlaps_works(self):
+ self._fixture_two(setup_overlaps=True)
+
+ @testing.provide_metadata
+ def test_double_rel_same_mapper_warns(self):
+ assert_raises_message(
+ exc.SAWarning,
+ r"relationship 'Parent.child[12]' will copy column parent.id to "
+ r"column child.parent_id, which conflicts with relationship\(s\): "
+ r"'Parent.child[12]' \(copies parent.id to child.parent_id\)",
+ self._fixture_three,
+ use_same_mappers=True,
+ setup_overlaps=False,
+ )
+
+ @testing.provide_metadata
+ def test_double_rel_same_mapper_overlaps_works(self):
+ self._fixture_three(use_same_mappers=True, setup_overlaps=True)
+
+ @testing.provide_metadata
+ def test_double_rel_aliased_mapper_works(self):
+ self._fixture_three(use_same_mappers=False, setup_overlaps=False)
+
@testing.provide_metadata
def test_warn_one(self):
assert_raises_message(
add_b_a=True,
)
+ @testing.provide_metadata
+ def test_warn_four(self):
+ assert_raises_message(
+ exc.SAWarning,
+ r"relationship '(?:B.a|BSub2.a_member|B.a)' will copy column "
+ r"(?:a.id|a_member.a_id) to column b.a_id",
+ self._fixture_one,
+ add_bsub2_a_viewonly=True,
+ add_b_a=True,
+ )
+
@testing.provide_metadata
def test_works_one(self):
self._test_fixture_one_run(
@testing.provide_metadata
def test_works_two(self):
- self._test_fixture_one_run(add_b_a=True, add_bsub2_a_viewonly=True)
+ # doesn't actually work with real FKs beacuse it creates conflicts :)
+ self._fixture_one(
+ add_b_a=True, add_b_a_overlaps="a_member", add_bsub1_a=True
+ )
class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
Address, lazy="selectin", order_by=addresses.c.id
),
open_orders=relationship(
- open_mapper, lazy="selectin", order_by=open_mapper.id
+ open_mapper,
+ lazy="selectin",
+ order_by=open_mapper.id,
+ overlaps="closed_orders",
),
closed_orders=relationship(
- closed_mapper, lazy="selectin", order_by=closed_mapper.id
+ closed_mapper,
+ lazy="selectin",
+ order_by=closed_mapper.id,
+ overlaps="open_orders",
),
),
)
),
lazy="selectin",
order_by=open_mapper.id,
+ viewonly=True,
),
closed_orders=relationship(
closed_mapper,
),
lazy="selectin",
order_by=closed_mapper.id,
+ viewonly=True,
),
),
)
),
lazy="selectin",
order_by=orders.c.id,
+ overlaps="closed_orders",
),
closed_orders=relationship(
Order,
),
lazy="selectin",
order_by=orders.c.id,
+ overlaps="open_orders",
),
),
)
id = Column(Integer, primary_key=True)
b_id = Column(ForeignKey("b.id"))
b = relationship("B")
- b_no_omit_join = relationship("B", omit_join=False)
+ b_no_omit_join = relationship("B", omit_join=False, overlaps="b")
q = Column(Integer)
class B(fixtures.ComparableEntity, Base):
),
lazy="subquery",
order_by=open_mapper.id,
+ overlaps="closed_orders",
),
closed_orders=relationship(
closed_mapper,
),
lazy="subquery",
order_by=closed_mapper.id,
+ overlaps="open_orders",
),
),
)
),
lazy="subquery",
order_by=orders.c.id,
+ viewonly=True,
),
closed_orders=relationship(
Order,
),
lazy="subquery",
order_by=orders.c.id,
+ viewonly=True,
),
),
)
users.c.id == addresses.c.user_id,
addresses.c.email_address.like("%boston%"),
),
+ overlaps="newyork_addresses",
),
"newyork_addresses": relationship(
m2,
users.c.id == addresses.c.user_id,
addresses.c.email_address.like("%newyork%"),
),
+ overlaps="boston_addresses",
),
},
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
