return True
def _generate_backref(self):
+ """Interpret the 'backref' instruction to create a
+ :func:`.relationship` complementary to this one."""
+
if not self.is_primary():
return
if self.backref is not None and not self.back_populates:
"'%s' on relationship '%s': property of that "
"name exists on mapper '%s'" % (backref_key,
self, mapper))
+
+ # determine primaryjoin/secondaryjoin for the
+ # backref. Use the one we had, so that
+ # a custom join doesn't have to be specified in
+ # both directions.
if self.secondary is not None:
+ # for many to many, just switch primaryjoin/
+ # secondaryjoin.
pj = kwargs.pop('primaryjoin', self.secondaryjoin)
sj = kwargs.pop('secondaryjoin', self.primaryjoin)
else:
sj = kwargs.pop('secondaryjoin', None)
if sj:
raise sa_exc.InvalidRequestError(
- "Can't assign 'secondaryjoin' on a backref against "
- "a non-secondary relationship."
- )
+ "Can't assign 'secondaryjoin' on a backref "
+ "against a non-secondary relationship."
+ )
foreign_keys = kwargs.pop('foreign_keys',
self._user_defined_foreign_keys)
from sqlalchemy import sql, util, log, exc as sa_exc, schema
from sqlalchemy.sql.util import ClauseAdapter, criterion_as_pairs, \
- join_condition, _shallow_annotate, visit_binary_product
+ join_condition, _shallow_annotate, visit_binary_product,\
+ _deep_deannotate
from sqlalchemy.sql import operators, expression, visitors
from sqlalchemy.orm.interfaces import MANYTOMANY, MANYTOONE, ONETOMANY
@util.memoized_property
def primaryjoin_reverse_remote(self):
- def replace(element):
- if "remote" in element._annotations:
- v = element._annotations.copy()
- del v['remote']
- v['local'] = True
- return element._with_annotations(v)
- elif "local" in element._annotations:
- v = element._annotations.copy()
- del v['local']
- v['remote'] = True
- return element._with_annotations(v)
- return visitors.replacement_traverse(self.primaryjoin, {}, replace)
+ """Return the primaryjoin condition suitable for the
+ "reverse" direction.
+
+ If the primaryjoin was delivered here with pre-existing
+ "remote" annotations, the local/remote annotations
+ are reversed. Otherwise, the local/remote annotations
+ are removed.
+
+ """
+ if self._has_remote_annotations:
+ def replace(element):
+ if "remote" in element._annotations:
+ v = element._annotations.copy()
+ del v['remote']
+ v['local'] = True
+ return element._with_annotations(v)
+ elif "local" in element._annotations:
+ v = element._annotations.copy()
+ del v['local']
+ v['remote'] = True
+ return element._with_annotations(v)
+ return visitors.replacement_traverse(
+ self.primaryjoin, {}, replace)
+ else:
+ if self._has_foreign_annotations:
+ # TODO: coverage
+ return _deep_deannotate(self.primaryjoin,
+ values=("local", "remote"))
+ else:
+ return _deep_deannotate(self.primaryjoin)
def _has_annotation(self, clause, annotation):
for col in visitors.iterate(clause, {}):
else:
return False
+ @util.memoized_property
+ def _has_foreign_annotations(self):
+ return self._has_annotation(self.primaryjoin, "foreign")
+
+ @util.memoized_property
+ def _has_remote_annotations(self):
+ return self._has_annotation(self.primaryjoin, "remote")
+
def _annotate_fks(self):
- if self._has_annotation(self.primaryjoin, "foreign"):
+ """Annotate the primaryjoin and secondaryjoin
+ structures with 'foreign' annotations marking columns
+ considered as foreign.
+
+ """
+ if self._has_foreign_annotations:
return
if self.consider_as_foreign_keys:
)
def _refers_to_parent_table(self):
+ """Return True if the join condition contains column
+ comparisons where both columns are in both tables.
+
+ """
pt = self.parent_selectable
mt = self.child_selectable
result = [False]
mt.is_derived_from(f.table)
):
result[0] = True
-
visitors.traverse(
self.primaryjoin,
{},
)
return result[0]
+ def _tables_overlap(self):
+ """Return True if parent/child tables have some overlap."""
+
+ return self.parent_selectable.is_derived_from(
+ self.child_local_selectable) or \
+ self.child_selectable.is_derived_from(
+ self.parent_local_selectable)
+
def _annotate_remote(self):
- if self._has_annotation(self.primaryjoin, "remote"):
+ """Annotate the primaryjoin and secondaryjoin
+ structures with 'remote' annotations marking columns
+ considered as part of the 'remote' side.
+
+ """
+ if self._has_remote_annotations:
return
parentcols = util.column_set(self.parent_selectable.c)
- def _annotate_selfref(fn):
- def visit_binary(binary):
- equated = binary.left.compare(binary.right)
- if isinstance(binary.left, sql.ColumnElement) and \
- isinstance(binary.right, sql.ColumnElement):
- # assume one to many - FKs are "remote"
- if fn(binary.left):
- binary.left = binary.left._annotate({"remote":True})
- if fn(binary.right) and \
- not equated:
- binary.right = binary.right._annotate(
- {"remote":True})
-
- self.primaryjoin = visitors.cloned_traverse(
- self.primaryjoin, {},
- {"binary":visit_binary})
-
if self.secondary is not None:
- def repl(element):
- if self.secondary.c.contains_column(element):
- return element._annotate({"remote":True})
- self.primaryjoin = visitors.replacement_traverse(
- self.primaryjoin, {}, repl)
- self.secondaryjoin = visitors.replacement_traverse(
- self.secondaryjoin, {}, repl)
+ self._annotate_remote_secondary()
elif self._local_remote_pairs or self._remote_side:
+ self._annotate_remote_from_args()
+ elif self._refers_to_parent_table():
+ self._annotate_selfref(lambda col:"foreign" in col._annotations)
+ elif self._tables_overlap():
+ self._annotate_remote_with_overlap()
+ else:
+ self._annotate_remote_distinct_selectables()
- if self._local_remote_pairs:
- if self._remote_side:
- raise sa_exc.ArgumentError(
- "remote_side argument is redundant "
- "against more detailed _local_remote_side "
- "argument.")
-
- remote_side = [r for (l, r) in self._local_remote_pairs]
+ def _annotate_remote_secondary(self):
+ """annotate 'remote' in primaryjoin, secondaryjoin
+ when 'secondary' is present.
+
+ """
+ def repl(element):
+ if self.secondary.c.contains_column(element):
+ return element._annotate({"remote":True})
+ self.primaryjoin = visitors.replacement_traverse(
+ self.primaryjoin, {}, repl)
+ self.secondaryjoin = visitors.replacement_traverse(
+ self.secondaryjoin, {}, repl)
+
+ def _annotate_selfref(self, fn):
+ """annotate 'remote' in primaryjoin, secondaryjoin
+ when the relationship is detected as self-referential.
+
+ """
+ def visit_binary(binary):
+ equated = binary.left.compare(binary.right)
+ if isinstance(binary.left, expression.ColumnClause) and \
+ isinstance(binary.right, expression.ColumnClause):
+ # assume one to many - FKs are "remote"
+ if fn(binary.left):
+ binary.left = binary.left._annotate({"remote":True})
+ if fn(binary.right) and \
+ not equated:
+ binary.right = binary.right._annotate(
+ {"remote":True})
else:
- remote_side = self._remote_side
+ self._warn_non_column_elements()
- if self._refers_to_parent_table():
- _annotate_selfref(lambda col:col in remote_side)
- else:
- def repl(element):
- if element in remote_side:
- return element._annotate({"remote":True})
- self.primaryjoin = visitors.replacement_traverse(
- self.primaryjoin, {}, repl)
- elif self._refers_to_parent_table():
- _annotate_selfref(lambda col:"foreign" in col._annotations)
+ self.primaryjoin = visitors.cloned_traverse(
+ self.primaryjoin, {},
+ {"binary":visit_binary})
+
+ def _annotate_remote_from_args(self):
+ """annotate 'remote' in primaryjoin, secondaryjoin
+ when the 'remote_side' or '_local_remote_pairs'
+ arguments are used.
+
+ """
+ if self._local_remote_pairs:
+ if self._remote_side:
+ raise sa_exc.ArgumentError(
+ "remote_side argument is redundant "
+ "against more detailed _local_remote_side "
+ "argument.")
+
+ remote_side = [r for (l, r) in self._local_remote_pairs]
+ else:
+ remote_side = self._remote_side
+
+ if self._refers_to_parent_table():
+ self._annotate_selfref(lambda col:col in remote_side)
else:
def repl(element):
- if self.child_selectable.c.contains_column(element) and \
- (
- not self.parent_local_selectable.c.contains_column(element)
- or self.child_local_selectable.c.contains_column(element)
- ):
+ if element in remote_side:
return element._annotate({"remote":True})
-
self.primaryjoin = visitors.replacement_traverse(
self.primaryjoin, {}, repl)
+ def _annotate_remote_with_overlap(self):
+ """annotate 'remote' in primaryjoin, secondaryjoin
+ when the parent/child tables have some set of
+ tables in common, though is not a fully self-referential
+ relationship.
+
+ """
+ def visit_binary(binary):
+ binary.left, binary.right = proc_left_right(binary.left,
+ binary.right)
+ binary.right, binary.left = proc_left_right(binary.right,
+ binary.left)
+ def proc_left_right(left, right):
+ if isinstance(left, expression.ColumnClause) and \
+ isinstance(right, expression.ColumnClause):
+ if self.child_selectable.c.contains_column(right) and \
+ self.parent_selectable.c.contains_column(left):
+ right = right._annotate({"remote":True})
+ else:
+ self._warn_non_column_elements()
+
+ return left, right
+
+ self.primaryjoin = visitors.cloned_traverse(
+ self.primaryjoin, {},
+ {"binary":visit_binary})
+
+ def _annotate_remote_distinct_selectables(self):
+ """annotate 'remote' in primaryjoin, secondaryjoin
+ when the parent/child tables are entirely
+ separate.
+
+ """
+ def repl(element):
+ if self.child_selectable.c.contains_column(element) and \
+ (
+ not self.parent_local_selectable.c.\
+ contains_column(element)
+ or self.child_local_selectable.c.\
+ contains_column(element)
+ ):
+ return element._annotate({"remote":True})
+ self.primaryjoin = visitors.replacement_traverse(
+ self.primaryjoin, {}, repl)
+
+ def _warn_non_column_elements(self):
+ util.warn(
+ "Non-simple column elements in primary "
+ "join condition for property %s - consider using "
+ "remote() annotations to mark the remote side."
+ % self.prop
+ )
+
def _annotate_local(self):
+ """Annotate the primaryjoin and secondaryjoin
+ structures with 'local' annotations.
+
+ This annotates all column elements found
+ simultaneously in the parent table
+ and the join condition that don't have a
+ 'remote' annotation set up from
+ _annotate_remote() or user-defined.
+
+ """
if self._has_annotation(self.primaryjoin, "local"):
return
if not self.local_remote_pairs:
raise sa_exc.ArgumentError('Relationship %s could '
'not determine any local/remote column '
- 'pairs from remote side argument %r'
- % (self.prop, self._remote_side))
+ 'pairs.'
+ % (self.prop, ))
def _check_foreign_cols(self, join_condition, primary):
"""Check the foreign key columns collected and emit error messages."""
err += "Ensure that referencing columns are associated with a "\
"a ForeignKey or ForeignKeyConstraint, or are annotated "\
"in the join condition with the foreign() annotation."
+ raise sa_exc.ArgumentError(err)
def _determine_direction(self):
"""Determine if this relationship is one to many, many to one,
self.can_be_synced_fn(right):
lrp.add((right, left))
if binary.operator is operators.eq:
- # and \
- #binary.left.compare(left) and \
- #binary.right.compare(right):
if "foreign" in right._annotations:
collection.append((left, right))
elif "foreign" in left._annotations:
self.synchronize_pairs = sync_pairs
self.secondary_synchronize_pairs = secondary_sync_pairs
-
@util.memoized_property
def remote_columns(self):
return self._gather_join_annotations("remote")
target_adapter.exclude_fn = None
else:
target_adapter = None
- return primaryjoin, secondaryjoin, secondary, target_adapter, dest_selectable
+ return primaryjoin, secondaryjoin, secondary, \
+ target_adapter, dest_selectable
################# everything below is TODO ################################
-def _criterion_exists(self, criterion=None, **kwargs):
- if getattr(self, '_of_type', None):
- target_mapper = self._of_type
- to_selectable = target_mapper._with_polymorphic_selectable
- if self.property._is_self_referential:
- to_selectable = to_selectable.alias()
-
- single_crit = target_mapper._single_table_criterion
- if single_crit is not None:
- if criterion is not None:
- criterion = single_crit & criterion
- else:
- criterion = single_crit
- else:
- to_selectable = None
-
- if self.adapter:
- source_selectable = self.__clause_element__()
- else:
- source_selectable = None
-
- pj, sj, source, dest, secondary, target_adapter = \
- self.property._create_joins(dest_polymorphic=True,
- dest_selectable=to_selectable,
- source_selectable=source_selectable)
-
- for k in kwargs:
- crit = getattr(self.property.mapper.class_, k) == kwargs[k]
- if criterion is None:
- criterion = crit
- else:
- criterion = criterion & crit
-
- # annotate the *local* side of the join condition, in the case
- # of pj + sj this is the full primaryjoin, in the case of just
- # pj its the local side of the primaryjoin.
- if sj is not None:
- j = _orm_annotate(pj) & sj
- else:
- j = _orm_annotate(pj, exclude=self.property.remote_side)
-
- # MARKMARK
- if criterion is not None and target_adapter:
- # limit this adapter to annotated only?
- criterion = target_adapter.traverse(criterion)
-
- # only have the "joined left side" of what we
- # return be subject to Query adaption. The right
- # side of it is used for an exists() subquery and
- # should not correlate or otherwise reach out
- # to anything in the enclosing query.
- if criterion is not None:
- criterion = criterion._annotate({'no_replacement_traverse': True})
-
- crit = j & criterion
-
- return sql.exists([1], crit, from_obj=dest).\
- correlate(source._annotate({'_orm_adapt':True}))
-
-
-def __negated_contains_or_equals(self, other):
- if self.property.direction == MANYTOONE:
- state = attributes.instance_state(other)
-
- def state_bindparam(x, state, col):
- o = state.obj() # strong ref
- return sql.bindparam(x, unique=True, callable_=lambda : \
- self.property.mapper._get_committed_attr_by_column(o,
- col))
-
- def adapt(col):
- if self.adapter:
- return self.adapter(col)
- else:
- return col
-
- if self.property._use_get:
- return sql.and_(*[
- sql.or_(
- adapt(x) != state_bindparam(adapt(x), state, y),
- adapt(x) == None)
- for (x, y) in self.property.local_remote_pairs])
-
- criterion = sql.and_(*[x==y for (x, y) in
- zip(
- self.property.mapper.primary_key,
- self.property.\
- mapper.\
- primary_key_from_instance(other))
- ])
- return ~self._criterion_exists(criterion)
"""
return sqlutil.Annotated(self, values)
- def _deannotate(self, values=None):
+ def _deannotate(self, values=None, clone=False):
"""return a copy of this :class:`.ClauseElement` with annotations
removed.
to remove.
"""
- # since we have no annotations we return
- # self
- return self
+ if clone:
+ # clone is used when we are also copying
+ # the expression for a deep deannotation
+ return self._clone()
+ else:
+ # if no clone, since we have no annotations we return
+ # self
+ return self
def unique_params(self, *optionaldict, **kwargs):
"""Return a copy with :func:`bindparam()` elments replaced.
clone._annotations = values
return clone
- def _deannotate(self, values=None):
+ def _deannotate(self, values=None, clone=True):
if values is None:
return self.__element
else:
"""Deep copy the given element, removing annotations."""
def clone(elem):
- elem = elem._deannotate(values=values)
+ elem = elem._deannotate(values=values, clone=True)
elem._copy_internals(clone=clone)
return elem
if elem in stop_on:
return elem
else:
- if elem not in cloned:
- cloned[elem] = newelem = elem._clone()
+ if id(elem) not in cloned:
+ cloned[id(elem)] = newelem = elem._clone()
newelem._copy_internals(clone=clone)
meth = visitors.get(newelem.__visit_name__, None)
if meth:
meth(newelem)
- return cloned[elem]
+ return cloned[id(elem)]
if obj is not None:
obj = clone(obj)
assert User.x.property.columns[0] is not expr
assert User.x.property.columns[0].element.left is users.c.name
- # a full deannotate goes back to the original element
- assert User.x.property.columns[0].element.right is expr.right
+ # a deannotate needs to clone the base, in case
+ # the original one referenced annotated elements.
+ assert User.x.property.columns[0].element.right is not expr.right
assert User.y.property.columns[0] is not expr2
assert User.y.property.columns[0].element.\
from test.lib import fixtures
from sqlalchemy.orm import relationships, foreign, remote, remote_foreign
from sqlalchemy import MetaData, Table, Column, ForeignKey, Integer, \
- select, ForeignKeyConstraint, exc, func
+ select, ForeignKeyConstraint, exc, func, and_
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
**kw
)
+ def _join_fixture_o2m_composite_selfref_func(self, **kw):
+ return relationships.JoinCondition(
+ self.composite_selfref,
+ self.composite_selfref,
+ self.composite_selfref,
+ self.composite_selfref,
+ primaryjoin=and_(
+ self.composite_selfref.c.group_id==
+ func.foo(self.composite_selfref.c.group_id),
+ self.composite_selfref.c.parent_id==
+ self.composite_selfref.c.id
+ ),
+ **kw
+ )
+
+ def _join_fixture_o2m_composite_selfref_func_annotated(self, **kw):
+ return relationships.JoinCondition(
+ self.composite_selfref,
+ self.composite_selfref,
+ self.composite_selfref,
+ self.composite_selfref,
+ primaryjoin=and_(
+ remote(self.composite_selfref.c.group_id)==
+ func.foo(self.composite_selfref.c.group_id),
+ remote(self.composite_selfref.c.parent_id)==
+ self.composite_selfref.c.id
+ ),
+ **kw
+ )
+
def _join_fixture_compound_expression_1(self, **kw):
return relationships.JoinCondition(
self.left,
self.right_w_base_rel,
)
- User, users = self.classes.User, self.tables.users
- Address, addresses = self.classes.Address, self.tables.addresses
- class SubUser(User):
- pass
- m = mapper(User, users)
- m2 = mapper(SubUser, addresses, inherits=User)
- m3 = mapper(Address, addresses, properties={
- 'foo':relationship(m2)
- })
+ def _join_fixture_m2o_sub_to_joined_sub_func(self, **kw):
+ # see test.orm.test_mapper:MapperTest.test_add_column_prop_deannotate,
+ right = self.base.join(self.right_w_base_rel,
+ self.base.c.id==self.right_w_base_rel.c.id)
+ return relationships.JoinCondition(
+ self.right_w_base_rel,
+ right,
+ self.right_w_base_rel,
+ self.right_w_base_rel,
+ primaryjoin=self.right_w_base_rel.c.base_id==\
+ func.foo(self.base.c.id)
+ )
def _join_fixture_o2o_joined_sub_to_base(self, **kw):
left = self.base.join(self.sub,
**kw
)
+ def _assert_non_simple_warning(self, fn):
+ assert_raises_message(
+ exc.SAWarning,
+ "Non-simple column elements in "
+ "primary join condition for property "
+ r"None - consider using remote\(\) "
+ "annotations to mark the remote side.",
+ fn
+ )
class ColumnCollectionsTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
def test_determine_local_remote_pairs_o2o_joined_sub_to_base(self):
set([self.selfref.c.sid])
)
- def test_determine_remote_columns_o2m_composite_selfref(self):
+ def test_determine_local_remote_pairs_o2m_composite_selfref(self):
joincond = self._join_fixture_o2m_composite_selfref()
eq_(
- joincond.remote_columns,
- set([self.composite_selfref.c.parent_id,
- self.composite_selfref.c.group_id])
+ joincond.local_remote_pairs,
+ [
+ (self.composite_selfref.c.group_id, self.composite_selfref.c.group_id),
+ (self.composite_selfref.c.id, self.composite_selfref.c.parent_id),
+ ]
+ )
+
+ def test_determine_local_remote_pairs_o2m_composite_selfref_func_warning(self):
+ self._assert_non_simple_warning(
+ self._join_fixture_o2m_composite_selfref_func
+ )
+
+ def test_determine_local_remote_pairs_o2m_overlap_func_warning(self):
+ self._assert_non_simple_warning(
+ self._join_fixture_m2o_sub_to_joined_sub_func
+ )
+
+ def test_determine_local_remote_pairs_o2m_composite_selfref_func_annotated(self):
+ joincond = self._join_fixture_o2m_composite_selfref_func_annotated()
+ eq_(
+ joincond.local_remote_pairs,
+ [
+ (self.composite_selfref.c.group_id, self.composite_selfref.c.group_id),
+ (self.composite_selfref.c.id, self.composite_selfref.c.parent_id),
+ ]
)
def test_determine_remote_columns_m2o_composite_selfref(self):
def test_determine_local_remote_pairs_o2m_backref(self):
joincond = self._join_fixture_o2m()
- joincond2 = self._join_fixture_m2m(
+ joincond2 = self._join_fixture_m2o(
primaryjoin=joincond.primaryjoin_reverse_remote,
)
eq_(
"AND composite_selfref.id = composite_selfref.parent_id"
)
-
-
def test_determine_join_m2o(self):
joincond = self._join_fixture_m2o()
self.assert_compile(
pj, "lft.id = pj.lid"
)
+ def test_join_targets_o2m_composite_selfref(self):
+ joincond = self._join_fixture_o2m_composite_selfref()
+ right = select([joincond.child_selectable]).alias('pj')
+ pj, sj, sec, adapter, ds = joincond.join_targets(
+ joincond.parent_selectable,
+ right,
+ True)
+ self.assert_compile(
+ pj,
+ "pj.group_id = composite_selfref.group_id "
+ "AND composite_selfref.id = pj.parent_id"
+ )
+
+ def test_join_targets_m2o_composite_selfref(self):
+ joincond = self._join_fixture_m2o_composite_selfref()
+ right = select([joincond.child_selectable]).alias('pj')
+ pj, sj, sec, adapter, ds = joincond.join_targets(
+ joincond.parent_selectable,
+ right,
+ True)
+ self.assert_compile(
+ pj,
+ "pj.group_id = composite_selfref.group_id "
+ "AND pj.id = composite_selfref.parent_id"
+ )
+
class LazyClauseTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
def _test_lazy_clause_o2m(self):
from sqlalchemy.orm import mapper, relationship, relation, \
backref, create_session, configure_mappers, \
clear_mappers, sessionmaker, attributes,\
- Session, composite, column_property, foreign
+ Session, composite, column_property, foreign,\
+ remote
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
from test.lib.testing import eq_, startswith_, AssertsCompiledSQL, is_
from test.lib import fixtures
employee_t.c.company_id==employee_t.c.company_id
),
remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
- foreign_keys=[employee_t.c.reports_to_id],
- backref=backref('employees', foreign_keys=None))
+ foreign_keys=[employee_t.c.reports_to_id, employee_t.c.company_id],
+ backref=backref('employees',
+ foreign_keys=[employee_t.c.reports_to_id,
+ employee_t.c.company_id]))
})
self._test()
'company':relationship(Company, backref='employees'),
'reports_to':relationship(Employee,
remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
- foreign_keys=[employee_t.c.reports_to_id],
- backref=backref('employees', foreign_keys=None)
+ foreign_keys=[employee_t.c.reports_to_id,
+ employee_t.c.company_id],
+ backref=backref('employees', foreign_keys=
+ [employee_t.c.reports_to_id, employee_t.c.company_id])
)
})
(employee_t.c.reports_to_id, employee_t.c.emp_id),
(employee_t.c.company_id, employee_t.c.company_id)
],
- foreign_keys=[employee_t.c.reports_to_id],
- backref=backref('employees', foreign_keys=None)
+ foreign_keys=[employee_t.c.reports_to_id,
+ employee_t.c.company_id],
+ backref=backref('employees', foreign_keys=
+ [employee_t.c.reports_to_id, employee_t.c.company_id])
+ )
+ })
+
+ self._test()
+
+ def test_annotated(self):
+ Employee, Company, employee_t, company_t = (self.classes.Employee,
+ self.classes.Company,
+ self.tables.employee_t,
+ self.tables.company_t)
+
+ mapper(Company, company_t)
+ mapper(Employee, employee_t, properties= {
+ 'company':relationship(Company, backref='employees'),
+ 'reports_to':relationship(Employee,
+ primaryjoin=sa.and_(
+ remote(employee_t.c.emp_id)==employee_t.c.reports_to_id,
+ remote(employee_t.c.company_id)==employee_t.c.company_id
+ ),
+ backref=backref('employees')
)
})
self._test()
def _test(self):
+ self._test_relationships()
sess = Session()
self._setup_data(sess)
self._test_lazy_relations(sess)
self._test_join_aliasing(sess)
+ def _test_relationships(self):
+ configure_mappers()
+ Employee = self.classes.Employee
+ employee_t = self.tables.employee_t
+ eq_(
+ set(Employee.employees.property.local_remote_pairs),
+ set([
+ (employee_t.c.company_id, employee_t.c.company_id),
+ (employee_t.c.emp_id, employee_t.c.reports_to_id),
+ ])
+ )
+ eq_(
+ Employee.employees.property.remote_side,
+ set([employee_t.c.company_id, employee_t.c.reports_to_id])
+ )
+ eq_(
+ set(Employee.reports_to.property.local_remote_pairs),
+ set([
+ (employee_t.c.company_id, employee_t.c.company_id),
+ (employee_t.c.reports_to_id, employee_t.c.emp_id),
+ ])
+ )
+
def _setup_data(self, sess):
Employee, Company = self.classes.Employee, self.classes.Company
"on relationship",
sa.orm.configure_mappers)
- def test_no_fks_warning_1(self):
- foobars_with_many_columns, bars, Bar, foobars, Foo, foos = (self.tables.foobars_with_many_columns,
- self.tables.bars,
- self.classes.Bar,
- self.tables.foobars,
- self.classes.Foo,
- self.tables.foos)
-
- mapper(Foo, foos, properties={
- 'bars': relationship(Bar, secondary=foobars,
- primaryjoin=foos.c.id==foobars.c.fid,
- secondaryjoin=foobars.c.bid==bars.c.id)})
- mapper(Bar, bars)
-
- assert_raises_message(sa.exc.SAWarning,
- "No ForeignKey objects were present in "
- "secondary table 'foobars'. Assumed "
- "referenced foreign key columns "
- "'foobars.bid', 'foobars.fid' for join "
- "condition 'foos.id = foobars.fid' on "
- "relationship Foo.bars",
- sa.orm.configure_mappers)
-
- sa.orm.clear_mappers()
- mapper(Foo, foos, properties={
- 'bars': relationship(Bar,
- secondary=foobars_with_many_columns,
- primaryjoin=foos.c.id==
- foobars_with_many_columns.c.fid,
- secondaryjoin=foobars_with_many_columns.c.bid==
- bars.c.id)})
- mapper(Bar, bars)
-
- assert_raises_message(sa.exc.SAWarning,
- "No ForeignKey objects were present in "
- "secondary table 'foobars_with_many_colum"
- "ns'. Assumed referenced foreign key "
- "columns 'foobars_with_many_columns.bid',"
- " 'foobars_with_many_columns.bid1', "
- "'foobars_with_many_columns.bid2', "
- "'foobars_with_many_columns.fid', "
- "'foobars_with_many_columns.fid1', "
- "'foobars_with_many_columns.fid2' for "
- "join condition 'foos.id = "
- "foobars_with_many_columns.fid' on "
- "relationship Foo.bars",
- sa.orm.configure_mappers)
-
- @testing.emits_warning(r'No ForeignKey objects.*')
- def test_no_fks_warning_2(self):
+ def test_no_fks(self):
foobars_with_many_columns, bars, Bar, foobars, Foo, foos = (self.tables.foobars_with_many_columns,
self.tables.bars,
self.classes.Bar,
assert b2.left is not bin.left
assert b3.left is not b2.left is not bin.left
assert b4.left is bin.left # since column is immutable
- assert b4.right is bin.right
- assert b2.right is not bin.right
- assert b3.right is b4.right is bin.right
+ # deannotate copies the element
+ assert bin.right is not b2.right is not b3.right is not b4.right
+ def test_deannotate_2(self):
+ table1 = table('table1', column("col1"), column("col2"))
+ j = table1.c.col1._annotate({"remote":True}) == \
+ table1.c.col2._annotate({"local":True})
+ j2 = sql_util._deep_deannotate(j)
+ eq_(
+ j.left._annotations, {"remote":True}
+ )
+ eq_(
+ j2.left._annotations, {}
+ )
+
+ def test_deannotate_3(self):
+ table1 = table('table1', column("col1"), column("col2"),
+ column("col3"), column("col4"))
+ j = and_(
+ table1.c.col1._annotate({"remote":True})==
+ table1.c.col2._annotate({"local":True}),
+ table1.c.col3._annotate({"remote":True})==
+ table1.c.col4._annotate({"local":True})
+ )
+ j2 = sql_util._deep_deannotate(j)
+ eq_(
+ j.clauses[0].left._annotations, {"remote":True}
+ )
+ eq_(
+ j2.clauses[0].left._annotations, {}
+ )
+
+
+
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
