PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-06-02 05:48:52 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
7d96ad4d535dc02a8ab1384df1db94dea2a045b5
sqlalchemy/test/orm/test_relationships.py
T
Mike Bayer 414af7b612 - The system by which a :class:.Column considers itself to be an 
"auto increment" column has been changed, such that autoincrement
is no longer implicitly enabled for a :class:`.Table` that has a
composite primary key.  In order to accommodate being able to enable
autoincrement for a composite PK member column while at the same time
maintaining SQLAlchemy's long standing behavior of enabling
implicit autoincrement for a single integer primary key, a third
state has been added to the :paramref:`.Column.autoincrement` parameter
``"auto"``, which is now the default. fixes #3216
- The MySQL dialect no longer generates an extra "KEY" directive when
generating CREATE TABLE DDL for a table using InnoDB with a
composite primary key with AUTO_INCREMENT on a column that isn't the
first column;  to overcome InnoDB's limitation here, the PRIMARY KEY
constraint is now generated with the AUTO_INCREMENT column placed
first in the list of columns.
2015-10-07 10:02:45 -04:00

3989 lines
128 KiB
Python
Raw Blame History

from sqlalchemy.testing import assert_raises, assert_raises_message
import datetime
import sqlalchemy as sa
from sqlalchemy import testing
from sqlalchemy import Integer, String, ForeignKey, MetaData, and_
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, relation, \
backref, create_session, configure_mappers, \
clear_mappers, sessionmaker, attributes,\
Session, composite, column_property, foreign,\
remote, synonym, joinedload, subqueryload
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE
from sqlalchemy.testing import eq_, startswith_, AssertsCompiledSQL, is_
from sqlalchemy.testing import fixtures
from test.orm import _fixtures
from sqlalchemy import exc
from sqlalchemy import inspect
class _RelationshipErrors(object):
def _assert_raises_no_relevant_fks(self, fn, expr, relname,
primary, *arg, **kw):
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any relevant foreign key columns "
"for %s join condition '%s' on relationship %s. "
"Ensure that referencing columns are associated with "
"a ForeignKey or ForeignKeyConstraint, or are annotated "
r"in the join condition with the foreign\(\) annotation."
% (
primary, expr, relname
),
fn, *arg, **kw
)
def _assert_raises_no_equality(self, fn, expr, relname,
primary, *arg, **kw):
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any simple equality expressions "
"involving locally mapped foreign key columns for %s join "
"condition '%s' on relationship %s. "
"Ensure that referencing columns are associated with a "
"ForeignKey or ForeignKeyConstraint, or are annotated in "
r"the join condition with the foreign\(\) annotation. "
"To allow comparison operators other than '==', "
"the relationship can be marked as viewonly=True." % (
primary, expr, relname
),
fn, *arg, **kw
)
def _assert_raises_ambig_join(self, fn, relname, secondary_arg,
*arg, **kw):
if secondary_arg is not None:
assert_raises_message(
exc.ArgumentError,
"Could not determine join condition between "
"parent/child tables on relationship %s - "
"there are multiple foreign key paths linking the "
"tables via secondary table '%s'. "
"Specify the 'foreign_keys' argument, providing a list "
"of those columns which should be counted as "
"containing a foreign key reference from the "
"secondary table to each of the parent and child tables."
% (relname, secondary_arg),
fn, *arg, **kw)
else:
assert_raises_message(
exc.ArgumentError,
"Could not determine join "
"condition between parent/child tables on "
"relationship %s - there are multiple foreign key "
"paths linking the tables. Specify the "
"'foreign_keys' argument, providing a list of those "
"columns which should be counted as containing a "
"foreign key reference to the parent table."
% (relname,),
fn, *arg, **kw)
def _assert_raises_no_join(self, fn, relname, secondary_arg,
*arg, **kw):
if secondary_arg is not None:
assert_raises_message(
exc.NoForeignKeysError,
"Could not determine join condition between "
"parent/child tables on relationship %s - "
"there are no foreign keys linking these tables "
"via secondary table '%s'. "
"Ensure that referencing columns are associated with a "
"ForeignKey "
"or ForeignKeyConstraint, or specify 'primaryjoin' and "
"'secondaryjoin' expressions"
% (relname, secondary_arg),
fn, *arg, **kw)
else:
assert_raises_message(
exc.NoForeignKeysError,
"Could not determine join condition between "
"parent/child tables on relationship %s - "
"there are no foreign keys linking these tables. "
"Ensure that referencing columns are associated with a "
"ForeignKey "
"or ForeignKeyConstraint, or specify a 'primaryjoin' "
"expression."
% (relname,),
fn, *arg, **kw)
def _assert_raises_ambiguous_direction(self, fn, relname, *arg, **kw):
assert_raises_message(
sa.exc.ArgumentError,
"Can't determine relationship"
" direction for relationship '%s' - foreign "
"key columns within the join condition are present "
"in both the parent and the child's mapped tables. "
"Ensure that only those columns referring to a parent column "
r"are marked as foreign, either via the foreign\(\) annotation or "
"via the foreign_keys argument."
% relname,
fn, *arg, **kw
)
def _assert_raises_no_local_remote(self, fn, relname, *arg, **kw):
assert_raises_message(
sa.exc.ArgumentError,
"Relationship %s could not determine "
"any unambiguous local/remote column "
"pairs based on join condition and remote_side arguments. "
r"Consider using the remote\(\) annotation to "
"accurately mark those elements of the join "
"condition that are on the remote side of the relationship." % (
relname
),
fn, *arg, **kw
)
class DependencyTwoParentTest(fixtures.MappedTest):
"""Test flush() when a mapper is dependent on multiple relationships"""
run_setup_mappers = 'once'
run_inserts = 'once'
run_deletes = None
@classmethod
def define_tables(cls, metadata):
Table("tbl_a", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("name", String(128)))
Table("tbl_b", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("name", String(128)))
Table("tbl_c", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("tbl_a_id", Integer, ForeignKey("tbl_a.id"),
nullable=False),
Column("name", String(128)))
Table("tbl_d", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("tbl_c_id", Integer, ForeignKey("tbl_c.id"),
nullable=False),
Column("tbl_b_id", Integer, ForeignKey("tbl_b.id")),
Column("name", String(128)))
@classmethod
def setup_classes(cls):
class A(cls.Basic):
pass
class B(cls.Basic):
pass
class C(cls.Basic):
pass
class D(cls.Basic):
pass
@classmethod
def setup_mappers(cls):
A, C, B, D, tbl_b, tbl_c, tbl_a, tbl_d = (cls.classes.A,
cls.classes.C,
cls.classes.B,
cls.classes.D,
cls.tables.tbl_b,
cls.tables.tbl_c,
cls.tables.tbl_a,
cls.tables.tbl_d)
mapper(A, tbl_a, properties=dict(
c_rows=relationship(C, cascade="all, delete-orphan",
backref="a_row")))
mapper(B, tbl_b)
mapper(C, tbl_c, properties=dict(
d_rows=relationship(D, cascade="all, delete-orphan",
backref="c_row")))
mapper(D, tbl_d, properties=dict(
b_row=relationship(B)))
@classmethod
def insert_data(cls):
A, C, B, D = (cls.classes.A,
cls.classes.C,
cls.classes.B,
cls.classes.D)
session = create_session()
a = A(name='a1')
b = B(name='b1')
c = C(name='c1', a_row=a)
d1 = D(name='d1', b_row=b, c_row=c) # noqa
d2 = D(name='d2', b_row=b, c_row=c) # noqa
d3 = D(name='d3', b_row=b, c_row=c) # noqa
session.add(a)
session.add(b)
session.flush()
def test_DeleteRootTable(self):
A = self.classes.A
session = create_session()
a = session.query(A).filter_by(name='a1').one()
session.delete(a)
session.flush()
def test_DeleteMiddleTable(self):
C = self.classes.C
session = create_session()
c = session.query(C).filter_by(name='c1').one()
session.delete(c)
session.flush()
class M2ODontOverwriteFKTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table(
'a', metadata,
Column('id', Integer, primary_key=True),
Column('bid', ForeignKey('b.id'))
)
Table(
'b', metadata,
Column('id', Integer, primary_key=True),
)
def _fixture(self, uselist=False):
a, b = self.tables.a, self.tables.b
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
mapper(A, a, properties={
'b': relationship(B, uselist=uselist)
})
mapper(B, b)
return A, B
def test_joinedload_doesnt_produce_bogus_event(self):
A, B = self._fixture()
sess = Session()
b1 = B()
sess.add(b1)
sess.flush()
a1 = A()
sess.add(a1)
sess.commit()
# test that was broken by #3060
a1 = sess.query(A).options(joinedload("b")).first()
a1.bid = b1.id
sess.flush()
eq_(a1.bid, b1.id)
def test_init_doesnt_produce_scalar_event(self):
A, B = self._fixture()
sess = Session()
b1 = B()
sess.add(b1)
sess.flush()
a1 = A()
assert a1.b is None
a1.bid = b1.id
sess.add(a1)
sess.flush()
assert a1.bid is not None
def test_init_doesnt_produce_collection_event(self):
A, B = self._fixture(uselist=True)
sess = Session()
b1 = B()
sess.add(b1)
sess.flush()
a1 = A()
assert a1.b == []
a1.bid = b1.id
sess.add(a1)
sess.flush()
assert a1.bid is not None
def test_scalar_relationship_overrides_fk(self):
A, B = self._fixture()
sess = Session()
b1 = B()
sess.add(b1)
sess.flush()
a1 = A()
a1.bid = b1.id
a1.b = None
sess.add(a1)
sess.flush()
assert a1.bid is None
def test_collection_relationship_overrides_fk(self):
A, B = self._fixture(uselist=True)
sess = Session()
b1 = B()
sess.add(b1)
sess.flush()
a1 = A()
a1.bid = b1.id
a1.b = []
sess.add(a1)
sess.flush()
# this is weird
assert a1.bid is not None
class DirectSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
"""Tests the ultimate join condition, a single column
that points to itself, e.g. within a SQL function or similar.
The test is against a materialized path setup.
this is an **extremely** unusual case::
Entity
------
path -------+
^ |
+---------+
In this case, one-to-many and many-to-one are no longer accurate.
Both relationships return collections. I'm not sure if this is a good
idea.
"""
__dialect__ = 'default'
@classmethod
def define_tables(cls, metadata):
Table('entity', metadata,
Column('path', String(100), primary_key=True)
)
@classmethod
def setup_classes(cls):
class Entity(cls.Basic):
def __init__(self, path):
self.path = path
def _descendants_fixture(self, data=True):
Entity = self.classes.Entity
entity = self.tables.entity
m = mapper(Entity, entity, properties={
"descendants": relationship(
Entity,
primaryjoin=remote(foreign(entity.c.path)).like(
entity.c.path.concat('/%')),
viewonly=True,
order_by=entity.c.path)
})
configure_mappers()
assert m.get_property("descendants").direction is ONETOMANY
if data:
return self._fixture()
def _anscestors_fixture(self, data=True):
Entity = self.classes.Entity
entity = self.tables.entity
m = mapper(Entity, entity, properties={
"anscestors": relationship(
Entity,
primaryjoin=entity.c.path.like(
remote(foreign(entity.c.path)).concat('/%')),
viewonly=True,
order_by=entity.c.path)
})
configure_mappers()
assert m.get_property("anscestors").direction is ONETOMANY
if data:
return self._fixture()
def _fixture(self):
Entity = self.classes.Entity
sess = Session()
sess.add_all([
Entity("/foo"),
Entity("/foo/bar1"),
Entity("/foo/bar2"),
Entity("/foo/bar2/bat1"),
Entity("/foo/bar2/bat2"),
Entity("/foo/bar3"),
Entity("/bar"),
Entity("/bar/bat1")
])
return sess
def test_descendants_lazyload_clause(self):
self._descendants_fixture(data=False)
Entity = self.classes.Entity
self.assert_compile(
Entity.descendants.property.strategy._lazywhere,
"entity.path LIKE (:param_1 || :path_1)"
)
self.assert_compile(
Entity.descendants.property.strategy._rev_lazywhere,
":param_1 LIKE (entity.path || :path_1)"
)
def test_ancestors_lazyload_clause(self):
self._anscestors_fixture(data=False)
Entity = self.classes.Entity
# :param_1 LIKE (:param_1 || :path_1)
self.assert_compile(
Entity.anscestors.property.strategy._lazywhere,
":param_1 LIKE (entity.path || :path_1)"
)
self.assert_compile(
Entity.anscestors.property.strategy._rev_lazywhere,
"entity.path LIKE (:param_1 || :path_1)"
)
def test_descendants_lazyload(self):
sess = self._descendants_fixture()
Entity = self.classes.Entity
e1 = sess.query(Entity).filter_by(path="/foo").first()
eq_(
[e.path for e in e1.descendants],
["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
"/foo/bar2/bat2", "/foo/bar3"]
)
def test_anscestors_lazyload(self):
sess = self._anscestors_fixture()
Entity = self.classes.Entity
e1 = sess.query(Entity).filter_by(path="/foo/bar2/bat1").first()
eq_(
[e.path for e in e1.anscestors],
["/foo", "/foo/bar2"]
)
def test_descendants_joinedload(self):
sess = self._descendants_fixture()
Entity = self.classes.Entity
e1 = sess.query(Entity).filter_by(path="/foo").\
options(joinedload(Entity.descendants)).first()
eq_(
[e.path for e in e1.descendants],
["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
"/foo/bar2/bat2", "/foo/bar3"]
)
def test_descendants_subqueryload(self):
sess = self._descendants_fixture()
Entity = self.classes.Entity
e1 = sess.query(Entity).filter_by(path="/foo").\
options(subqueryload(Entity.descendants)).first()
eq_(
[e.path for e in e1.descendants],
["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
"/foo/bar2/bat2", "/foo/bar3"]
)
def test_anscestors_joinedload(self):
sess = self._anscestors_fixture()
Entity = self.classes.Entity
e1 = sess.query(Entity).filter_by(path="/foo/bar2/bat1").\
options(joinedload(Entity.anscestors)).first()
eq_(
[e.path for e in e1.anscestors],
["/foo", "/foo/bar2"]
)
def test_plain_join_descendants(self):
self._descendants_fixture(data=False)
Entity = self.classes.Entity
sess = Session()
self.assert_compile(
sess.query(Entity).join(Entity.descendants, aliased=True),
"SELECT entity.path AS entity_path FROM entity JOIN entity AS "
"entity_1 ON entity_1.path LIKE (entity.path || :path_1)"
)
class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
"""Tests a composite FK where, in
the relationship(), one col points
to itself in the same table.
this is a very unusual case::
company employee
---------- ----------
company_id <--- company_id ------+
name ^ |
+------------+
emp_id <---------+
name |
reports_to_id ---+
employee joins to its sub-employees
both on reports_to_id, *and on company_id to itself*.
"""
__dialect__ = 'default'
@classmethod
def define_tables(cls, metadata):
Table('company_t', metadata,
Column('company_id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30)))
Table('employee_t', metadata,
Column('company_id', Integer, primary_key=True),
Column('emp_id', Integer, primary_key=True),
Column('name', String(30)),
Column('reports_to_id', Integer),
sa.ForeignKeyConstraint(
['company_id'],
['company_t.company_id']),
sa.ForeignKeyConstraint(
['company_id', 'reports_to_id'],
['employee_t.company_id', 'employee_t.emp_id']))
@classmethod
def setup_classes(cls):
class Company(cls.Basic):
def __init__(self, name):
self.name = name
class Employee(cls.Basic):
def __init__(self, name, company, emp_id, reports_to=None):
self.name = name
self.company = company
self.emp_id = emp_id
self.reports_to = reports_to
def test_explicit(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,
primaryjoin=employee_t.c.company_id ==
company_t.c.company_id,
backref='employees'),
'reports_to': relationship(Employee, primaryjoin=sa.and_(
employee_t.c.emp_id == employee_t.c.reports_to_id,
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, 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_implicit(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,
remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
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_very_implicit(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,
remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
backref='employees'
)
})
self._test()
def test_very_explicit(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,
_local_remote_pairs=[
(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,
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._assert_lazy_clauses()
self._test()
def test_overlapping_warning(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')
)
})
assert_raises_message(
exc.SAWarning,
r"relationship .* will copy column .* to column "
"employee_t.company_id, which conflicts with relationship\(s\)",
configure_mappers
)
def test_annotated_no_overwriting(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) ==
foreign(employee_t.c.reports_to_id),
remote(employee_t.c.company_id) == employee_t.c.company_id
),
backref=backref('employees')
)
})
self._assert_lazy_clauses()
self._test_no_warning()
def _test_no_overwrite(self, sess, expect_failure):
# test [ticket:3230]
Employee, Company = self.classes.Employee, self.classes.Company
c1 = sess.query(Company).filter_by(name='c1').one()
e3 = sess.query(Employee).filter_by(name='emp3').one()
e3.reports_to = None
if expect_failure:
# if foreign() isn't applied specifically to
# employee_t.c.reports_to_id only, then
# employee_t.c.company_id goes foreign as well and then
# this happens
assert_raises_message(
AssertionError,
"Dependency rule tried to blank-out primary key column "
"'employee_t.company_id'",
sess.flush
)
else:
sess.flush()
eq_(e3.company, c1)
@testing.emits_warning("relationship .* will copy column ")
def _test(self):
self._test_no_warning(overwrites=True)
def _test_no_warning(self, overwrites=False):
configure_mappers()
self._test_relationships()
sess = Session()
self._setup_data(sess)
self._test_lazy_relations(sess)
self._test_join_aliasing(sess)
self._test_no_overwrite(sess, expect_failure=overwrites)
@testing.emits_warning("relationship .* will copy column ")
def _assert_lazy_clauses(self):
configure_mappers()
Employee = self.classes.Employee
self.assert_compile(
Employee.employees.property.strategy._lazywhere,
":param_1 = employee_t.reports_to_id AND "
":param_2 = employee_t.company_id"
)
self.assert_compile(
Employee.employees.property.strategy._rev_lazywhere,
"employee_t.emp_id = :param_1 AND "
"employee_t.company_id = :param_2"
)
def _test_relationships(self):
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
c1 = Company('c1')
c2 = Company('c2')
e1 = Employee('emp1', c1, 1)
e2 = Employee('emp2', c1, 2, e1) # noqa
e3 = Employee('emp3', c1, 3, e1)
e4 = Employee('emp4', c1, 4, e3) # noqa
e5 = Employee('emp5', c2, 1)
e6 = Employee('emp6', c2, 2, e5) # noqa
e7 = Employee('emp7', c2, 3, e5) # noqa
sess.add_all((c1, c2))
sess.commit()
sess.close()
def _test_lazy_relations(self, sess):
Employee, Company = self.classes.Employee, self.classes.Company
c1 = sess.query(Company).filter_by(name='c1').one()
c2 = sess.query(Company).filter_by(name='c2').one()
e1 = sess.query(Employee).filter_by(name='emp1').one()
e5 = sess.query(Employee).filter_by(name='emp5').one()
test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
assert test_e1.name == 'emp1', test_e1.name
test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
assert test_e5.name == 'emp5', test_e5.name
assert [x.name for x in test_e1.employees] == ['emp2', 'emp3']
assert sess.query(Employee).\
get([c1.company_id, 3]).reports_to.name == 'emp1'
assert sess.query(Employee).\
get([c2.company_id, 3]).reports_to.name == 'emp5'
def _test_join_aliasing(self, sess):
Employee, Company = self.classes.Employee, self.classes.Company
eq_(
[n for n, in sess.query(Employee.name).
join(Employee.reports_to, aliased=True).
filter_by(name='emp5').
reset_joinpoint().
order_by(Employee.name)],
['emp6', 'emp7']
)
class CompositeJoinPartialFK(fixtures.MappedTest, AssertsCompiledSQL):
__dialect__ = 'default'
@classmethod
def define_tables(cls, metadata):
Table("parent", metadata,
Column('x', Integer, primary_key=True),
Column('y', Integer, primary_key=True),
Column('z', Integer),
)
Table("child", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('x', Integer),
Column('y', Integer),
Column('z', Integer),
# note 'z' is not here
sa.ForeignKeyConstraint(
["x", "y"],
["parent.x", "parent.y"]
)
)
@classmethod
def setup_mappers(cls):
parent, child = cls.tables.parent, cls.tables.child
class Parent(cls.Comparable):
pass
class Child(cls.Comparable):
pass
mapper(Parent, parent, properties={
'children': relationship(Child, primaryjoin=and_(
parent.c.x == child.c.x,
parent.c.y == child.c.y,
parent.c.z == child.c.z,
))
})
mapper(Child, child)
def test_joins_fully(self):
Parent, Child = self.classes.Parent, self.classes.Child
self.assert_compile(
Parent.children.property.strategy._lazywhere,
":param_1 = child.x AND :param_2 = child.y AND :param_3 = child.z"
)
class SynonymsAsFKsTest(fixtures.MappedTest):
"""Syncrules on foreign keys that are also primary"""
@classmethod
def define_tables(cls, metadata):
Table("tableA", metadata,
Column("id", Integer, primary_key=True),
Column("foo", Integer,),
test_needs_fk=True)
Table("tableB", metadata,
Column("id", Integer, primary_key=True),
Column("_a_id", Integer, key='a_id', primary_key=True),
test_needs_fk=True)
@classmethod
def setup_classes(cls):
class A(cls.Basic):
pass
class B(cls.Basic):
@property
def a_id(self):
return self._a_id
def test_synonym_fk(self):
"""test that active history is enabled on a
one-to-many/one that has use_get==True"""
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(B, tableB, properties={
'a_id': synonym('_a_id', map_column=True)})
mapper(A, tableA, properties={
'b': relationship(B, primaryjoin=(tableA.c.id == foreign(B.a_id)),
uselist=False)})
sess = create_session()
b = B(id=0)
a = A(id=0, b=b)
sess.add(a)
sess.add(b)
sess.flush()
sess.expunge_all()
assert a.b == b
assert a.id == b.a_id
assert a.id == b._a_id
class FKsAsPksTest(fixtures.MappedTest):
"""Syncrules on foreign keys that are also primary"""
@classmethod
def define_tables(cls, metadata):
Table("tableA", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("foo", Integer,),
test_needs_fk=True)
Table("tableB", metadata,
Column("id", Integer, ForeignKey("tableA.id"), primary_key=True),
test_needs_fk=True)
@classmethod
def setup_classes(cls):
class A(cls.Basic):
pass
class B(cls.Basic):
pass
def test_onetoone_switch(self):
"""test that active history is enabled on a
one-to-many/one that has use_get==True"""
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(A, tableA, properties={
'b': relationship(B, cascade="all,delete-orphan", uselist=False)})
mapper(B, tableB)
configure_mappers()
assert A.b.property.strategy.use_get
sess = create_session()
a1 = A()
sess.add(a1)
sess.flush()
sess.close()
a1 = sess.query(A).first()
a1.b = B()
sess.flush()
def test_no_delete_PK_AtoB(self):
"""A cant be deleted without B because B would have no PK value."""
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(A, tableA, properties={
'bs': relationship(B, cascade="save-update")})
mapper(B, tableB)
a1 = A()
a1.bs.append(B())
sess = create_session()
sess.add(a1)
sess.flush()
sess.delete(a1)
try:
sess.flush()
assert False
except AssertionError as e:
startswith_(str(e),
"Dependency rule tried to blank-out "
"primary key column 'tableB.id' on instance ")
def test_no_delete_PK_BtoA(self):
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(B, tableB, properties={
'a': relationship(A, cascade="save-update")})
mapper(A, tableA)
b1 = B()
a1 = A()
b1.a = a1
sess = create_session()
sess.add(b1)
sess.flush()
b1.a = None
try:
sess.flush()
assert False
except AssertionError as e:
startswith_(str(e),
"Dependency rule tried to blank-out "
"primary key column 'tableB.id' on instance ")
@testing.fails_on_everything_except('sqlite', 'mysql')
def test_nullPKsOK_BtoA(self):
A, tableA = self.classes.A, self.tables.tableA
# postgresql cant handle a nullable PK column...?
tableC = Table(
'tablec', tableA.metadata,
Column('id', Integer, primary_key=True),
Column('a_id', Integer, ForeignKey('tableA.id'),
primary_key=True, nullable=True))
tableC.create()
class C(fixtures.BasicEntity):
pass
mapper(C, tableC, properties={
'a': relationship(A, cascade="save-update")
})
mapper(A, tableA)
c1 = C()
c1.id = 5
c1.a = None
sess = create_session()
sess.add(c1)
# test that no error is raised.
sess.flush()
def test_delete_cascade_BtoA(self):
"""No 'blank the PK' error when the child is to
be deleted as part of a cascade"""
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
for cascade in ("save-update, delete",
#"save-update, delete-orphan",
"save-update, delete, delete-orphan"):
mapper(B, tableB, properties={
'a': relationship(A, cascade=cascade, single_parent=True)
})
mapper(A, tableA)
b1 = B()
a1 = A()
b1.a = a1
sess = create_session()
sess.add(b1)
sess.flush()
sess.delete(b1)
sess.flush()
assert a1 not in sess
assert b1 not in sess
sess.expunge_all()
sa.orm.clear_mappers()
def test_delete_cascade_AtoB(self):
"""No 'blank the PK' error when the child is to
be deleted as part of a cascade"""
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
for cascade in ("save-update, delete",
#"save-update, delete-orphan",
"save-update, delete, delete-orphan"):
mapper(A, tableA, properties={
'bs': relationship(B, cascade=cascade)
})
mapper(B, tableB)
a1 = A()
b1 = B()
a1.bs.append(b1)
sess = create_session()
sess.add(a1)
sess.flush()
sess.delete(a1)
sess.flush()
assert a1 not in sess
assert b1 not in sess
sess.expunge_all()
sa.orm.clear_mappers()
def test_delete_manual_AtoB(self):
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(A, tableA, properties={
'bs': relationship(B, cascade="none")})
mapper(B, tableB)
a1 = A()
b1 = B()
a1.bs.append(b1)
sess = create_session()
sess.add(a1)
sess.add(b1)
sess.flush()
sess.delete(a1)
sess.delete(b1)
sess.flush()
assert a1 not in sess
assert b1 not in sess
sess.expunge_all()
def test_delete_manual_BtoA(self):
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
mapper(B, tableB, properties={
'a': relationship(A, cascade="none")})
mapper(A, tableA)
b1 = B()
a1 = A()
b1.a = a1
sess = create_session()
sess.add(b1)
sess.add(a1)
sess.flush()
sess.delete(b1)
sess.delete(a1)
sess.flush()
assert a1 not in sess
assert b1 not in sess
class UniqueColReferenceSwitchTest(fixtures.MappedTest):
"""test a relationship based on a primary
join against a unique non-pk column"""
@classmethod
def define_tables(cls, metadata):
Table("table_a", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("ident", String(10), nullable=False,
unique=True),
)
Table("table_b", metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("a_ident", String(10),
ForeignKey('table_a.ident'),
nullable=False),
)
@classmethod
def setup_classes(cls):
class A(cls.Comparable):
pass
class B(cls.Comparable):
pass
def test_switch_parent(self):
A, B, table_b, table_a = (self.classes.A,
self.classes.B,
self.tables.table_b,
self.tables.table_a)
mapper(A, table_a)
mapper(B, table_b, properties={"a": relationship(A, backref="bs")})
session = create_session()
a1, a2 = A(ident="uuid1"), A(ident="uuid2")
session.add_all([a1, a2])
a1.bs = [
B(), B()
]
session.flush()
session.expire_all()
a1, a2 = session.query(A).all()
for b in list(a1.bs):
b.a = a2
session.delete(a1)
session.flush()
class RelationshipToSelectableTest(fixtures.MappedTest):
"""Test a map to a select that relates to a map to the table."""
@classmethod
def define_tables(cls, metadata):
Table('items', metadata,
Column('item_policy_num', String(10), primary_key=True,
key='policyNum'),
Column('item_policy_eff_date', sa.Date, primary_key=True,
key='policyEffDate'),
Column('item_type', String(20), primary_key=True,
key='type'),
Column('item_id', Integer, primary_key=True,
key='id', autoincrement=False))
def test_basic(self):
items = self.tables.items
class Container(fixtures.BasicEntity):
pass
class LineItem(fixtures.BasicEntity):
pass
container_select = sa.select(
[items.c.policyNum, items.c.policyEffDate, items.c.type],
distinct=True,
).alias('container_select')
mapper(LineItem, items)
mapper(
Container,
container_select,
order_by=sa.asc(container_select.c.type),
properties=dict(
lineItems=relationship(
LineItem,
lazy='select',
cascade='all, delete-orphan',
order_by=sa.asc(items.c.id),
primaryjoin=sa.and_(
container_select.c.policyNum == items.c.policyNum,
container_select.c.policyEffDate ==
items.c.policyEffDate,
container_select.c.type == items.c.type),
foreign_keys=[
items.c.policyNum,
items.c.policyEffDate,
items.c.type
]
)
)
)
session = create_session()
con = Container()
con.policyNum = "99"
con.policyEffDate = datetime.date.today()
con.type = "TESTER"
session.add(con)
for i in range(0, 10):
li = LineItem()
li.id = i
con.lineItems.append(li)
session.add(li)
session.flush()
session.expunge_all()
newcon = session.query(Container).first()
assert con.policyNum == newcon.policyNum
assert len(newcon.lineItems) == 10
for old, new in zip(con.lineItems, newcon.lineItems):
eq_(old.id, new.id)
class FKEquatedToConstantTest(fixtures.MappedTest):
"""test a relationship with a non-column entity in the primary join,
is not viewonly, and also has the non-column's clause mentioned in the
foreign keys list.
"""
@classmethod
def define_tables(cls, metadata):
Table('tags', metadata, Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column("data", String(50)),
)
Table('tag_foo', metadata,
Column("id", Integer, primary_key=True,
test_needs_autoincrement=True),
Column('tagid', Integer),
Column("data", String(50)),
)
def test_basic(self):
tag_foo, tags = self.tables.tag_foo, self.tables.tags
class Tag(fixtures.ComparableEntity):
pass
class TagInstance(fixtures.ComparableEntity):
pass
mapper(Tag, tags, properties={
'foo': relationship(
TagInstance,
primaryjoin=sa.and_(tag_foo.c.data == 'iplc_case',
tag_foo.c.tagid == tags.c.id),
foreign_keys=[tag_foo.c.tagid, tag_foo.c.data]),
})
mapper(TagInstance, tag_foo)
sess = create_session()
t1 = Tag(data='some tag')
t1.foo.append(TagInstance(data='iplc_case'))
t1.foo.append(TagInstance(data='not_iplc_case'))
sess.add(t1)
sess.flush()
sess.expunge_all()
# relationship works
eq_(
sess.query(Tag).all(),
[Tag(data='some tag', foo=[TagInstance(data='iplc_case')])]
)
# both TagInstances were persisted
eq_(
sess.query(TagInstance).order_by(TagInstance.data).all(),
[TagInstance(data='iplc_case'), TagInstance(data='not_iplc_case')]
)
class BackrefPropagatesForwardsArgs(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('users', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(50))
)
Table('addresses', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('user_id', Integer),
Column('email', String(50))
)
@classmethod
def setup_classes(cls):
class User(cls.Comparable):
pass
class Address(cls.Comparable):
pass
def test_backref(self):
User, Address, users, addresses = (self.classes.User,
self.classes.Address,
self.tables.users,
self.tables.addresses)
mapper(User, users, properties={
'addresses': relationship(
Address,
primaryjoin=addresses.c.user_id == users.c.id,
foreign_keys=addresses.c.user_id,
backref='user')
})
mapper(Address, addresses)
sess = sessionmaker()()
u1 = User(name='u1', addresses=[Address(email='a1')])
sess.add(u1)
sess.commit()
eq_(sess.query(Address).all(), [
Address(email='a1', user=User(name='u1'))
])
class AmbiguousJoinInterpretedAsSelfRef(fixtures.MappedTest):
"""test ambiguous joins due to FKs on both sides treated as
self-referential.
this mapping is very similar to that of
test/orm/inheritance/query.py
SelfReferentialTestJoinedToBase , except that inheritance is
not used here.
"""
@classmethod
def define_tables(cls, metadata):
Table(
'subscriber', metadata,
Column(
'id', Integer, primary_key=True,
test_needs_autoincrement=True))
Table(
'address', metadata,
Column(
'subscriber_id', Integer,
ForeignKey('subscriber.id'), primary_key=True),
Column('type', String(1), primary_key=True),
)
@classmethod
def setup_mappers(cls):
subscriber, address = cls.tables.subscriber, cls.tables.address
subscriber_and_address = subscriber.join(
address,
and_(address.c.subscriber_id == subscriber.c.id,
address.c.type.in_(['A', 'B', 'C'])))
class Address(cls.Comparable):
pass
class Subscriber(cls.Comparable):
pass
mapper(Address, address)
mapper(Subscriber, subscriber_and_address, properties={
'id': [subscriber.c.id, address.c.subscriber_id],
'addresses': relationship(Address,
backref=backref("customer"))
})
def test_mapping(self):
Subscriber, Address = self.classes.Subscriber, self.classes.Address
sess = create_session()
assert Subscriber.addresses.property.direction is ONETOMANY
assert Address.customer.property.direction is MANYTOONE
s1 = Subscriber(type='A',
addresses=[
Address(type='D'),
Address(type='E'),
]
)
a1 = Address(type='B', customer=Subscriber(type='C'))
assert s1.addresses[0].customer is s1
assert a1.customer.addresses[0] is a1
sess.add_all([s1, a1])
sess.flush()
sess.expunge_all()
eq_(
sess.query(Subscriber).order_by(Subscriber.type).all(),
[
Subscriber(id=1, type='A'),
Subscriber(id=2, type='B'),
Subscriber(id=2, type='C')
]
)
class ManualBackrefTest(_fixtures.FixtureTest):
"""Test explicit relationships that are backrefs to each other."""
run_inserts = None
def test_o2m(self):
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
mapper(User, users, properties={
'addresses': relationship(Address, back_populates='user')
})
mapper(Address, addresses, properties={
'user': relationship(User, back_populates='addresses')
})
sess = create_session()
u1 = User(name='u1')
a1 = Address(email_address='foo')
u1.addresses.append(a1)
assert a1.user is u1
sess.add(u1)
sess.flush()
sess.expire_all()
assert sess.query(Address).one() is a1
assert a1.user is u1
assert a1 in u1.addresses
def test_invalid_key(self):
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
mapper(User, users, properties={
'addresses': relationship(Address, back_populates='userr')
})
mapper(Address, addresses, properties={
'user': relationship(User, back_populates='addresses')
})
assert_raises(sa.exc.InvalidRequestError, configure_mappers)
def test_invalid_target(self):
addresses, Dingaling, User, dingalings, Address, users = (
self.tables.addresses,
self.classes.Dingaling,
self.classes.User,
self.tables.dingalings,
self.classes.Address,
self.tables.users)
mapper(User, users, properties={
'addresses': relationship(Address, back_populates='dingaling'),
})
mapper(Dingaling, dingalings)
mapper(Address, addresses, properties={
'dingaling': relationship(Dingaling)
})
assert_raises_message(sa.exc.ArgumentError,
r"reverse_property 'dingaling' on relationship "
"User.addresses references "
"relationship Address.dingaling, which does not "
"reference mapper Mapper\|User\|users",
configure_mappers)
class JoinConditionErrorTest(fixtures.TestBase):
def test_clauseelement_pj(self):
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class C1(Base):
__tablename__ = 'c1'
id = Column('id', Integer, primary_key=True)
class C2(Base):
__tablename__ = 'c2'
id = Column('id', Integer, primary_key=True)
c1id = Column('c1id', Integer, ForeignKey('c1.id'))
c2 = relationship(C1, primaryjoin=C1.id)
assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_clauseelement_pj_false(self):
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class C1(Base):
__tablename__ = 'c1'
id = Column('id', Integer, primary_key=True)
class C2(Base):
__tablename__ = 'c2'
id = Column('id', Integer, primary_key=True)
c1id = Column('c1id', Integer, ForeignKey('c1.id'))
c2 = relationship(C1, primaryjoin="x" == "y")
assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_only_column_elements(self):
m = MetaData()
t1 = Table('t1', m,
Column('id', Integer, primary_key=True),
Column('foo_id', Integer, ForeignKey('t2.id')),
)
t2 = Table('t2', m,
Column('id', Integer, primary_key=True),
)
class C1(object):
pass
class C2(object):
pass
mapper(C1, t1, properties={
'c2': relationship(C2, primaryjoin=t1.join(t2))})
mapper(C2, t2)
assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_invalid_string_args(self):
from sqlalchemy.ext.declarative import declarative_base
for argname, arg in [
('remote_side', ['c1.id']),
('remote_side', ['id']),
('foreign_keys', ['c1id']),
('foreign_keys', ['C2.c1id']),
('order_by', ['id']),
]:
clear_mappers()
kw = {argname: arg}
Base = declarative_base()
class C1(Base):
__tablename__ = 'c1'
id = Column('id', Integer, primary_key=True)
class C2(Base):
__tablename__ = 'c2'
id_ = Column('id', Integer, primary_key=True)
c1id = Column('c1id', Integer, ForeignKey('c1.id'))
c2 = relationship(C1, **kw)
assert_raises_message(
sa.exc.ArgumentError,
"Column-based expression object expected "
"for argument '%s'; got: '%s', type %r" %
(argname, arg[0], type(arg[0])),
configure_mappers)
def test_fk_error_not_raised_unrelated(self):
m = MetaData()
t1 = Table('t1', m,
Column('id', Integer, primary_key=True),
Column('foo_id', Integer, ForeignKey('t2.nonexistent_id')),
)
t2 = Table('t2', m, # noqa
Column('id', Integer, primary_key=True),
)
t3 = Table('t3', m,
Column('id', Integer, primary_key=True),
Column('t1id', Integer, ForeignKey('t1.id'))
)
class C1(object):
pass
class C2(object):
pass
mapper(C1, t1, properties={'c2': relationship(C2)})
mapper(C2, t3)
assert C1.c2.property.primaryjoin.compare(t1.c.id == t3.c.t1id)
def test_join_error_raised(self):
m = MetaData()
t1 = Table('t1', m,
Column('id', Integer, primary_key=True),
)
t2 = Table('t2', m, # noqa
Column('id', Integer, primary_key=True),
)
t3 = Table('t3', m,
Column('id', Integer, primary_key=True),
Column('t1id', Integer)
)
class C1(object):
pass
class C2(object):
pass
mapper(C1, t1, properties={'c2': relationship(C2)})
mapper(C2, t3)
assert_raises(sa.exc.ArgumentError, configure_mappers)
def teardown(self):
clear_mappers()
class TypeMatchTest(fixtures.MappedTest):
"""test errors raised when trying to add items
whose type is not handled by a relationship"""
@classmethod
def define_tables(cls, metadata):
Table("a", metadata,
Column('aid', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('adata', String(30)))
Table("b", metadata,
Column('bid', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("a_id", Integer, ForeignKey("a.aid")),
Column('bdata', String(30)))
Table("c", metadata,
Column('cid', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("b_id", Integer, ForeignKey("b.bid")),
Column('cdata', String(30)))
Table("d", metadata,
Column('did', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("a_id", Integer, ForeignKey("a.aid")),
Column('ddata', String(30)))
def test_o2m_oncascade(self):
a, c, b = (self.tables.a,
self.tables.c,
self.tables.b)
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
class C(fixtures.BasicEntity):
pass
mapper(A, a, properties={'bs': relationship(B)})
mapper(B, b)
mapper(C, c)
a1 = A()
b1 = B()
c1 = C()
a1.bs.append(b1)
a1.bs.append(c1)
sess = create_session()
try:
sess.add(a1)
assert False
except AssertionError as err:
eq_(str(err),
"Attribute 'bs' on class '%s' doesn't handle "
"objects of type '%s'" % (A, C))
def test_o2m_onflush(self):
a, c, b = (self.tables.a,
self.tables.c,
self.tables.b)
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
class C(fixtures.BasicEntity):
pass
mapper(A, a, properties={'bs': relationship(B, cascade="none")})
mapper(B, b)
mapper(C, c)
a1 = A()
b1 = B()
c1 = C()
a1.bs.append(b1)
a1.bs.append(c1)
sess = create_session()
sess.add(a1)
sess.add(b1)
sess.add(c1)
assert_raises_message(sa.orm.exc.FlushError,
"Attempting to flush an item",
sess.flush)
def test_o2m_nopoly_onflush(self):
a, c, b = (self.tables.a,
self.tables.c,
self.tables.b)
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
class C(B):
pass
mapper(A, a, properties={'bs': relationship(B, cascade="none")})
mapper(B, b)
mapper(C, c, inherits=B)
a1 = A()
b1 = B()
c1 = C()
a1.bs.append(b1)
a1.bs.append(c1)
sess = create_session()
sess.add(a1)
sess.add(b1)
sess.add(c1)
assert_raises_message(sa.orm.exc.FlushError,
"Attempting to flush an item",
sess.flush)
def test_m2o_nopoly_onflush(self):
a, b, d = (self.tables.a,
self.tables.b,
self.tables.d)
class A(fixtures.BasicEntity):
pass
class B(A):
pass
class D(fixtures.BasicEntity):
pass
mapper(A, a)
mapper(B, b, inherits=A)
mapper(D, d, properties={"a": relationship(A, cascade="none")})
b1 = B()
d1 = D()
d1.a = b1
sess = create_session()
sess.add(b1)
sess.add(d1)
assert_raises_message(sa.orm.exc.FlushError,
"Attempting to flush an item",
sess.flush)
def test_m2o_oncascade(self):
a, b, d = (self.tables.a,
self.tables.b,
self.tables.d)
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
class D(fixtures.BasicEntity):
pass
mapper(A, a)
mapper(B, b)
mapper(D, d, properties={"a": relationship(A)})
b1 = B()
d1 = D()
d1.a = b1
sess = create_session()
assert_raises_message(AssertionError,
"doesn't handle objects of type",
sess.add, d1)
class TypedAssociationTable(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
class MySpecialType(sa.types.TypeDecorator):
impl = String
def process_bind_param(self, value, dialect):
return "lala" + value
def process_result_value(self, value, dialect):
return value[4:]
Table('t1', metadata,
Column('col1', MySpecialType(30), primary_key=True),
Column('col2', String(30)))
Table('t2', metadata,
Column('col1', MySpecialType(30), primary_key=True),
Column('col2', String(30)))
Table('t3', metadata,
Column('t1c1', MySpecialType(30), ForeignKey('t1.col1')),
Column('t2c1', MySpecialType(30), ForeignKey('t2.col1')))
def test_m2m(self):
"""Many-to-many tables with special types for candidate keys."""
t2, t3, t1 = (self.tables.t2,
self.tables.t3,
self.tables.t1)
class T1(fixtures.BasicEntity):
pass
class T2(fixtures.BasicEntity):
pass
mapper(T2, t2)
mapper(T1, t1, properties={
't2s': relationship(T2, secondary=t3, backref='t1s')})
a = T1()
a.col1 = "aid"
b = T2()
b.col1 = "bid"
c = T2()
c.col1 = "cid"
a.t2s.append(b)
a.t2s.append(c)
sess = create_session()
sess.add(a)
sess.flush()
assert t3.count().scalar() == 2
a.t2s.remove(c)
sess.flush()
assert t3.count().scalar() == 1
class CustomOperatorTest(fixtures.MappedTest, AssertsCompiledSQL):
"""test op() in conjunction with join conditions"""
run_create_tables = run_deletes = None
__dialect__ = 'default'
@classmethod
def define_tables(cls, metadata):
Table('a', metadata,
Column('id', Integer, primary_key=True),
Column('foo', String(50))
)
Table('b', metadata,
Column('id', Integer, primary_key=True),
Column('foo', String(50))
)
def test_join_on_custom_op(self):
class A(fixtures.BasicEntity):
pass
class B(fixtures.BasicEntity):
pass
mapper(A, self.tables.a, properties={
'bs': relationship(B,
primaryjoin=self.tables.a.c.foo.op(
'&*', is_comparison=True
)(foreign(self.tables.b.c.foo)),
viewonly=True
)
})
mapper(B, self.tables.b)
self.assert_compile(
Session().query(A).join(A.bs),
"SELECT a.id AS a_id, a.foo AS a_foo "
"FROM a JOIN b ON a.foo &* b.foo"
)
class ViewOnlyHistoryTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table("t1", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)))
Table("t2", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
Column('t1id', Integer, ForeignKey('t1.id')))
def _assert_fk(self, a1, b1, is_set):
s = Session(testing.db)
s.add_all([a1, b1])
s.flush()
if is_set:
eq_(b1.t1id, a1.id)
else:
eq_(b1.t1id, None)
return s
def test_o2m_viewonly_oneside(self):
class A(fixtures.ComparableEntity):
pass
class B(fixtures.ComparableEntity):
pass
mapper(A, self.tables.t1, properties={
"bs": relationship(B, viewonly=True,
backref=backref("a", viewonly=False))
})
mapper(B, self.tables.t2)
a1 = A()
b1 = B()
a1.bs.append(b1)
assert b1.a is a1
assert not inspect(a1).attrs.bs.history.has_changes()
assert inspect(b1).attrs.a.history.has_changes()
sess = self._assert_fk(a1, b1, True)
a1.bs.remove(b1)
assert a1 not in sess.dirty
assert b1 in sess.dirty
def test_m2o_viewonly_oneside(self):
class A(fixtures.ComparableEntity):
pass
class B(fixtures.ComparableEntity):
pass
mapper(A, self.tables.t1, properties={
"bs": relationship(B, viewonly=False,
backref=backref("a", viewonly=True))
})
mapper(B, self.tables.t2)
a1 = A()
b1 = B()
b1.a = a1
assert b1 in a1.bs
assert inspect(a1).attrs.bs.history.has_changes()
assert not inspect(b1).attrs.a.history.has_changes()
sess = self._assert_fk(a1, b1, True)
a1.bs.remove(b1)
assert a1 in sess.dirty
assert b1 not in sess.dirty
def test_o2m_viewonly_only(self):
class A(fixtures.ComparableEntity):
pass
class B(fixtures.ComparableEntity):
pass
mapper(A, self.tables.t1, properties={
"bs": relationship(B, viewonly=True)
})
mapper(B, self.tables.t2)
a1 = A()
b1 = B()
a1.bs.append(b1)
assert not inspect(a1).attrs.bs.history.has_changes()
self._assert_fk(a1, b1, False)
def test_m2o_viewonly_only(self):
class A(fixtures.ComparableEntity):
pass
class B(fixtures.ComparableEntity):
pass
mapper(A, self.tables.t1)
mapper(B, self.tables.t2, properties={
'a': relationship(A, viewonly=True)
})
a1 = A()
b1 = B()
b1.a = a1
assert not inspect(b1).attrs.a.history.has_changes()
self._assert_fk(a1, b1, False)
class ViewOnlyM2MBackrefTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table("t1", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)))
Table("t2", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
)
Table("t1t2", metadata,
Column('t1id', Integer, ForeignKey('t1.id'), primary_key=True),
Column('t2id', Integer, ForeignKey('t2.id'), primary_key=True),
)
def test_viewonly(self):
t1t2, t2, t1 = (self.tables.t1t2,
self.tables.t2,
self.tables.t1)
class A(fixtures.ComparableEntity):
pass
class B(fixtures.ComparableEntity):
pass
mapper(A, t1, properties={
'bs': relationship(B, secondary=t1t2,
backref=backref('as_', viewonly=True))
})
mapper(B, t2)
sess = create_session()
a1 = A()
b1 = B(as_=[a1])
assert not inspect(b1).attrs.as_.history.has_changes()
sess.add(a1)
sess.flush()
eq_(
sess.query(A).first(), A(bs=[B(id=b1.id)])
)
eq_(
sess.query(B).first(), B(as_=[A(id=a1.id)])
)
class ViewOnlyOverlappingNames(fixtures.MappedTest):
"""'viewonly' mappings with overlapping PK column names."""
@classmethod
def define_tables(cls, metadata):
Table("t1", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)))
Table("t2", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
Column('t1id', Integer, ForeignKey('t1.id')))
Table("t3", metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
Column('t2id', Integer, ForeignKey('t2.id')))
def test_three_table_view(self):
"""A three table join with overlapping PK names.
A third table is pulled into the primary join condition using
overlapping PK column names and should not produce 'conflicting column'
error.
"""
t2, t3, t1 = (self.tables.t2,
self.tables.t3,
self.tables.t1)
class C1(fixtures.BasicEntity):
pass
class C2(fixtures.BasicEntity):
pass
class C3(fixtures.BasicEntity):
pass
mapper(C1, t1, properties={
't2s': relationship(C2),
't2_view': relationship(
C2,
viewonly=True,
primaryjoin=sa.and_(t1.c.id == t2.c.t1id,
t3.c.t2id == t2.c.id,
t3.c.data == t1.c.data))})
mapper(C2, t2)
mapper(C3, t3, properties={
't2': relationship(C2)})
c1 = C1()
c1.data = 'c1data'
c2a = C2()
c1.t2s.append(c2a)
c2b = C2()
c1.t2s.append(c2b)
c3 = C3()
c3.data = 'c1data'
c3.t2 = c2b
sess = create_session()
sess.add(c1)
sess.add(c3)
sess.flush()
sess.expunge_all()
c1 = sess.query(C1).get(c1.id)
assert set([x.id for x in c1.t2s]) == set([c2a.id, c2b.id])
assert set([x.id for x in c1.t2_view]) == set([c2b.id])
class ViewOnlyUniqueNames(fixtures.MappedTest):
"""'viewonly' mappings with unique PK column names."""
@classmethod
def define_tables(cls, metadata):
Table("t1", metadata,
Column('t1id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)))
Table("t2", metadata,
Column('t2id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
Column('t1id_ref', Integer, ForeignKey('t1.t1id')))
Table("t3", metadata,
Column('t3id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(40)),
Column('t2id_ref', Integer, ForeignKey('t2.t2id')))
def test_three_table_view(self):
"""A three table join with overlapping PK names.
A third table is pulled into the primary join condition using unique
PK column names and should not produce 'mapper has no columnX' error.
"""
t2, t3, t1 = (self.tables.t2,
self.tables.t3,
self.tables.t1)
class C1(fixtures.BasicEntity):
pass
class C2(fixtures.BasicEntity):
pass
class C3(fixtures.BasicEntity):
pass
mapper(C1, t1, properties={
't2s': relationship(C2),
't2_view': relationship(
C2,
viewonly=True,
primaryjoin=sa.and_(t1.c.t1id == t2.c.t1id_ref,
t3.c.t2id_ref == t2.c.t2id,
t3.c.data == t1.c.data))})
mapper(C2, t2)
mapper(C3, t3, properties={
't2': relationship(C2)})
c1 = C1()
c1.data = 'c1data'
c2a = C2()
c1.t2s.append(c2a)
c2b = C2()
c1.t2s.append(c2b)
c3 = C3()
c3.data = 'c1data'
c3.t2 = c2b
sess = create_session()
sess.add_all((c1, c3))
sess.flush()
sess.expunge_all()
c1 = sess.query(C1).get(c1.t1id)
assert set([x.t2id for x in c1.t2s]) == set([c2a.t2id, c2b.t2id])
assert set([x.t2id for x in c1.t2_view]) == set([c2b.t2id])
class ViewOnlyLocalRemoteM2M(fixtures.TestBase):
"""test that local-remote is correctly determined for m2m"""
def test_local_remote(self):
meta = MetaData()
t1 = Table('t1', meta,
Column('id', Integer, primary_key=True),
)
t2 = Table('t2', meta,
Column('id', Integer, primary_key=True),
)
t12 = Table('tab', meta,
Column('t1_id', Integer, ForeignKey('t1.id',)),
Column('t2_id', Integer, ForeignKey('t2.id',)),
)
class A(object):
pass
class B(object):
pass
mapper(B, t2, )
m = mapper(A, t1, properties=dict(
b_view=relationship(B, secondary=t12, viewonly=True),
b_plain=relationship(B, secondary=t12),
)
)
configure_mappers()
assert m.get_property('b_view').local_remote_pairs == \
m.get_property('b_plain').local_remote_pairs == \
[(t1.c.id, t12.c.t1_id), (t2.c.id, t12.c.t2_id)]
class ViewOnlyNonEquijoin(fixtures.MappedTest):
"""'viewonly' mappings based on non-equijoins."""
@classmethod
def define_tables(cls, metadata):
Table('foos', metadata,
Column('id', Integer, primary_key=True))
Table('bars', metadata,
Column('id', Integer, primary_key=True),
Column('fid', Integer))
def test_viewonly_join(self):
bars, foos = self.tables.bars, self.tables.foos
class Foo(fixtures.ComparableEntity):
pass
class Bar(fixtures.ComparableEntity):
pass
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id > bars.c.fid,
foreign_keys=[bars.c.fid],
viewonly=True)})
mapper(Bar, bars)
sess = create_session()
sess.add_all((Foo(id=4),
Foo(id=9),
Bar(id=1, fid=2),
Bar(id=2, fid=3),
Bar(id=3, fid=6),
Bar(id=4, fid=7)))
sess.flush()
sess = create_session()
eq_(sess.query(Foo).filter_by(id=4).one(),
Foo(id=4, bars=[Bar(fid=2), Bar(fid=3)]))
eq_(sess.query(Foo).filter_by(id=9).one(),
Foo(id=9, bars=[Bar(fid=2), Bar(fid=3), Bar(fid=6), Bar(fid=7)]))
class ViewOnlyRepeatedRemoteColumn(fixtures.MappedTest):
"""'viewonly' mappings that contain the same 'remote' column twice"""
@classmethod
def define_tables(cls, metadata):
Table('foos', metadata,
Column(
'id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('bid1', Integer, ForeignKey('bars.id')),
Column('bid2', Integer, ForeignKey('bars.id')))
Table('bars', metadata,
Column(
'id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)))
def test_relationship_on_or(self):
bars, foos = self.tables.bars, self.tables.foos
class Foo(fixtures.ComparableEntity):
pass
class Bar(fixtures.ComparableEntity):
pass
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=sa.or_(bars.c.id == foos.c.bid1,
bars.c.id == foos.c.bid2),
uselist=True,
viewonly=True)})
mapper(Bar, bars)
sess = create_session()
b1 = Bar(id=1, data='b1')
b2 = Bar(id=2, data='b2')
b3 = Bar(id=3, data='b3')
f1 = Foo(bid1=1, bid2=2)
f2 = Foo(bid1=3, bid2=None)
sess.add_all((b1, b2, b3))
sess.flush()
sess.add_all((f1, f2))
sess.flush()
sess.expunge_all()
eq_(sess.query(Foo).filter_by(id=f1.id).one(),
Foo(bars=[Bar(data='b1'), Bar(data='b2')]))
eq_(sess.query(Foo).filter_by(id=f2.id).one(),
Foo(bars=[Bar(data='b3')]))
class ViewOnlyRepeatedLocalColumn(fixtures.MappedTest):
"""'viewonly' mappings that contain the same 'local' column twice"""
@classmethod
def define_tables(cls, metadata):
Table('foos', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)))
Table('bars', metadata, Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('fid1', Integer, ForeignKey('foos.id')),
Column('fid2', Integer, ForeignKey('foos.id')),
Column('data', String(50)))
def test_relationship_on_or(self):
bars, foos = self.tables.bars, self.tables.foos
class Foo(fixtures.ComparableEntity):
pass
class Bar(fixtures.ComparableEntity):
pass
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=sa.or_(bars.c.fid1 == foos.c.id,
bars.c.fid2 == foos.c.id),
viewonly=True)})
mapper(Bar, bars)
sess = create_session()
f1 = Foo(id=1, data='f1')
f2 = Foo(id=2, data='f2')
b1 = Bar(fid1=1, data='b1')
b2 = Bar(fid2=1, data='b2')
b3 = Bar(fid1=2, data='b3')
b4 = Bar(fid1=1, fid2=2, data='b4')
sess.add_all((f1, f2))
sess.flush()
sess.add_all((b1, b2, b3, b4))
sess.flush()
sess.expunge_all()
eq_(sess.query(Foo).filter_by(id=f1.id).one(),
Foo(bars=[Bar(data='b1'), Bar(data='b2'), Bar(data='b4')]))
eq_(sess.query(Foo).filter_by(id=f2.id).one(),
Foo(bars=[Bar(data='b3'), Bar(data='b4')]))
class ViewOnlyComplexJoin(_RelationshipErrors, fixtures.MappedTest):
"""'viewonly' mappings with a complex join condition."""
@classmethod
def define_tables(cls, metadata):
Table('t1', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)))
Table('t2', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)),
Column('t1id', Integer, ForeignKey('t1.id')))
Table('t3', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)))
Table('t2tot3', metadata,
Column('t2id', Integer, ForeignKey('t2.id')),
Column('t3id', Integer, ForeignKey('t3.id')))
@classmethod
def setup_classes(cls):
class T1(cls.Comparable):
pass
class T2(cls.Comparable):
pass
class T3(cls.Comparable):
pass
def test_basic(self):
T1, t2, T2, T3, t3, t2tot3, t1 = (self.classes.T1,
self.tables.t2,
self.classes.T2,
self.classes.T3,
self.tables.t3,
self.tables.t2tot3,
self.tables.t1)
mapper(T1, t1, properties={
't3s': relationship(T3, primaryjoin=sa.and_(
t1.c.id == t2.c.t1id,
t2.c.id == t2tot3.c.t2id,
t3.c.id == t2tot3.c.t3id),
viewonly=True,
foreign_keys=t3.c.id, remote_side=t2.c.t1id)
})
mapper(T2, t2, properties={
't1': relationship(T1),
't3s': relationship(T3, secondary=t2tot3)
})
mapper(T3, t3)
sess = create_session()
sess.add(T2(data='t2', t1=T1(data='t1'), t3s=[T3(data='t3')]))
sess.flush()
sess.expunge_all()
a = sess.query(T1).first()
eq_(a.t3s, [T3(data='t3')])
def test_remote_side_escalation(self):
T1, t2, T2, T3, t3, t2tot3, t1 = (self.classes.T1,
self.tables.t2,
self.classes.T2,
self.classes.T3,
self.tables.t3,
self.tables.t2tot3,
self.tables.t1)
mapper(T1, t1, properties={
't3s': relationship(T3,
primaryjoin=sa.and_(t1.c.id == t2.c.t1id,
t2.c.id == t2tot3.c.t2id,
t3.c.id == t2tot3.c.t3id
),
viewonly=True,
foreign_keys=t3.c.id)})
mapper(T2, t2, properties={
't1': relationship(T1),
't3s': relationship(T3, secondary=t2tot3)})
mapper(T3, t3)
self._assert_raises_no_local_remote(configure_mappers, "T1.t3s")
class RemoteForeignBetweenColsTest(fixtures.DeclarativeMappedTest):
"""test a complex annotation using between().
Using declarative here as an integration test for the local()
and remote() annotations in conjunction with already annotated
instrumented attributes, etc.
"""
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
class Network(fixtures.ComparableEntity, Base):
__tablename__ = "network"
id = Column(sa.Integer, primary_key=True,
test_needs_autoincrement=True)
ip_net_addr = Column(Integer)
ip_broadcast_addr = Column(Integer)
addresses = relationship(
"Address",
primaryjoin="remote(foreign(Address.ip_addr)).between("
"Network.ip_net_addr,"
"Network.ip_broadcast_addr)",
viewonly=True
)
class Address(fixtures.ComparableEntity, Base):
__tablename__ = "address"
ip_addr = Column(Integer, primary_key=True)
@classmethod
def insert_data(cls):
Network, Address = cls.classes.Network, cls.classes.Address
s = Session(testing.db)
s.add_all([
Network(ip_net_addr=5, ip_broadcast_addr=10),
Network(ip_net_addr=15, ip_broadcast_addr=25),
Network(ip_net_addr=30, ip_broadcast_addr=35),
Address(ip_addr=17), Address(ip_addr=18), Address(ip_addr=9),
Address(ip_addr=27)
])
s.commit()
def test_col_query(self):
Network, Address = self.classes.Network, self.classes.Address
session = Session(testing.db)
eq_(
session.query(Address.ip_addr).
select_from(Network).
join(Network.addresses).
filter(Network.ip_net_addr == 15).
all(),
[(17, ), (18, )]
)
def test_lazyload(self):
Network, Address = self.classes.Network, self.classes.Address
session = Session(testing.db)
n3 = session.query(Network).filter(Network.ip_net_addr == 5).one()
eq_([a.ip_addr for a in n3.addresses], [9])
class ExplicitLocalRemoteTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('t1', metadata,
Column('id', String(50), primary_key=True),
Column('data', String(50)))
Table('t2', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('data', String(50)),
Column('t1id', String(50)))
@classmethod
def setup_classes(cls):
class T1(cls.Comparable):
pass
class T2(cls.Comparable):
pass
def test_onetomany_funcfk_oldstyle(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
# old _local_remote_pairs
mapper(T1, t1, properties={
't2s': relationship(
T2,
primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t1.c.id, t2.c.t1id)],
foreign_keys=[t2.c.t1id]
)
})
mapper(T2, t2)
self._test_onetomany()
def test_onetomany_funcfk_annotated(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
# use annotation
mapper(T1, t1, properties={
't2s': relationship(T2,
primaryjoin=t1.c.id ==
foreign(sa.func.lower(t2.c.t1id)),
)})
mapper(T2, t2)
self._test_onetomany()
def _test_onetomany(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
is_(T1.t2s.property.direction, ONETOMANY)
eq_(T1.t2s.property.local_remote_pairs, [(t1.c.id, t2.c.t1id)])
sess = create_session()
a1 = T1(id='number1', data='a1')
a2 = T1(id='number2', data='a2')
b1 = T2(data='b1', t1id='NuMbEr1')
b2 = T2(data='b2', t1id='Number1')
b3 = T2(data='b3', t1id='Number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
sess.expunge_all()
eq_(sess.query(T1).first(),
T1(id='number1', data='a1', t2s=[
T2(data='b1', t1id='NuMbEr1'),
T2(data='b2', t1id='Number1')]))
def test_manytoone_funcfk(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
mapper(T1, t1)
mapper(T2, t2, properties={
't1': relationship(T1,
primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t2.c.t1id, t1.c.id)],
foreign_keys=[t2.c.t1id],
uselist=True)})
sess = create_session()
a1 = T1(id='number1', data='a1')
a2 = T1(id='number2', data='a2')
b1 = T2(data='b1', t1id='NuMbEr1')
b2 = T2(data='b2', t1id='Number1')
b3 = T2(data='b3', t1id='Number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
sess.expunge_all()
eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
[T2(data='b1', t1=[T1(id='number1', data='a1')]),
T2(data='b2', t1=[T1(id='number1', data='a1')])])
def test_onetomany_func_referent(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
mapper(T1, t1, properties={
't2s': relationship(
T2,
primaryjoin=sa.func.lower(t1.c.id) == t2.c.t1id,
_local_remote_pairs=[(t1.c.id, t2.c.t1id)],
foreign_keys=[t2.c.t1id])})
mapper(T2, t2)
sess = create_session()
a1 = T1(id='NuMbeR1', data='a1')
a2 = T1(id='NuMbeR2', data='a2')
b1 = T2(data='b1', t1id='number1')
b2 = T2(data='b2', t1id='number1')
b3 = T2(data='b2', t1id='number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
sess.expunge_all()
eq_(sess.query(T1).first(),
T1(id='NuMbeR1', data='a1', t2s=[
T2(data='b1', t1id='number1'),
T2(data='b2', t1id='number1')]))
def test_manytoone_func_referent(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
mapper(T1, t1)
mapper(T2, t2, properties={
't1': relationship(T1,
primaryjoin=sa.func.lower(t1.c.id) == t2.c.t1id,
_local_remote_pairs=[(t2.c.t1id, t1.c.id)],
foreign_keys=[t2.c.t1id], uselist=True)})
sess = create_session()
a1 = T1(id='NuMbeR1', data='a1')
a2 = T1(id='NuMbeR2', data='a2')
b1 = T2(data='b1', t1id='number1')
b2 = T2(data='b2', t1id='number1')
b3 = T2(data='b3', t1id='number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
sess.expunge_all()
eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
[T2(data='b1', t1=[T1(id='NuMbeR1', data='a1')]),
T2(data='b2', t1=[T1(id='NuMbeR1', data='a1')])])
def test_escalation_1(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
mapper(T1, t1, properties={
't2s': relationship(
T2,
primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t1.c.id, t2.c.t1id)],
foreign_keys=[t2.c.t1id],
remote_side=[t2.c.t1id])})
mapper(T2, t2)
assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
def test_escalation_2(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
mapper(T1, t1, properties={
't2s': relationship(
T2,
primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t1.c.id, t2.c.t1id)])})
mapper(T2, t2)
assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
class InvalidRemoteSideTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('t1', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(50)),
Column('t_id', Integer, ForeignKey('t1.id'))
)
@classmethod
def setup_classes(cls):
class T1(cls.Comparable):
pass
def test_o2m_backref(self):
T1, t1 = self.classes.T1, self.tables.t1
mapper(T1, t1, properties={
't1s': relationship(T1, backref='parent')
})
assert_raises_message(
sa.exc.ArgumentError,
"T1.t1s and back-reference T1.parent are "
r"both of the same direction symbol\('ONETOMANY'\). Did you "
"mean to set remote_side on the many-to-one side ?",
configure_mappers)
def test_m2o_backref(self):
T1, t1 = self.classes.T1, self.tables.t1
mapper(T1, t1, properties={
't1s': relationship(T1,
backref=backref('parent', remote_side=t1.c.id),
remote_side=t1.c.id)
})
assert_raises_message(
sa.exc.ArgumentError,
"T1.t1s and back-reference T1.parent are "
r"both of the same direction symbol\('MANYTOONE'\). Did you "
"mean to set remote_side on the many-to-one side ?",
configure_mappers)
def test_o2m_explicit(self):
T1, t1 = self.classes.T1, self.tables.t1
mapper(T1, t1, properties={
't1s': relationship(T1, back_populates='parent'),
'parent': relationship(T1, back_populates='t1s'),
})
# can't be sure of ordering here
assert_raises_message(
sa.exc.ArgumentError,
r"both of the same direction symbol\('ONETOMANY'\). Did you "
"mean to set remote_side on the many-to-one side ?",
configure_mappers)
def test_m2o_explicit(self):
T1, t1 = self.classes.T1, self.tables.t1
mapper(T1, t1, properties={
't1s': relationship(T1, back_populates='parent',
remote_side=t1.c.id),
'parent': relationship(T1, back_populates='t1s',
remote_side=t1.c.id)
})
# can't be sure of ordering here
assert_raises_message(
sa.exc.ArgumentError,
r"both of the same direction symbol\('MANYTOONE'\). Did you "
"mean to set remote_side on the many-to-one side ?",
configure_mappers)
class AmbiguousFKResolutionTest(_RelationshipErrors, fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table("a", metadata,
Column('id', Integer, primary_key=True)
)
Table("b", metadata,
Column('id', Integer, primary_key=True),
Column('aid_1', Integer, ForeignKey('a.id')),
Column('aid_2', Integer, ForeignKey('a.id')),
)
Table("atob", metadata,
Column('aid', Integer),
Column('bid', Integer),
)
Table("atob_ambiguous", metadata,
Column('aid1', Integer, ForeignKey('a.id')),
Column('bid1', Integer, ForeignKey('b.id')),
Column('aid2', Integer, ForeignKey('a.id')),
Column('bid2', Integer, ForeignKey('b.id')),
)
@classmethod
def setup_classes(cls):
class A(cls.Basic):
pass
class B(cls.Basic):
pass
def test_ambiguous_fks_o2m(self):
A, B = self.classes.A, self.classes.B
a, b = self.tables.a, self.tables.b
mapper(A, a, properties={
'bs': relationship(B)
})
mapper(B, b)
self._assert_raises_ambig_join(
configure_mappers,
"A.bs",
None
)
def test_with_fks_o2m(self):
A, B = self.classes.A, self.classes.B
a, b = self.tables.a, self.tables.b
mapper(A, a, properties={
'bs': relationship(B, foreign_keys=b.c.aid_1)
})
mapper(B, b)
sa.orm.configure_mappers()
assert A.bs.property.primaryjoin.compare(
a.c.id == b.c.aid_1
)
eq_(
A.bs.property._calculated_foreign_keys,
set([b.c.aid_1])
)
def test_with_pj_o2m(self):
A, B = self.classes.A, self.classes.B
a, b = self.tables.a, self.tables.b
mapper(A, a, properties={
'bs': relationship(B, primaryjoin=a.c.id == b.c.aid_1)
})
mapper(B, b)
sa.orm.configure_mappers()
assert A.bs.property.primaryjoin.compare(
a.c.id == b.c.aid_1
)
eq_(
A.bs.property._calculated_foreign_keys,
set([b.c.aid_1])
)
def test_with_annotated_pj_o2m(self):
A, B = self.classes.A, self.classes.B
a, b = self.tables.a, self.tables.b
mapper(A, a, properties={
'bs': relationship(B, primaryjoin=a.c.id == foreign(b.c.aid_1))
})
mapper(B, b)
sa.orm.configure_mappers()
assert A.bs.property.primaryjoin.compare(
a.c.id == b.c.aid_1
)
eq_(
A.bs.property._calculated_foreign_keys,
set([b.c.aid_1])
)
def test_no_fks_m2m(self):
A, B = self.classes.A, self.classes.B
a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob
mapper(A, a, properties={
'bs': relationship(B, secondary=a_to_b)
})
mapper(B, b)
self._assert_raises_no_join(
sa.orm.configure_mappers,
"A.bs", a_to_b,
)
def test_ambiguous_fks_m2m(self):
A, B = self.classes.A, self.classes.B
a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
mapper(A, a, properties={
'bs': relationship(B, secondary=a_to_b)
})
mapper(B, b)
self._assert_raises_ambig_join(
configure_mappers,
"A.bs",
"atob_ambiguous"
)
def test_with_fks_m2m(self):
A, B = self.classes.A, self.classes.B
a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
mapper(A, a, properties={
'bs': relationship(B, secondary=a_to_b,
foreign_keys=[a_to_b.c.aid1, a_to_b.c.bid1])
})
mapper(B, b)
sa.orm.configure_mappers()
class SecondaryNestedJoinTest(fixtures.MappedTest, AssertsCompiledSQL,
testing.AssertsExecutionResults):
"""test support for a relationship where the 'secondary' table is a
compound join().
join() and joinedload() should use a "flat" alias, lazyloading needs
to ensure the join renders.
"""
run_setup_mappers = 'once'
run_inserts = 'once'
run_deletes = None
@classmethod
def define_tables(cls, metadata):
Table(
'a', metadata,
Column(
'id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30)),
Column('b_id', ForeignKey('b.id'))
)
Table('b', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30)),
Column('d_id', ForeignKey('d.id'))
)
Table('c', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30)),
Column('a_id', ForeignKey('a.id')),
Column('d_id', ForeignKey('d.id'))
)
Table('d', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30)),
)
@classmethod
def setup_classes(cls):
class A(cls.Comparable):
pass
class B(cls.Comparable):
pass
class C(cls.Comparable):
pass
class D(cls.Comparable):
pass
@classmethod
def setup_mappers(cls):
A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
a, b, c, d = cls.tables.a, cls.tables.b, cls.tables.c, cls.tables.d
j = sa.join(b, d, b.c.d_id == d.c.id).join(c, c.c.d_id == d.c.id)
#j = join(b, d, b.c.d_id == d.c.id).join(c, c.c.d_id == d.c.id).alias()
mapper(A, a, properties={
"b": relationship(B),
"d": relationship(
D, secondary=j,
primaryjoin=and_(a.c.b_id == b.c.id, a.c.id == c.c.a_id),
secondaryjoin=d.c.id == b.c.d_id,
#primaryjoin=and_(a.c.b_id == j.c.b_id, a.c.id == j.c.c_a_id),
#secondaryjoin=d.c.id == j.c.b_d_id,
uselist=False,
viewonly=True
)
})
mapper(B, b, properties={
"d": relationship(D)
})
mapper(C, c, properties={
"a": relationship(A),
"d": relationship(D)
})
mapper(D, d)
@classmethod
def insert_data(cls):
A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
sess = Session()
a1, a2, a3, a4 = A(name='a1'), A(name='a2'), A(name='a3'), A(name='a4')
b1, b2, b3, b4 = B(name='b1'), B(name='b2'), B(name='b3'), B(name='b4')
c1, c2, c3, c4 = C(name='c1'), C(name='c2'), C(name='c3'), C(name='c4')
d1, d2 = D(name='d1'), D(name='d2')
a1.b = b1
a2.b = b2
a3.b = b3
a4.b = b4
c1.a = a1
c2.a = a2
c3.a = a2
c4.a = a4
c1.d = d1
c2.d = d2
c3.d = d1
c4.d = d2
b1.d = d1
b2.d = d1
b3.d = d2
b4.d = d2
sess.add_all([a1, a2, a3, a4, b1, b2, b3, b4, c1, c2, c4, c4, d1, d2])
sess.commit()
def test_render_join(self):
A, D = self.classes.A, self.classes.D
sess = Session()
self.assert_compile(
sess.query(A).join(A.d),
"SELECT a.id AS a_id, a.name AS a_name, a.b_id AS a_b_id "
"FROM a JOIN (b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id "
"JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id "
"AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id",
dialect="postgresql"
)
def test_render_joinedload(self):
A, D = self.classes.A, self.classes.D
sess = Session()
self.assert_compile(
sess.query(A).options(joinedload(A.d)),
"SELECT a.id AS a_id, a.name AS a_name, a.b_id AS a_b_id, "
"d_1.id AS d_1_id, d_1.name AS d_1_name FROM a LEFT OUTER JOIN "
"(b AS b_1 JOIN d AS d_2 ON b_1.d_id = d_2.id JOIN c AS c_1 "
"ON c_1.d_id = d_2.id JOIN d AS d_1 ON d_1.id = b_1.d_id) "
"ON a.b_id = b_1.id AND a.id = c_1.a_id",
dialect="postgresql"
)
def test_render_lazyload(self):
from sqlalchemy.testing.assertsql import CompiledSQL
A, D = self.classes.A, self.classes.D
sess = Session()
a1 = sess.query(A).filter(A.name == 'a1').first()
def go():
a1.d
# here, the "lazy" strategy has to ensure the "secondary"
# table is part of the "select_from()", since it's a join().
# referring to just the columns wont actually render all those
# join conditions.
self.assert_sql_execution(
testing.db,
go,
CompiledSQL(
"SELECT d.id AS d_id, d.name AS d_name FROM b "
"JOIN d ON b.d_id = d.id JOIN c ON c.d_id = d.id "
"WHERE :param_1 = b.id AND :param_2 = c.a_id "
"AND d.id = b.d_id",
{'param_1': a1.id, 'param_2': a1.id}
)
)
mapping = {
"a1": "d1",
"a2": None,
"a3": None,
"a4": "d2"
}
def test_join(self):
A, D = self.classes.A, self.classes.D
sess = Session()
for a, d in sess.query(A, D).outerjoin(A.d):
eq_(self.mapping[a.name], d.name if d is not None else None)
def test_joinedload(self):
A, D = self.classes.A, self.classes.D
sess = Session()
for a in sess.query(A).options(joinedload(A.d)):
d = a.d
eq_(self.mapping[a.name], d.name if d is not None else None)
def test_lazyload(self):
A, D = self.classes.A, self.classes.D
sess = Session()
for a in sess.query(A):
d = a.d
eq_(self.mapping[a.name], d.name if d is not None else None)
class InvalidRelationshipEscalationTest(
_RelationshipErrors, fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('foos', metadata,
Column('id', Integer, primary_key=True),
Column('fid', Integer))
Table('bars', metadata,
Column('id', Integer, primary_key=True),
Column('fid', Integer))
Table('foos_with_fks', metadata,
Column('id', Integer, primary_key=True),
Column('fid', Integer, ForeignKey('foos_with_fks.id')))
Table('bars_with_fks', metadata,
Column('id', Integer, primary_key=True),
Column('fid', Integer, ForeignKey('foos_with_fks.id')))
@classmethod
def setup_classes(cls):
class Foo(cls.Basic):
pass
class Bar(cls.Basic):
pass
def test_no_join(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar)})
mapper(Bar, bars)
self._assert_raises_no_join(sa.orm.configure_mappers,
"Foo.bars", None
)
def test_no_join_self_ref(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'foos': relationship(Foo)})
mapper(Bar, bars)
self._assert_raises_no_join(
configure_mappers,
"Foo.foos",
None
)
def test_no_equated(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id > bars.c.fid)})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
configure_mappers,
"foos.id > bars.fid", "Foo.bars", "primary"
)
def test_no_equated_fks(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id > bars.c.fid,
foreign_keys=bars.c.fid)})
mapper(Bar, bars)
self._assert_raises_no_equality(
sa.orm.configure_mappers,
"foos.id > bars.fid", "Foo.bars", "primary"
)
def test_no_equated_wo_fks_works_on_relaxed(self):
foos_with_fks, Foo, Bar, bars_with_fks, foos = (
self.tables.foos_with_fks,
self.classes.Foo,
self.classes.Bar,
self.tables.bars_with_fks,
self.tables.foos)
# very unique - the join between parent/child
# has no fks, but there is an fk join between two other
# tables in the join condition, for those users that try creating
# these big-long-string-of-joining-many-tables primaryjoins.
# in this case we don't get eq_pairs, but we hit the
# "works if viewonly" rule. so here we add another clause regarding
# "try foreign keys".
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=and_(
bars_with_fks.c.fid == foos_with_fks.c.id,
foos_with_fks.c.id == foos.c.id,
)
)})
mapper(Bar, bars_with_fks)
self._assert_raises_no_equality(
sa.orm.configure_mappers,
"bars_with_fks.fid = foos_with_fks.id "
"AND foos_with_fks.id = foos.id",
"Foo.bars", "primary"
)
def test_ambiguous_fks(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id == bars.c.fid,
foreign_keys=[foos.c.id, bars.c.fid])})
mapper(Bar, bars)
self._assert_raises_ambiguous_direction(
sa.orm.configure_mappers,
"Foo.bars"
)
def test_ambiguous_remoteside_o2m(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id == bars.c.fid,
foreign_keys=[bars.c.fid],
remote_side=[foos.c.id, bars.c.fid],
viewonly=True
)})
mapper(Bar, bars)
self._assert_raises_no_local_remote(
configure_mappers,
"Foo.bars",
)
def test_ambiguous_remoteside_m2o(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id == bars.c.fid,
foreign_keys=[foos.c.id],
remote_side=[foos.c.id, bars.c.fid],
viewonly=True
)})
mapper(Bar, bars)
self._assert_raises_no_local_remote(
configure_mappers,
"Foo.bars",
)
def test_no_equated_self_ref_no_fks(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id > foos.c.fid)})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
configure_mappers,
"foos.id > foos.fid", "Foo.foos", "primary"
)
def test_no_equated_self_ref_no_equality(self):
bars, Foo, Bar, foos = (self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id > foos.c.fid,
foreign_keys=[foos.c.fid])})
mapper(Bar, bars)
self._assert_raises_no_equality(configure_mappers,
"foos.id > foos.fid", "Foo.foos", "primary"
)
def test_no_equated_viewonly(self):
bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
self.tables.bars,
self.classes.Bar,
self.tables.bars_with_fks,
self.tables.foos_with_fks,
self.classes.Foo,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id > bars.c.fid,
viewonly=True)})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
sa.orm.configure_mappers,
"foos.id > bars.fid", "Foo.bars", "primary"
)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
'bars': relationship(
Bar,
primaryjoin=foos_with_fks.c.id > bars_with_fks.c.fid,
viewonly=True)})
mapper(Bar, bars_with_fks)
sa.orm.configure_mappers()
def test_no_equated_self_ref_viewonly(self):
bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
self.tables.bars,
self.classes.Bar,
self.tables.bars_with_fks,
self.tables.foos_with_fks,
self.classes.Foo,
self.tables.foos)
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id > foos.c.fid,
viewonly=True)})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
sa.orm.configure_mappers,
"foos.id > foos.fid", "Foo.foos", "primary"
)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
'foos': relationship(
Foo,
primaryjoin=foos_with_fks.c.id > foos_with_fks.c.fid,
viewonly=True)})
mapper(Bar, bars_with_fks)
sa.orm.configure_mappers()
def test_no_equated_self_ref_viewonly_fks(self):
Foo, foos = self.classes.Foo, self.tables.foos
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id > foos.c.fid,
viewonly=True,
foreign_keys=[foos.c.fid])})
sa.orm.configure_mappers()
eq_(Foo.foos.property.local_remote_pairs, [(foos.c.id, foos.c.fid)])
def test_equated(self):
bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
self.tables.bars,
self.classes.Bar,
self.tables.bars_with_fks,
self.tables.foos_with_fks,
self.classes.Foo,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
primaryjoin=foos.c.id == bars.c.fid)})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
configure_mappers,
"foos.id = bars.fid", "Foo.bars", "primary"
)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
'bars': relationship(
Bar,
primaryjoin=foos_with_fks.c.id == bars_with_fks.c.fid)})
mapper(Bar, bars_with_fks)
sa.orm.configure_mappers()
def test_equated_self_ref(self):
Foo, foos = self.classes.Foo, self.tables.foos
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id == foos.c.fid)})
self._assert_raises_no_relevant_fks(
configure_mappers,
"foos.id = foos.fid", "Foo.foos", "primary"
)
def test_equated_self_ref_wrong_fks(self):
bars, Foo, foos = (self.tables.bars,
self.classes.Foo,
self.tables.foos)
mapper(Foo, foos, properties={
'foos': relationship(Foo,
primaryjoin=foos.c.id == foos.c.fid,
foreign_keys=[bars.c.id])})
self._assert_raises_no_relevant_fks(
configure_mappers,
"foos.id = foos.fid", "Foo.foos", "primary"
)
class InvalidRelationshipEscalationTestM2M(
_RelationshipErrors, fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('foos', metadata,
Column('id', Integer, primary_key=True))
Table('foobars', metadata,
Column('fid', Integer), Column('bid', Integer))
Table('bars', metadata,
Column('id', Integer, primary_key=True))
Table('foobars_with_fks', metadata,
Column('fid', Integer, ForeignKey('foos.id')),
Column('bid', Integer, ForeignKey('bars.id'))
)
Table('foobars_with_many_columns', metadata,
Column('fid', Integer),
Column('bid', Integer),
Column('fid1', Integer),
Column('bid1', Integer),
Column('fid2', Integer),
Column('bid2', Integer),
)
@classmethod
def setup_classes(cls):
class Foo(cls.Basic):
pass
class Bar(cls.Basic):
pass
def test_no_join(self):
foobars, bars, Foo, Bar, foos = (self.tables.foobars,
self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar, secondary=foobars)})
mapper(Bar, bars)
self._assert_raises_no_join(
configure_mappers,
"Foo.bars",
"foobars"
)
def test_no_secondaryjoin(self):
foobars, bars, Foo, Bar, foos = (self.tables.foobars,
self.tables.bars,
self.classes.Foo,
self.classes.Bar,
self.tables.foos)
mapper(Foo, foos, properties={
'bars': relationship(Bar,
secondary=foobars,
primaryjoin=foos.c.id > foobars.c.fid)})
mapper(Bar, bars)
self._assert_raises_no_join(
configure_mappers,
"Foo.bars",
"foobars"
)
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,
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)
sa.orm.configure_mappers()
eq_(
Foo.bars.property.synchronize_pairs,
[(foos.c.id, foobars.c.fid)]
)
eq_(
Foo.bars.property.secondary_synchronize_pairs,
[(bars.c.id, foobars.c.bid)]
)
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)
sa.orm.configure_mappers()
eq_(
Foo.bars.property.synchronize_pairs,
[(foos.c.id, foobars_with_many_columns.c.fid)]
)
eq_(
Foo.bars.property.secondary_synchronize_pairs,
[(bars.c.id, foobars_with_many_columns.c.bid)]
)
def test_local_col_setup(self):
foobars_with_fks, bars, Bar, Foo, foos = (
self.tables.foobars_with_fks,
self.tables.bars,
self.classes.Bar,
self.classes.Foo,
self.tables.foos)
# ensure m2m backref is set up with correct annotations
# [ticket:2578]
mapper(Foo, foos, properties={
'bars': relationship(Bar, secondary=foobars_with_fks, backref="foos")
})
mapper(Bar, bars)
sa.orm.configure_mappers()
eq_(
Foo.bars.property._join_condition.local_columns,
set([foos.c.id])
)
eq_(
Bar.foos.property._join_condition.local_columns,
set([bars.c.id])
)
def test_bad_primaryjoin(self):
foobars_with_fks, bars, Bar, foobars, Foo, foos = (
self.tables.foobars_with_fks,
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)
self._assert_raises_no_equality(
configure_mappers,
'foos.id > foobars.fid',
"Foo.bars",
"primary")
sa.orm.clear_mappers()
mapper(Foo, foos, properties={
'bars': relationship(
Bar,
secondary=foobars_with_fks,
primaryjoin=foos.c.id > foobars_with_fks.c.fid,
secondaryjoin=foobars_with_fks.c.bid <= bars.c.id)})
mapper(Bar, bars)
self._assert_raises_no_equality(
configure_mappers,
'foos.id > foobars_with_fks.fid',
"Foo.bars",
"primary")
sa.orm.clear_mappers()
mapper(Foo, foos, properties={
'bars': relationship(
Bar,
secondary=foobars_with_fks,
primaryjoin=foos.c.id > foobars_with_fks.c.fid,
secondaryjoin=foobars_with_fks.c.bid <= bars.c.id,
viewonly=True)})
mapper(Bar, bars)
sa.orm.configure_mappers()
def test_bad_secondaryjoin(self):
foobars, bars, Foo, Bar, foos = (self.tables.foobars,
self.tables.bars,
self.classes.Foo,
self.classes.Bar,
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,
foreign_keys=[foobars.c.fid])})
mapper(Bar, bars)
self._assert_raises_no_relevant_fks(
configure_mappers,
"foobars.bid <= bars.id",
"Foo.bars",
"secondary"
)
def test_no_equated_secondaryjoin(self):
foobars, bars, Foo, Bar, foos = (self.tables.foobars,
self.tables.bars,
self.classes.Foo,
self.classes.Bar,
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,
foreign_keys=[foobars.c.fid, foobars.c.bid])})
mapper(Bar, bars)
self._assert_raises_no_equality(
configure_mappers,
"foobars.bid <= bars.id",
"Foo.bars",
"secondary"
)
class ActiveHistoryFlagTest(_fixtures.FixtureTest):
run_inserts = None
run_deletes = None
def _test_attribute(self, obj, attrname, newvalue):
sess = Session()
sess.add(obj)
oldvalue = getattr(obj, attrname)
sess.commit()
# expired
assert attrname not in obj.__dict__
setattr(obj, attrname, newvalue)
eq_(
attributes.get_history(obj, attrname),
([newvalue, ], (), [oldvalue, ])
)
def test_column_property_flag(self):
User, users = self.classes.User, self.tables.users
mapper(User, users, properties={
'name': column_property(users.c.name,
active_history=True)
})
u1 = User(name='jack')
self._test_attribute(u1, 'name', 'ed')
def test_relationship_property_flag(self):
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
mapper(Address, addresses, properties={
'user': relationship(User, active_history=True)
})
mapper(User, users)
u1 = User(name='jack')
u2 = User(name='ed')
a1 = Address(email_address='a1', user=u1)
self._test_attribute(a1, 'user', u2)
def test_composite_property_flag(self):
Order, orders = self.classes.Order, self.tables.orders
class MyComposite(object):
def __init__(self, description, isopen):
self.description = description
self.isopen = isopen
def __composite_values__(self):
return [self.description, self.isopen]
def __eq__(self, other):
return isinstance(other, MyComposite) and \
other.description == self.description
mapper(Order, orders, properties={
'composite': composite(
MyComposite,
orders.c.description,
orders.c.isopen,
active_history=True)
})
o1 = Order(composite=MyComposite('foo', 1))
self._test_attribute(o1, "composite", MyComposite('bar', 1))
class RelationDeprecationTest(fixtures.MappedTest):
"""test usage of the old 'relation' function."""
run_inserts = 'once'
run_deletes = None
@classmethod
def define_tables(cls, metadata):
Table('users_table', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(64)))
Table('addresses_table', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', Integer, ForeignKey('users_table.id')),
Column('email_address', String(128)),
Column('purpose', String(16)),
Column('bounces', Integer, default=0))
@classmethod
def setup_classes(cls):
class User(cls.Basic):
pass
class Address(cls.Basic):
pass
@classmethod
def fixtures(cls):
return dict(
users_table=(
('id', 'name'),
(1, 'jack'),
(2, 'ed'),
(3, 'fred'),
(4, 'chuck')),
addresses_table=(
('id', 'user_id', 'email_address', 'purpose', 'bounces'),
(1, 1, 'jack@jack.home', 'Personal', 0),
(2, 1, 'jack@jack.bizz', 'Work', 1),
(3, 2, 'ed@foo.bar', 'Personal', 0),
(4, 3, 'fred@the.fred', 'Personal', 10)))
def test_relation(self):
addresses_table, User, users_table, Address = (
self.tables.addresses_table,
self.classes.User,
self.tables.users_table,
self.classes.Address)
mapper(User, users_table, properties=dict(
addresses=relation(Address, backref='user'),
))
mapper(Address, addresses_table)
session = create_session()
session.query(User).filter(User.addresses.any(
Address.email_address == 'ed@foo.bar')).one()
Reference in New Issue View Git Blame Copy Permalink