assorted_eager.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

            stat=Stat()
            stat.data = data[rid]
            stat.somedata=somedata
            s.save(stat)

        s.flush()

        arb_data=select(
            [stats.c.data_id,func.max(stats.c.somedata).label('max')],
            stats.c.data_id<=25,
            group_by=[stats.c.data_id]).alias('arb')

        arb_result = arb_data.execute().fetchall()
        # order the result list descending based on 'max'
        arb_result.sort(lambda a, b:cmp(b['max'],a['max']))
        # extract just the "data_id" from it
        arb_result = [row['data_id'] for row in arb_result]

        # now query for Data objects using that above select, adding the
        # "order by max desc" separately
        q=(s.query(Data).options(eagerload('foo')).
           select_from(datas.join(arb_data,arb_data.c.data_id==datas.c.id)).
           order_by(desc(arb_data.c.max)).
           limit(10))

        # extract "data_id" from the list of result objects
        verify_result = [d.id for d in q]

        # assert equality including ordering (may break if the DB "ORDER BY" and python's sort() used differing
        # algorithms and there are repeated 'somedata' values in the list)
        assert verify_result == arb_result

class EagerTest4(ORMTest):
    def define_tables(self, metadata):
        global departments, employees
        departments = Table('departments', metadata,
                            Column('department_id', Integer, primary_key=True),
                            Column('name', String(50)))

        employees = Table('employees', metadata,
                          Column('person_id', Integer, primary_key=True),
                          Column('name', String(50)),
                          Column('department_id', Integer,
                                 ForeignKey('departments.department_id')))

    @testing.fails_on('maxdb')
    def test_basic(self):
        class Department(object):
            def __init__(self, **kwargs):
                for k, v in kwargs.iteritems():
                    setattr(self, k, v)
            def __repr__(self):
                return "<Department %s>" % (self.name,)

        class Employee(object):
            def __init__(self, **kwargs):
                for k, v in kwargs.iteritems():
                    setattr(self, k, v)
            def __repr__(self):
                return "<Employee %s>" % (self.name,)

        mapper(Employee, employees)
        mapper(Department, departments,
                      properties=dict(employees=relation(Employee,
                                                         lazy=False,
                                                         backref='department')))

        d1 = Department(name='One')
        for e in 'Jim Jack John Susan'.split():
            d1.employees.append(Employee(name=e))

        d2 = Department(name='Two')
        for e in 'Joe Bob Mary Wally'.split():
            d2.employees.append(Employee(name=e))

        sess = create_session()
        sess.save(d1)
        sess.save(d2)
        sess.flush()

        q = sess.query(Department)
        q = q.join('employees').filter(Employee.c.name.startswith('J')).distinct().order_by([desc(Department.c.name)])
        assert q.count() == 2
        assert q[0] is d2

class EagerTest5(ORMTest):
    """test the construction of AliasedClauses for the same eager load property but different
    parent mappers, due to inheritance"""
    def define_tables(self, metadata):
        global base, derived, derivedII, comments
        base = Table(
            'base', metadata,
            Column('uid', String(30), primary_key=True),
            Column('x', String(30))
            )

        derived = Table(
            'derived', metadata,
            Column('uid', String(30), ForeignKey(base.c.uid), primary_key=True),
            Column('y', String(30))
            )

        derivedII = Table(
            'derivedII', metadata,
            Column('uid', String(30), ForeignKey(base.c.uid), primary_key=True),
            Column('z', String(30))
            )

        comments = Table(
            'comments', metadata,
            Column('id', Integer, primary_key=True),
            Column('uid', String(30), ForeignKey(base.c.uid)),
            Column('comment', String(30))
            )
    def test_basic(self):
        class Base(object):
            def __init__(self, uid, x):
                self.uid = uid
                self.x = x

        class Derived(Base):
            def __init__(self, uid, x, y):
                self.uid = uid
                self.x = x
                self.y = y

        class DerivedII(Base):
            def __init__(self, uid, x, z):
                self.uid = uid
                self.x = x
                self.z = z

        class Comment(object):
            def __init__(self, uid, comment):
                self.uid = uid
                self.comment = comment


        commentMapper = mapper(Comment, comments)

        baseMapper = mapper(
            Base, base,
                properties={
                'comments': relation(
                    Comment, lazy=False, cascade='all, delete-orphan'
                    )
                }
            )

        derivedMapper = mapper(Derived, derived, inherits=baseMapper)
        derivedIIMapper = mapper(DerivedII, derivedII, inherits=baseMapper)
        sess = create_session()
        d = Derived('uid1', 'x', 'y')
        d.comments = [Comment('uid1', 'comment')]
        d2 = DerivedII('uid2', 'xx', 'z')
        d2.comments = [Comment('uid2', 'comment')]
        sess.save(d)
        sess.save(d2)
        sess.flush()
        sess.clear()
        # this eager load sets up an AliasedClauses for the "comment" relationship,
        # then stores it in clauses_by_lead_mapper[mapper for Derived]
        d = sess.query(Derived).get('uid1')
        sess.clear()
        assert len([c for c in d.comments]) == 1

        # this eager load sets up an AliasedClauses for the "comment" relationship,
        # and should store it in clauses_by_lead_mapper[mapper for DerivedII].
        # the bug was that the previous AliasedClause create prevented this population
        # from occurring.
        d2 = sess.query(DerivedII).get('uid2')
        sess.clear()
        # 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(
            inheritedParts=relation(InheritedPart,
                                    cascade="all, delete-orphan",
                                    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=backref("parts", cascade="all, delete-orphan")))

        #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):
    @testing.uses_deprecated('SessionContext')
    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)

    @testing.uses_deprecated('SessionContext')
    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={