'orm.mapper',
'orm.generative',
'orm.lazytest1',
- 'orm.eagertest1',
- 'orm.eagertest2',
- 'orm.eagertest3',
+ 'orm.assorted_eager',
'orm.sessioncontext',
'orm.unitofwork',
-from testbase import PersistTest, AssertMixin
+"""eager loading unittests derived from mailing list-reported problems and trac tickets."""
+
+from testbase import PersistTest, AssertMixin, ORMTest
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
-from sqlalchemy.ext.selectresults import SelectResults
+from sqlalchemy.ext.sessioncontext import SessionContext
from testbase import Table, Column
-import random
+import random, datetime
class EagerTest(AssertMixin):
def setUpAll(self):
assert result == [u'1 Some Category', u'3 Some Category']
def test_dslish(self):
- """test the same as witheagerload except building the query via SelectResults"""
+ """test the same as witheagerload except using generative"""
s = create_session()
q=s.query(Test).options(eagerload('category'))
l=q.filter (
def tearDown(self):
for t in metadata.table_iterator(reverse=True):
t.delete().execute()
+
def testeagerterminate(self):
"""test that eager query generation does not include the same mapper's table twice.
# object is not in the session; therefore the lazy load cant trigger here,
# eager load had to succeed
assert len([c for c in d2.comments]) == 1
+
+class EagerTest6(ORMTest):
+ def define_tables(self, metadata):
+ global designType, design, part, inheritedPart
+ designType = Table('design_types', metadata,
+ Column('design_type_id', Integer, primary_key=True),
+ )
+
+ design =Table('design', metadata,
+ Column('design_id', Integer, primary_key=True),
+ Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
+
+ part = Table('parts', metadata,
+ Column('part_id', Integer, primary_key=True),
+ Column('design_id', Integer, ForeignKey('design.design_id')),
+ Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
+
+ inheritedPart = Table('inherited_part', metadata,
+ Column('ip_id', Integer, primary_key=True),
+ Column('part_id', Integer, ForeignKey('parts.part_id')),
+ Column('design_id', Integer, ForeignKey('design.design_id')),
+ )
+
+ def testone(self):
+ class Part(object):pass
+ class Design(object):pass
+ class DesignType(object):pass
+ class InheritedPart(object):pass
+
+ mapper(Part, part)
+
+ mapper(InheritedPart, inheritedPart, properties=dict(
+ part=relation(Part, lazy=False)
+ ))
+
+ mapper(Design, design, properties=dict(
+ parts=relation(Part, private=True, backref="design"),
+ inheritedParts=relation(InheritedPart, private=True, backref="design"),
+ ))
+
+ mapper(DesignType, designType, properties=dict(
+ # designs=relation(Design, private=True, backref="type"),
+ ))
+
+ class_mapper(Design).add_property("type", relation(DesignType, lazy=False, backref="designs"))
+ class_mapper(Part).add_property("design", relation(Design, lazy=False, backref="parts"))
+ #Part.mapper.add_property("designType", relation(DesignType))
+
+ d = Design()
+ sess = create_session()
+ sess.save(d)
+ sess.flush()
+ sess.clear()
+ x = sess.query(Design).get(1)
+ x.inheritedParts
+
+class EagerTest7(ORMTest):
+ def define_tables(self, metadata):
+ global companies_table, addresses_table, invoice_table, phones_table, items_table, ctx
+ global Company, Address, Phone, Item,Invoice
+
+ ctx = SessionContext(create_session)
+
+ companies_table = Table('companies', metadata,
+ Column('company_id', Integer, Sequence('company_id_seq', optional=True), primary_key = True),
+ Column('company_name', String(40)),
+
+ )
+
+ addresses_table = Table('addresses', metadata,
+ Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True),
+ Column('company_id', Integer, ForeignKey("companies.company_id")),
+ Column('address', String(40)),
+ )
+
+ phones_table = Table('phone_numbers', metadata,
+ Column('phone_id', Integer, Sequence('phone_id_seq', optional=True), primary_key = True),
+ Column('address_id', Integer, ForeignKey('addresses.address_id')),
+ Column('type', String(20)),
+ Column('number', String(10)),
+ )
+
+ invoice_table = Table('invoices', metadata,
+ Column('invoice_id', Integer, Sequence('invoice_id_seq', optional=True), primary_key = True),
+ Column('company_id', Integer, ForeignKey("companies.company_id")),
+ Column('date', DateTime),
+ )
+
+ items_table = Table('items', metadata,
+ Column('item_id', Integer, Sequence('item_id_seq', optional=True), primary_key = True),
+ Column('invoice_id', Integer, ForeignKey('invoices.invoice_id')),
+ Column('code', String(20)),
+ Column('qty', Integer),
+ )
+
+ class Company(object):
+ def __init__(self):
+ self.company_id = None
+ def __repr__(self):
+ return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses])
+
+ class Address(object):
+ def __repr__(self):
+ return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address) + str([repr(ph) for ph in getattr(self, 'phones', [])])
+
+ class Phone(object):
+ def __repr__(self):
+ return "Phone: " + repr(getattr(self, 'phone_id', None)) + " " + repr(getattr(self, 'address_id', None)) + " " + repr(self.type) + " " + repr(self.number)
+
+ class Invoice(object):
+ def __init__(self):
+ self.invoice_id = None
+ def __repr__(self):
+ return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company) + " " + str([repr(item) for item in self.items])
+
+ class Item(object):
+ def __repr__(self):
+ return "Item: " + repr(getattr(self, 'item_id', None)) + " " + repr(getattr(self, 'invoice_id', None)) + " " + repr(self.code) + " " + repr(self.qty)
+
+ def testone(self):
+ """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated
+ the bug, which is that when the single Company is loaded, no further processing of the rows
+ occurred in order to load the Company's second Address object."""
+
+ mapper(Address, addresses_table, properties={
+ }, extension=ctx.mapper_extension)
+ mapper(Company, companies_table, properties={
+ 'addresses' : relation(Address, lazy=False),
+ }, extension=ctx.mapper_extension)
+ mapper(Invoice, invoice_table, properties={
+ 'company': relation(Company, lazy=False, )
+ }, extension=ctx.mapper_extension)
+
+ c1 = Company()
+ c1.company_name = 'company 1'
+ a1 = Address()
+ a1.address = 'a1 address'
+ c1.addresses.append(a1)
+ a2 = Address()
+ a2.address = 'a2 address'
+ c1.addresses.append(a2)
+ i1 = Invoice()
+ i1.date = datetime.datetime.now()
+ i1.company = c1
+
+ ctx.current.flush()
+
+ company_id = c1.company_id
+ invoice_id = i1.invoice_id
+
+ ctx.current.clear()
+
+ c = ctx.current.query(Company).get(company_id)
+
+ ctx.current.clear()
+
+ i = ctx.current.query(Invoice).get(invoice_id)
+
+ self.echo(repr(c))
+ self.echo(repr(i.company))
+ self.assert_(repr(c) == repr(i.company))
+
+ def testtwo(self):
+ """this is the original testcase that includes various complicating factors"""
+
+ mapper(Phone, phones_table, extension=ctx.mapper_extension)
+
+ mapper(Address, addresses_table, properties={
+ 'phones': relation(Phone, lazy=False, backref='address')
+ }, extension=ctx.mapper_extension)
+
+ mapper(Company, companies_table, properties={
+ 'addresses' : relation(Address, lazy=False, backref='company'),
+ }, extension=ctx.mapper_extension)
+
+ mapper(Item, items_table, extension=ctx.mapper_extension)
+
+ mapper(Invoice, invoice_table, properties={
+ 'items': relation(Item, lazy=False, backref='invoice'),
+ 'company': relation(Company, lazy=False, backref='invoices')
+ }, extension=ctx.mapper_extension)
+
+ ctx.current.clear()
+ c1 = Company()
+ c1.company_name = 'company 1'
+
+ a1 = Address()
+ a1.address = 'a1 address'
+
+ p1 = Phone()
+ p1.type = 'home'
+ p1.number = '1111'
+
+ a1.phones.append(p1)
+
+ p2 = Phone()
+ p2.type = 'work'
+ p2.number = '22222'
+ a1.phones.append(p2)
+
+ c1.addresses.append(a1)
+
+ a2 = Address()
+ a2.address = 'a2 address'
+
+ p3 = Phone()
+ p3.type = 'home'
+ p3.number = '3333'
+ a2.phones.append(p3)
+
+ p4 = Phone()
+ p4.type = 'work'
+ p4.number = '44444'
+ a2.phones.append(p4)
+
+ c1.addresses.append(a2)
+
+ ctx.current.flush()
+
+ company_id = c1.company_id
+
+ ctx.current.clear()
+
+ a = ctx.current.query(Company).get(company_id)
+ self.echo(repr(a))
+
+ # set up an invoice
+ i1 = Invoice()
+ i1.date = datetime.datetime.now()
+ i1.company = c1
+
+ item1 = Item()
+ item1.code = 'aaaa'
+ item1.qty = 1
+ item1.invoice = i1
+
+ item2 = Item()
+ item2.code = 'bbbb'
+ item2.qty = 2
+ item2.invoice = i1
+
+ item3 = Item()
+ item3.code = 'cccc'
+ item3.qty = 3
+ item3.invoice = i1
+
+ ctx.current.flush()
+
+ invoice_id = i1.invoice_id
+
+ ctx.current.clear()
+
+ c = ctx.current.query(Company).get(company_id)
+ self.echo(repr(c))
+
+ ctx.current.clear()
+
+ i = ctx.current.query(Invoice).get(invoice_id)
+
+ assert repr(i.company) == repr(c), repr(i.company) + " does not match " + repr(c)
if __name__ == "__main__":
q = sess.query(User)
assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(users.c.id == 7).all()
- assert self.user_address_result == q.all()
+ assert fixtures.user_address_result == q.all()
def test_no_orphan(self):
"""test that an eagerly loaded child object is not marked as an orphan"""
q = create_session().query(Item)
def go():
- assert self.item_keyword_result == q.all()
+ assert fixtures.item_keyword_result == q.all()
self.assert_sql_count(testbase.db, go, 1)
def go():
- assert self.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
+ assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
self.assert_sql_count(testbase.db, go, 1)
q = create_session().query(Item)
def go():
- assert self.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all()
+ assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all()
self.assert_sql_count(testbase.db, go, 1)
assert class_mapper(Address).props['user'].lazy is False
sess = create_session()
- assert self.user_address_result == sess.query(User).all()
+ assert fixtures.user_address_result == sess.query(User).all()
def test_double(self):
"""tests lazy loading with two relations simulatneously, from the same table, using aliases. """
if testbase.db.engine.name == 'mssql':
l = q.limit(2).all()
- assert self.user_all_result[:2] == l
+ assert fixtures.user_all_result[:2] == l
else:
l = q.limit(2).offset(1).all()
- print l
- print self.user_all_result[1:3]
- assert self.user_all_result[1:3] == l
+ assert fixtures.user_all_result[1:3] == l
def test_distinct(self):
# this is an involved 3x union of the users table to get a lot of rows.
def go():
l = q.filter(s.c.u2_id==User.c.id).distinct().all()
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def test_limit_2(self):
l = q.filter((Item.c.description=='item 2') | (Item.c.description=='item 5') | (Item.c.description=='item 3')).\
order_by(Item.c.id).limit(2).all()
- assert self.item_keyword_result[1:3] == l
+ assert fixtures.item_keyword_result[1:3] == l
def test_limit_3(self):
"""test that the ORDER BY is propigated from the inner select to the outer select, when using the
l = q.filter("users.id in (7, 8, 9)")
def go():
- assert self.user_order_result[0:3] == l.all()
+ assert fixtures.user_order_result[0:3] == l.all()
self.assert_sql_count(testbase.db, go, 1)
def test_double_with_aggregate(self):
] == q.all()
self.assert_sql_count(testbase.db, go, 1)
+ def test_wide(self):
+ mapper(Order, orders, properties={'items':relation(Item, secondary=order_items, lazy=False)})
+ mapper(Item, items)
+ mapper(User, users, properties = dict(
+ addresses = relation(mapper(Address, addresses), lazy = False),
+ orders = relation(Order, lazy = False),
+ ))
+ q = create_session().query(User)
+ l = q.select()
+ assert fixtures.user_all_result == q.all()
+
+ def test_against_select(self):
+ """test eager loading of a mapper which is against a select"""
+
+ s = select([orders], orders.c.isopen==1).alias('openorders')
+
+ mapper(Order, s, properties={
+ 'user':relation(User, lazy=False)
+ })
+ mapper(User, users)
+
+ q = create_session().query(Order)
+ assert [
+ Order(id=3, user=User(id=7)),
+ Order(id=4, user=User(id=9))
+ ] == q.all()
+
+ q = q.select_from(s.join(order_items).join(items)).filter(~items.c.id.in_(1, 2, 5))
+ assert [
+ Order(id=3, user=User(id=7)),
+ ] == q.all()
+
+ def test_aliasing(self):
+ """test that eager loading uses aliases to insulate the eager load from regular criterion against those tables."""
+
+ mapper(User, users, properties = dict(
+ addresses = relation(mapper(Address, addresses), lazy=False)
+ ))
+ q = create_session().query(User)
+ l = q.filter(addresses.c.email_address == 'ed@lala.com').filter(addresses.c.user_id==users.c.id)
+ assert fixtures.user_address_result[1:2] == l.all()
+
if __name__ == '__main__':
testbase.main()
+++ /dev/null
-from testbase import PersistTest, AssertMixin
-import testbase
-import unittest, sys, os
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from testbase import Table, Column
-import datetime
-
-class EagerTest(AssertMixin):
- def setUpAll(self):
- global designType, design, part, inheritedPart
- designType = Table('design_types', testbase.metadata,
- Column('design_type_id', Integer, primary_key=True),
- )
-
- design =Table('design', testbase.metadata,
- Column('design_id', Integer, primary_key=True),
- Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
-
- part = Table('parts', testbase.metadata,
- Column('part_id', Integer, primary_key=True),
- Column('design_id', Integer, ForeignKey('design.design_id')),
- Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
-
- inheritedPart = Table('inherited_part', testbase.metadata,
- Column('ip_id', Integer, primary_key=True),
- Column('part_id', Integer, ForeignKey('parts.part_id')),
- Column('design_id', Integer, ForeignKey('design.design_id')),
- )
-
- testbase.metadata.create_all()
- def tearDownAll(self):
- testbase.metadata.drop_all()
- testbase.metadata.clear()
- def testone(self):
- class Part(object):pass
- class Design(object):pass
- class DesignType(object):pass
- class InheritedPart(object):pass
-
- mapper(Part, part)
-
- mapper(InheritedPart, inheritedPart, properties=dict(
- part=relation(Part, lazy=False)
- ))
-
- mapper(Design, design, properties=dict(
- parts=relation(Part, private=True, backref="design"),
- inheritedParts=relation(InheritedPart, private=True, backref="design"),
- ))
-
- mapper(DesignType, designType, properties=dict(
- # designs=relation(Design, private=True, backref="type"),
- ))
-
- class_mapper(Design).add_property("type", relation(DesignType, lazy=False, backref="designs"))
- class_mapper(Part).add_property("design", relation(Design, lazy=False, backref="parts"))
- #Part.mapper.add_property("designType", relation(DesignType))
-
- d = Design()
- sess = create_session()
- sess.save(d)
- sess.flush()
- sess.clear()
- x = sess.query(Design).get(1)
- x.inheritedParts
-
-if __name__ == "__main__":
- testbase.main()
-
-
+++ /dev/null
-from testbase import PersistTest, AssertMixin
-import testbase
-import unittest, sys, os
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from sqlalchemy.ext.sessioncontext import SessionContext
-from testbase import Table, Column
-import datetime
-
-class EagerTest(AssertMixin):
- def setUpAll(self):
- global companies_table, addresses_table, invoice_table, phones_table, items_table, ctx, metadata
-
- metadata = BoundMetaData(testbase.db)
- ctx = SessionContext(create_session)
-
- companies_table = Table('companies', metadata,
- Column('company_id', Integer, Sequence('company_id_seq', optional=True), primary_key = True),
- Column('company_name', String(40)),
-
- )
-
- addresses_table = Table('addresses', metadata,
- Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True),
- Column('company_id', Integer, ForeignKey("companies.company_id")),
- Column('address', String(40)),
- )
-
- phones_table = Table('phone_numbers', metadata,
- Column('phone_id', Integer, Sequence('phone_id_seq', optional=True), primary_key = True),
- Column('address_id', Integer, ForeignKey('addresses.address_id')),
- Column('type', String(20)),
- Column('number', String(10)),
- )
-
- invoice_table = Table('invoices', metadata,
- Column('invoice_id', Integer, Sequence('invoice_id_seq', optional=True), primary_key = True),
- Column('company_id', Integer, ForeignKey("companies.company_id")),
- Column('date', DateTime),
- )
-
- items_table = Table('items', metadata,
- Column('item_id', Integer, Sequence('item_id_seq', optional=True), primary_key = True),
- Column('invoice_id', Integer, ForeignKey('invoices.invoice_id')),
- Column('code', String(20)),
- Column('qty', Integer),
- )
-
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
- def testone(self):
- """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated
- the bug, which is that when the single Company is loaded, no further processing of the rows
- occurred in order to load the Company's second Address object."""
- class Company(object):
- def __init__(self):
- self.company_id = None
- def __repr__(self):
- return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses])
-
- class Address(object):
- def __repr__(self):
- return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address)
-
- class Invoice(object):
- def __init__(self):
- self.invoice_id = None
- def __repr__(self):
- return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company)
-
- mapper(Address, addresses_table, properties={
- }, extension=ctx.mapper_extension)
- mapper(Company, companies_table, properties={
- 'addresses' : relation(Address, lazy=False),
- }, extension=ctx.mapper_extension)
- mapper(Invoice, invoice_table, properties={
- 'company': relation(Company, lazy=False, )
- }, extension=ctx.mapper_extension)
-
- c1 = Company()
- c1.company_name = 'company 1'
- a1 = Address()
- a1.address = 'a1 address'
- c1.addresses.append(a1)
- a2 = Address()
- a2.address = 'a2 address'
- c1.addresses.append(a2)
- i1 = Invoice()
- i1.date = datetime.datetime.now()
- i1.company = c1
-
- ctx.current.flush()
-
- company_id = c1.company_id
- invoice_id = i1.invoice_id
-
- ctx.current.clear()
-
- c = ctx.current.query(Company).get(company_id)
-
- ctx.current.clear()
-
- i = ctx.current.query(Invoice).get(invoice_id)
-
- self.echo(repr(c))
- self.echo(repr(i.company))
- self.assert_(repr(c) == repr(i.company))
-
- def testtwo(self):
- """this is the original testcase that includes various complicating factors"""
- class Company(object):
- def __init__(self):
- self.company_id = None
- def __repr__(self):
- return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses])
-
- class Address(object):
- def __repr__(self):
- return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address) + str([repr(ph) for ph in self.phones])
-
- class Phone(object):
- def __repr__(self):
- return "Phone: " + repr(getattr(self, 'phone_id', None)) + " " + repr(getattr(self, 'address_id', None)) + " " + repr(self.type) + " " + repr(self.number)
-
- class Invoice(object):
- def __init__(self):
- self.invoice_id = None
- def __repr__(self):
- return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company) + " " + str([repr(item) for item in self.items])
-
- class Item(object):
- def __repr__(self):
- return "Item: " + repr(getattr(self, 'item_id', None)) + " " + repr(getattr(self, 'invoice_id', None)) + " " + repr(self.code) + " " + repr(self.qty)
-
- mapper(Phone, phones_table, extension=ctx.mapper_extension)
-
- mapper(Address, addresses_table, properties={
- 'phones': relation(Phone, lazy=False, backref='address')
- }, extension=ctx.mapper_extension)
-
- mapper(Company, companies_table, properties={
- 'addresses' : relation(Address, lazy=False, backref='company'),
- }, extension=ctx.mapper_extension)
-
- mapper(Item, items_table, extension=ctx.mapper_extension)
-
- mapper(Invoice, invoice_table, properties={
- 'items': relation(Item, lazy=False, backref='invoice'),
- 'company': relation(Company, lazy=False, backref='invoices')
- }, extension=ctx.mapper_extension)
-
- ctx.current.clear()
- c1 = Company()
- c1.company_name = 'company 1'
-
- a1 = Address()
- a1.address = 'a1 address'
-
- p1 = Phone()
- p1.type = 'home'
- p1.number = '1111'
-
- a1.phones.append(p1)
-
- p2 = Phone()
- p2.type = 'work'
- p2.number = '22222'
- a1.phones.append(p2)
-
- c1.addresses.append(a1)
-
- a2 = Address()
- a2.address = 'a2 address'
-
- p3 = Phone()
- p3.type = 'home'
- p3.number = '3333'
- a2.phones.append(p3)
-
- p4 = Phone()
- p4.type = 'work'
- p4.number = '44444'
- a2.phones.append(p4)
-
- c1.addresses.append(a2)
-
- ctx.current.flush()
-
- company_id = c1.company_id
-
- ctx.current.clear()
-
- a = ctx.current.query(Company).get(company_id)
- self.echo(repr(a))
-
- # set up an invoice
- i1 = Invoice()
- i1.date = datetime.datetime.now()
- i1.company = c1
-
- item1 = Item()
- item1.code = 'aaaa'
- item1.qty = 1
- item1.invoice = i1
-
- item2 = Item()
- item2.code = 'bbbb'
- item2.qty = 2
- item2.invoice = i1
-
- item3 = Item()
- item3.code = 'cccc'
- item3.qty = 3
- item3.invoice = i1
-
- ctx.current.flush()
-
- invoice_id = i1.invoice_id
-
- ctx.current.clear()
-
- c = ctx.current.query(Company).get(company_id)
- self.echo(repr(c))
-
- ctx.current.clear()
-
- i = ctx.current.query(Invoice).get(invoice_id)
-
- assert repr(i.company) == repr(c), repr(i.company) + " does not match " + repr(c)
-
-if __name__ == "__main__":
- testbase.main()
dict(keyword_id=7, item_id=2),
dict(keyword_id=6, item_id=3)
)
+
+class Fixtures(object):
+ @property
+ def user_address_result(self):
+ return [
+ User(id=7, addresses=[
+ Address(id=1)
+ ]),
+ User(id=8, addresses=[
+ Address(id=2, email_address='ed@wood.com'),
+ Address(id=3, email_address='ed@bettyboop.com'),
+ Address(id=4, email_address='ed@lala.com'),
+ ]),
+ User(id=9, addresses=[
+ Address(id=5)
+ ]),
+ User(id=10, addresses=[])
+ ]
+
+ @property
+ def user_all_result(self):
+ return [
+ User(id=7, addresses=[
+ Address(id=1)
+ ], orders=[
+ Order(description='order 1', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
+ Order(description='order 3'),
+ Order(description='order 5'),
+ ]),
+ User(id=8, addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4)
+ ]),
+ User(id=9, addresses=[
+ Address(id=5)
+ ], orders=[
+ Order(description='order 2', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
+ Order(description='order 4', items=[Item(description='item 1'), Item(description='item 5')]),
+ ]),
+ User(id=10, addresses=[])
+ ]
+
+ @property
+ def user_order_result(self):
+ return [
+ User(id=7, orders=[
+ Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]),
+ Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]),
+ Order(id=5, items=[Item(id=5)]),
+ ]),
+ User(id=8, orders=[]),
+ User(id=9, orders=[
+ Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]),
+ Order(id=4, items=[Item(id=1), Item(id=5)]),
+ ]),
+ User(id=10)
+ ]
+
+ @property
+ def item_keyword_result(self):
+ return [
+ Item(id=1, keywords=[Keyword(name='red'), Keyword(name='big'), Keyword(name='round')]),
+ Item(id=2, keywords=[Keyword(name='red'), Keyword(name='small'), Keyword(name='square')]),
+ Item(id=3, keywords=[Keyword(name='green'), Keyword(name='big'), Keyword(name='round')]),
+ Item(id=4, keywords=[]),
+ Item(id=5, keywords=[]),
+ ]
+fixtures = Fixtures()
if testbase.db.engine.name == 'mssql':
l = q.limit(2).all()
- assert self.user_all_result[:2] == l
+ assert fixtures.user_all_result[:2] == l
else:
l = q.limit(2).offset(1).all()
- print l
- print self.user_all_result[1:3]
- assert self.user_all_result[1:3] == l
+ assert fixtures.user_all_result[1:3] == l
def test_distinct(self):
mapper(Item, items)
s = union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u')
print [key for key in s.c.keys()]
l = q.filter(s.c.u2_id==User.c.id).distinct().all()
- assert self.user_all_result == l
+ assert fixtures.user_all_result == l
def test_one_to_many_scalar(self):
mapper(User, users, properties = dict(
))
q = create_session().query(Item)
- assert self.item_keyword_result == q.all()
+ assert fixtures.item_keyword_result == q.all()
- assert self.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
+ assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
def test_uses_get(self):
"""test that a simple many-to-one lazyload optimizes to use query.get()."""
+"""tests general mapper operations with an emphasis on selecting/loading"""
+
from testbase import PersistTest, AssertMixin
import testbase
import unittest, sys, os
from tables import *
import tables
-"""tests general mapper operations with an emphasis on selecting/loading"""
-
class MapperSuperTest(AssertMixin):
def setUpAll(self):
tables.create()
{'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])},
)
-
class MapperExtensionTest(MapperSuperTest):
def testcreateinstance(self):
class Ext(MapperExtension):
l = q.select();
self.assert_result(l, User, *user_address_result)
-class EagerTest(MapperSuperTest):
-
- def testwithrepeat(self):
- """tests a one-to-many eager load where we also query on joined criterion, where the joined
- criterion is using the same tables that are used within the eager load. the mapper must insure that the
- criterion doesnt interfere with the eager load criterion."""
- m = mapper(User, users, properties = dict(
- addresses = relation(mapper(Address, addresses), primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False)
- ))
- q = create_session().query(m)
- l = q.select(and_(addresses.c.email_address == 'ed@lala.com', addresses.c.user_id==users.c.user_id))
- self.assert_result(l, User,
- {'user_id' : 8, 'addresses' : (Address, [{'address_id' : 2, 'email_address':'ed@wood.com'}, {'address_id':3, 'email_address':'ed@bettyboop.com'}, {'address_id':4, 'email_address':'ed@lala.com'}])},
- )
-
-
- def testonselect(self):
- """test eager loading of a mapper which is against a select"""
-
- s = select([orders], orders.c.isopen==1).alias('openorders')
- print "SELECT:", id(s), str(s)
- mapper(Order, s, properties={
- 'user':relation(User, lazy=False)
- })
- mapper(User, users)
-
- q = create_session().query(Order)
- self.assert_result(q.list(), Order,
- {'order_id':3, 'user' : (User, {'user_id':7})},
- {'order_id':4, 'user' : (User, {'user_id':9})},
- )
-
- q = q.select_from(s.outerjoin(orderitems)).filter(orderitems.c.item_name != 'item 2')
- self.assert_result(q.list(), Order,
- {'order_id':3, 'user' : (User, {'user_id':7})},
- )
-
-
- def testmulti(self):
- """tests eager loading with two relations simultaneously"""
- m = mapper(User, users, properties = dict(
- addresses = relation(mapper(Address, addresses), primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False),
- orders = relation(mapper(Order, orders), lazy = False),
- ))
- q = create_session().query(m)
- l = q.select()
- self.assert_result(l, User,
- {'user_id' : 7,
- 'addresses' : (Address, [{'address_id' : 1}]),
- 'orders' : (Order, [{'order_id' : 1}, {'order_id' : 3},{'order_id' : 5},])
- },
- {'user_id' : 8,
- 'addresses' : (Address, [{'address_id' : 2}, {'address_id' : 3}, {'address_id' : 4}]),
- 'orders' : (Order, [])
- },
- {'user_id' : 9,
- 'addresses' : (Address, []),
- 'orders' : (Order, [{'order_id' : 2},{'order_id' : 4}])
- }
- )
-
- def testnested(self):
- """tests eager loading of a parent item with two types of child items,
- where one of those child items eager loads its own child items."""
- ordermapper = mapper(Order, orders, properties = dict(
- items = relation(mapper(Item, orderitems), lazy = False)
- ))
-
- m = mapper(User, users, properties = dict(
- addresses = relation(mapper(Address, addresses), lazy = False),
- orders = relation(ordermapper, primaryjoin = users.c.user_id==orders.c.user_id, lazy = False),
- ))
- q = create_session().query(m)
- l = q.select()
- self.assert_result(l, User, *user_all_result)
-
if __name__ == "__main__":
})
mapper(Keyword, keywords)
- @property
- def user_address_result(self):
- return [
- User(id=7, addresses=[
- Address(id=1)
- ]),
- User(id=8, addresses=[
- Address(id=2, email_address='ed@wood.com'),
- Address(id=3, email_address='ed@bettyboop.com'),
- Address(id=4, email_address='ed@lala.com'),
- ]),
- User(id=9, addresses=[
- Address(id=5)
- ]),
- User(id=10, addresses=[])
- ]
-
- @property
- def user_all_result(self):
- return [
- User(id=7, addresses=[
- Address(id=1)
- ], orders=[
- Order(description='order 1', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
- Order(description='order 3'),
- Order(description='order 5'),
- ]),
- User(id=8, addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4)
- ]),
- User(id=9, addresses=[
- Address(id=5)
- ], orders=[
- Order(description='order 2', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
- Order(description='order 4', items=[Item(description='item 1'), Item(description='item 5')]),
- ]),
- User(id=10, addresses=[])
- ]
-
- @property
- def user_order_result(self):
- return [
- User(id=7, orders=[
- Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]),
- Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]),
- Order(id=5, items=[Item(id=5)]),
- ]),
- User(id=8, orders=[]),
- User(id=9, orders=[
- Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]),
- Order(id=4, items=[Item(id=1), Item(id=5)]),
- ]),
- User(id=10)
- ]
-
- @property
- def item_keyword_result(self):
- return [
- Item(id=1, keywords=[Keyword(name='red'), Keyword(name='big'), Keyword(name='round')]),
- Item(id=2, keywords=[Keyword(name='red'), Keyword(name='small'), Keyword(name='square')]),
- Item(id=3, keywords=[Keyword(name='green'), Keyword(name='big'), Keyword(name='round')]),
- Item(id=4, keywords=[]),
- Item(id=5, keywords=[]),
- ]
class GetTest(QueryTest):
def test_get(self):
def go():
l = q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute())
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def test_contains_eager(self):
def go():
l = q.options(contains_eager('addresses')).instances(selectquery.execute())
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def test_contains_eager_alias(self):
def go():
# test using a string alias name
l = q.options(contains_eager('addresses', alias="adalias")).instances(selectquery.execute())
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def go():
# test using the Alias object itself
l = q.options(contains_eager('addresses', alias=adalias)).instances(selectquery.execute())
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def decorate(row):
def go():
# test using a custom 'decorate' function
l = q.options(contains_eager('addresses', decorator=decorate)).instances(selectquery.execute())
- assert self.user_address_result == l
+ assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
def test_multi_mappers(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
