@classmethod
def define_tables(cls, metadata):
- global people, engineers
- people = Table('people', metadata,
+ Table('people', metadata,
Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)),
Column('type', String(30)))
- engineers = Table('engineers', metadata,
+ Table('engineers', metadata,
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('primary_language', String(50)),
Column('reports_to_id', Integer, ForeignKey('people.person_id'))
)
@classmethod
+ @testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
mapper(Engineer, engineers, inherits=Person,
'reports_to':relationship(Person, primaryjoin=people.c.person_id==engineers.c.reports_to_id)
})
+ @testing.resolve_artifact_names
def test_has(self):
p1 = Person(name='dogbert')
eq_(sess.query(Engineer).filter(Engineer.reports_to.has(Person.name=='dogbert')).first(), Engineer(name='dilbert'))
+ @testing.resolve_artifact_names
def test_oftype_aliases_in_exists(self):
e1 = Engineer(name='dilbert', primary_language='java')
e2 = Engineer(name='wally', primary_language='c++', reports_to=e1)
eq_(sess.query(Engineer).filter(Engineer.reports_to.of_type(Engineer).has(Engineer.name=='dilbert')).first(), e2)
+ @testing.resolve_artifact_names
def test_join(self):
p1 = Person(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1)
@classmethod
def define_tables(cls, metadata):
- global people, engineers, managers
people = Table('people', metadata,
Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)),
)
@classmethod
+ @testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
mapper(Manager, managers, inherits=Person, polymorphic_identity='manager')
'reports_to':relationship(Manager, primaryjoin=managers.c.person_id==engineers.c.reports_to_id, backref='engineers')
})
+ @testing.resolve_artifact_names
def test_has(self):
m1 = Manager(name='dogbert')
eq_(sess.query(Engineer).filter(Engineer.reports_to.has(Manager.name=='dogbert')).first(), Engineer(name='dilbert'))
+ @testing.resolve_artifact_names
def test_join(self):
m1 = Manager(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1)
sess.query(Engineer).join('reports_to', aliased=True).filter(Manager.name=='dogbert').first(),
Engineer(name='dilbert'))
+ @testing.resolve_artifact_names
def test_filter_aliasing(self):
m1 = Manager(name='dogbert')
m2 = Manager(name='foo')
]
)
+ @testing.resolve_artifact_names
def test_relationship_compare(self):
m1 = Manager(name='dogbert')
m2 = Manager(name='foo')
@classmethod
def define_tables(cls, metadata):
- global people, engineers, organizations, engineers_to_org
-
organizations = Table('organizations', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)),
)
- engineers_to_org = Table('engineers_org', metadata,
+ engineers_to_org = Table('engineers_to_org', metadata,
Column('org_id', Integer, ForeignKey('organizations.id')),
Column('engineer_id', Integer, ForeignKey('engineers.person_id')),
)
)
@classmethod
+ @testing.resolve_artifact_names
def setup_mappers(cls):
- global Organization
class Organization(_fixtures.Base):
pass
mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer')
@classmethod
+ @testing.resolve_artifact_names
def insert_data(cls):
e1 = Engineer(name='e1')
e2 = Engineer(name='e2')
sess.add(org2)
sess.flush()
+ @testing.resolve_artifact_names
def test_not_contains(self):
sess = create_session()
- e1 = sess.query(Person).filter(Engineer.name=='e1').one()
+ e1 = sess.query(Person).filter(Engineer.name == 'e1').one()
# this works
- eq_(sess.query(Organization).filter(~Organization.engineers.of_type(Engineer).contains(e1)).all(), [Organization(name='org2')])
+
+ eq_(sess.query(Organization).filter(
+ ~Organization.engineers.of_type(Engineer).contains(e1)
+ ).all(),
+ [Organization(name='org2')])
# this had a bug
- eq_(sess.query(Organization).filter(~Organization.engineers.contains(e1)).all(), [Organization(name='org2')])
+ eq_(sess.query(Organization).filter(~Organization.engineers.contains(e1)).all(),
+ [Organization(name='org2')])
+
+ @testing.resolve_artifact_names
def test_any(self):
sess = create_session()
- eq_(sess.query(Organization).filter(Organization.engineers.of_type(Engineer).any(Engineer.name=='e1')).all(), [Organization(name='org1')])
- eq_(sess.query(Organization).filter(Organization.engineers.any(Engineer.name=='e1')).all(), [Organization(name='org1')])
+ eq_(sess.query(Organization).filter(Organization.engineers.of_type(Engineer).any(Engineer.name
+ == 'e1')).all(), [Organization(name='org1')])
+ eq_(sess.query(Organization).filter(Organization.engineers.any(Engineer.name
+ == 'e1')).all(), [Organization(name='org1')])
class SelfReferentialM2MTest(_base.MappedTest, AssertsCompiledSQL):
- run_setup_mappers = 'once'
-
@classmethod
def define_tables(cls, metadata):
- global Parent, Child1, Child2
-
- Base = declarative_base(metadata=metadata)
-
- secondary_table = Table('secondary', Base.metadata,
+ Table('secondary', metadata,
Column('left_id', Integer, ForeignKey('parent.id'), nullable=False),
Column('right_id', Integer, ForeignKey('parent.id'), nullable=False))
- class Parent(Base):
- __tablename__ = 'parent'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
- cls = Column(String(50))
- __mapper_args__ = dict(polymorphic_on = cls )
+ Table('parent', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('cls', String(50))
+ )
+ Table('child1', metadata,
+ Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
+ )
+ Table('child2', metadata,
+ Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
+ )
+ @classmethod
+ def setup_classes(cls):
+ class Parent(_base.BasicEntity):
+ pass
class Child1(Parent):
- __tablename__ = 'child1'
- id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
- __mapper_args__ = dict(polymorphic_identity = 'child1')
-
+ pass
class Child2(Parent):
- __tablename__ = 'child2'
- id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
- __mapper_args__ = dict(polymorphic_identity = 'child2')
-
- Child1.left_child2 = relationship(Child2, secondary = secondary_table,
- primaryjoin = Parent.id == secondary_table.c.right_id,
- secondaryjoin = Parent.id == secondary_table.c.left_id,
- uselist = False, backref="right_children"
+ pass
+
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_mappers(cls):
+ mapper(Parent, parent, polymorphic_on=parent.c.cls)
+ mapper(Child1, child1, inherits=Parent, polymorphic_identity='child1', properties={
+ 'left_child2' : relationship(Child2,
+ secondary = secondary,
+ primaryjoin = parent.c.id == secondary.c.right_id,
+ secondaryjoin = parent.c.id == secondary.c.left_id,
+ uselist = False, backref="right_children"
)
+ })
+ mapper(Child2, child2, inherits=Parent, polymorphic_identity = 'child2')
+ @testing.resolve_artifact_names
def test_query_crit(self):
session = create_session()
c11, c12, c13 = Child1(), Child1(), Child1()
dialect=default.DefaultDialect()
)
+ @testing.resolve_artifact_names
def test_eager_join(self):
session = create_session()
assert q.first() is c1
+ @testing.resolve_artifact_names
def test_subquery_load(self):
session = create_session()
class MultiplePathTest(_base.MappedTest, AssertsCompiledSQL):
@classmethod
def define_tables(cls, metadata):
- global t1, t2, t1t2_1, t1t2_2
t1 = Table('t1', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('data', String(30))
Column('t2id', Integer, ForeignKey('t2.id'))
)
+ @testing.resolve_artifact_names
def test_basic(self):
class T1(object):pass
class T2(object):pass
@classmethod
def define_tables(cls, metadata):
- global nodes
- nodes = Table('nodes', metadata,
+ Table('nodes', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('parent_id', Integer, ForeignKey('nodes.id')),
Column('data', String(30)))
@classmethod
- def insert_data(cls):
- # TODO: somehow using setup_classes()
- # here normally is screwing up the other tests.
-
- global Node, Sub
- class Node(Base):
- def append(self, node):
- self.children.append(node)
+ def setup_classes(cls):
+ class Node(Base):
+ def append(self, node):
+ self.children.append(node)
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_mappers(cls):
mapper(Node, nodes, properties={
'children':relationship(Node, lazy='select', join_depth=3,
backref=backref('parent', remote_side=[nodes.c.id])
),
})
+ @classmethod
+ @testing.resolve_artifact_names
+ def insert_data(cls):
sess = create_session()
n1 = Node(data='n1')
n1.append(Node(data='n11'))
@classmethod
def define_tables(cls, metadata):
- global nodes, node_to_nodes
nodes = Table('nodes', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('data', String(30)))
)
@classmethod
- def insert_data(cls):
- global Node
-
+ def setup_classes(cls):
class Node(Base):
pass
+ @classmethod
+ @testing.resolve_artifact_names
+ def insert_data(cls):
+
mapper(Node, nodes, properties={
'children':relationship(Node, lazy='select', secondary=node_to_nodes,
primaryjoin=nodes.c.id==node_to_nodes.c.left_node_id,
sess.flush()
sess.close()
+ @testing.resolve_artifact_names
def test_any(self):
sess = create_session()
- eq_(sess.query(Node).filter(Node.children.any(Node.data=='n3')).all(), [Node(data='n1'), Node(data='n2')])
+ eq_(sess.query(Node).filter(Node.children.any(Node.data == 'n3'
+ )).all(), [Node(data='n1'), Node(data='n2')])
+ @testing.resolve_artifact_names
def test_contains(self):
sess = create_session()
n4 = sess.query(Node).filter_by(data='n4').one()
- eq_(sess.query(Node).filter(Node.children.contains(n4)).order_by(Node.data).all(), [Node(data='n1'), Node(data='n3')])
- eq_(sess.query(Node).filter(not_(Node.children.contains(n4))).order_by(Node.data).all(), [Node(data='n2'), Node(data='n4'), Node(data='n5'), Node(data='n6'), Node(data='n7')])
+ eq_(sess.query(Node).filter(Node.children.contains(n4)).order_by(Node.data).all(),
+ [Node(data='n1'), Node(data='n3')])
+ eq_(sess.query(Node).filter(not_(Node.children.contains(n4))).order_by(Node.data).all(),
+ [Node(data='n2'), Node(data='n4'), Node(data='n5'),
+ Node(data='n6'), Node(data='n7')])
+ @testing.resolve_artifact_names
def test_explicit_join(self):
sess = create_session()
n1 = aliased(Node)
eq_(
- sess.query(Node).select_from(join(Node, n1, 'children')).filter(n1.data.in_(['n3', 'n7'])).order_by(Node.id).all(),
+ sess.query(Node).select_from(join(Node, n1, 'children'
+ )).filter(n1.data.in_(['n3', 'n7'
+ ])).order_by(Node.id).all(),
[Node(data='n1'), Node(data='n2')]
)