polymorph.py

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

        assert c.managers[0].company is c

class RoundTripTest(PolymorphTest):
    pass

def generate_round_trip_test(include_base=False, lazy_relation=True, redefine_colprop=False, use_literal_join=False, polymorphic_fetch=None, use_outer_joins=False):
    """generates a round trip test.

    include_base - whether or not to include the base 'person' type in the union.
    lazy_relation - whether or not the Company relation to People is lazy or eager.
    redefine_colprop - if we redefine the 'name' column to be 'people_name' on the base Person class
    use_literal_join - primary join condition is explicitly specified
    """
    def test_roundtrip(self):
        # create a union that represents both types of joins.
        if not polymorphic_fetch == 'union':
            person_join = None
            manager_join = None
        elif include_base:
            if use_outer_joins:
                person_join = people.outerjoin(engineers).outerjoin(managers).outerjoin(boss)
                manager_join = people.join(managers).outerjoin(boss)
            else:
                person_join = polymorphic_union(
                    {
                        'engineer':people.join(engineers),
                        'manager':people.join(managers),
                        'person':people.select(people.c.type=='person'),
                    }, None, 'pjoin')

                manager_join = people.join(managers).outerjoin(boss)
        else:
            if use_outer_joins:
                person_join = people.outerjoin(engineers).outerjoin(managers).outerjoin(boss)
                manager_join = people.join(managers).outerjoin(boss)
            else:
                person_join = polymorphic_union(
                    {
                        'engineer':people.join(engineers),
                        'manager':people.join(managers),
                    }, None, 'pjoin')
                manager_join = people.join(managers).outerjoin(boss)

        if redefine_colprop:
            person_mapper = mapper(Person, people, select_table=person_join, polymorphic_fetch=polymorphic_fetch, polymorphic_on=people.c.type, polymorphic_identity='person', properties= {'person_name':people.c.name})
        else:
            person_mapper = mapper(Person, people, select_table=person_join, polymorphic_fetch=polymorphic_fetch, polymorphic_on=people.c.type, polymorphic_identity='person')

        mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
        mapper(Manager, managers, inherits=person_mapper, select_table=manager_join, polymorphic_identity='manager')

        mapper(Boss, boss, inherits=Manager, polymorphic_identity='boss')

        if use_literal_join:
            mapper(Company, companies, properties={
                'employees': relation(Person, lazy=lazy_relation,
                                      primaryjoin=(people.c.company_id ==
                                                   companies.c.company_id),
                                      cascade="all,delete-orphan",
                                      backref="company"
                )
            })
        else:
            mapper(Company, companies, properties={
                'employees': relation(Person, lazy=lazy_relation,
                                      cascade="all, delete-orphan",
                backref="company"
                )
            })

        if redefine_colprop:
            person_attribute_name = 'person_name'
        else:
            person_attribute_name = 'name'

        session = create_session()
        c = Company(name='company1')
        c.employees.append(Manager(status='AAB', manager_name='manager1', **{person_attribute_name:'pointy haired boss'}))
        c.employees.append(Engineer(status='BBA', engineer_name='engineer1', primary_language='java', **{person_attribute_name:'dilbert'}))
        dilbert = c.employees[-1]

        if include_base:
            c.employees.append(Person(status='HHH', **{person_attribute_name:'joesmith'}))
        c.employees.append(Engineer(status='CGG', engineer_name='engineer2', primary_language='python', **{person_attribute_name:'wally'}))
        c.employees.append(Manager(status='ABA', manager_name='manager2', **{person_attribute_name:'jsmith'}))
        session.save(c)
        print session.new
        session.flush()
        session.clear()

        dilbert = session.query(Person).get(dilbert.person_id)
        assert getattr(dilbert, person_attribute_name) == 'dilbert'
        session.clear()

        dilbert = session.query(Person).filter(Person.person_id==dilbert.person_id).one()
        assert getattr(dilbert, person_attribute_name) == 'dilbert'
        session.clear()

        id = c.company_id
        def go():
            c = session.query(Company).get(id)
            for e in c.employees:
                print e, e._instance_key, e.company
                assert e._instance_key[0] == Person
            if include_base:
                assert sets.Set([(e.get_name(), getattr(e, 'status', None)) for e in c.employees]) == sets.Set([('pointy haired boss', 'AAB'), ('dilbert', 'BBA'), ('joesmith', None), ('wally', 'CGG'), ('jsmith', 'ABA')])
            else:
                assert sets.Set([(e.get_name(), e.status) for e in c.employees]) == sets.Set([('pointy haired boss', 'AAB'), ('dilbert', 'BBA'), ('wally', 'CGG'), ('jsmith', 'ABA')])
            print "\n"

        if not lazy_relation:
            if polymorphic_fetch=='union':
                self.assert_sql_count(testing.db, go, 1)
            else:
                self.assert_sql_count(testing.db, go, 5)

        else:
            if polymorphic_fetch=='union':
                self.assert_sql_count(testing.db, go, 2)
            else:
                self.assert_sql_count(testing.db, go, 6)

        # test selecting from the query, using the base mapped table (people) as the selection criterion.
        # in the case of the polymorphic Person query, the "people" selectable should be adapted to be "person_join"
        dilbert = session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
        dilbert2 = session.query(Engineer).filter(getattr(Person, person_attribute_name)=='dilbert').first()
        assert dilbert is dilbert2

        # test selecting from the query, joining against an alias of the base "people" table.  test that
        # the "palias" alias does *not* get sucked up into the "person_join" conversion.
        palias = people.alias("palias")
        session.query(Person).filter((palias.c.name=='dilbert') & (palias.c.person_id==Person.person_id)).first()
        dilbert2 = session.query(Engineer).filter((palias.c.name=='dilbert') & (palias.c.person_id==Person.person_id)).first()
        assert dilbert is dilbert2

        session.query(Person).filter((Engineer.engineer_name=="engineer1") & (Engineer.person_id==people.c.person_id)).first()

        dilbert2 = session.query(Engineer).filter(Engineer.engineer_name=="engineer1")[0]
        assert dilbert is dilbert2

        dilbert.engineer_name = 'hes dibert!'

        session.flush()
        session.clear()

        # save/load some managers/bosses
        b = Boss(status='BBB', manager_name='boss', golf_swing='fore', **{person_attribute_name:'daboss'})
        session.save(b)
        session.flush()
        session.clear()
        c = session.query(Manager).all()
        assert sets.Set([repr(x) for x in c]) == sets.Set(["Manager pointy haired boss, status AAB, manager_name manager1", "Manager jsmith, status ABA, manager_name manager2", "Boss daboss, status BBB, manager_name boss golf swing fore"]), repr([repr(x) for x in c])

        c = session.query(Company).get(id)
        for e in c.employees:
            print e, e._instance_key

        session.delete(c)
        session.flush()

    test_roundtrip = _function_named(
        test_roundtrip, "test_%s%s%s%s%s" % (
          (lazy_relation and "lazy" or "eager"),
          (include_base and "_inclbase" or ""),
          (redefine_colprop and "_redefcol" or ""),
          (polymorphic_fetch != 'union' and '_' + polymorphic_fetch or (use_literal_join and "_litjoin" or "")),
          (use_outer_joins and '_outerjoins' or '')))
    setattr(RoundTripTest, test_roundtrip.__name__, test_roundtrip)

for include_base in [True, False]:
    for lazy_relation in [True, False]:
        for redefine_colprop in [True, False]:
            for use_literal_join in [True, False]:
                for polymorphic_fetch in ['union', 'select', 'deferred']:
                    if polymorphic_fetch == 'union':
                        for use_outer_joins in [True, False]:
                            generate_round_trip_test(include_base, lazy_relation, redefine_colprop, use_literal_join, polymorphic_fetch, use_outer_joins)
                    else:
                        generate_round_trip_test(include_base, lazy_relation, redefine_colprop, use_literal_join, polymorphic_fetch, False)

if __name__ == "__main__":
    testenv.main()