expr.py

Python的一个ORM,现在很火

        @compile_parent.when(C)
        def compile_in_parent(compile, state, expr):
            return "parent"
        statement, parameters = compile_child(C())
        self.assertEquals(statement, "parent")

        @compile_child.when(C)
        def compile_in_child(compile, state, expr):
            return "child"
        statement, parameters = compile_child(C())
        self.assertEquals(statement, "child")

    def test_precedence(self):
        for i in range(10):
            exec "e%d = SQLRaw('%d')" % (i, i)
        expr = And(e1, Or(e2, e3),
                   Add(e4, Mul(e5, Sub(e6, Div(e7, Div(e8, e9))))))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "1 AND (2 OR 3) AND 4+5*(6-7/(8/9))")

        expr = Func1(Select(Count()), [Select(Count())])
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "func1((SELECT COUNT(*)), (SELECT COUNT(*)))")

    def test_get_precedence(self):
        self.assertTrue(compile.get_precedence(Or) <
                        compile.get_precedence(And))
        self.assertTrue(compile.get_precedence(Add) <
                        compile.get_precedence(Mul))
        self.assertTrue(compile.get_precedence(Sub) <
                        compile.get_precedence(Div))

    def test_customize_precedence(self):
        expr = And(elem1, Or(elem2, elem3))
        custom_compile = compile.fork()
        custom_compile.set_precedence(10, And)

        custom_compile.set_precedence(11, Or)
        statement, parameters = custom_compile(expr)
        self.assertEquals(statement, "elem1 AND elem2 OR elem3")

        custom_compile.set_precedence(10, Or)
        statement, parameters = custom_compile(expr)
        self.assertEquals(statement, "elem1 AND elem2 OR elem3")

        custom_compile.set_precedence(9, Or)
        statement, parameters = custom_compile(expr)
        self.assertEquals(statement, "elem1 AND (elem2 OR elem3)")

    def test_customize_precedence_inheritance(self):
        compile_parent = compile.fork()
        compile_child = compile_parent.fork()

        expr = And(elem1, Or(elem2, elem3))

        compile_parent.set_precedence(10, And)

        compile_parent.set_precedence(11, Or)
        self.assertEquals(compile_child.get_precedence(Or), 11)
        self.assertEquals(compile_parent.get_precedence(Or), 11)
        statement, parameters = compile_child(expr)
        self.assertEquals(statement, "elem1 AND elem2 OR elem3")

        compile_parent.set_precedence(10, Or)
        self.assertEquals(compile_child.get_precedence(Or), 10)
        self.assertEquals(compile_parent.get_precedence(Or), 10)
        statement, parameters = compile_child(expr)
        self.assertEquals(statement, "elem1 AND elem2 OR elem3")

        compile_child.set_precedence(9, Or)
        self.assertEquals(compile_child.get_precedence(Or), 9)
        self.assertEquals(compile_parent.get_precedence(Or), 10)
        statement, parameters = compile_child(expr)
        self.assertEquals(statement, "elem1 AND (elem2 OR elem3)")


    def test_compile_sequence(self):
        expr = [elem1, Func1(), (Func2(), None)]
        statement, parameters = compile(expr)
        self.assertEquals(statement, "elem1, func1(), func2(), NULL")

    def test_compile_invalid(self):
        self.assertRaises(CompileError, compile, object())
        self.assertRaises(CompileError, compile, [object()])

    def test_str(self):
        statement, parameters = compile("str")
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [CharsVariable("str")])

    def test_unicode(self):
        statement, parameters = compile(u"str")
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [UnicodeVariable(u"str")])

    def test_int(self):
        statement, parameters = compile(1)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [IntVariable(1)])

    def test_long(self):
        statement, parameters = compile(1L)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [IntVariable(1)])

    def test_bool(self):
        statement, parameters = compile(True)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [BoolVariable(1)])

    def test_float(self):
        statement, parameters = compile(1.1)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [FloatVariable(1.1)])

    def test_datetime(self):
        dt = datetime(1977, 5, 4, 12, 34)
        statement, parameters = compile(dt)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [DateTimeVariable(dt)])

    def test_date(self):
        d = date(1977, 5, 4)
        statement, parameters = compile(d)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [DateVariable(d)])

    def test_time(self):
        t = time(12, 34)
        statement, parameters = compile(t)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [TimeVariable(t)])

    def test_timedelta(self):
        td = timedelta(days=1, seconds=2, microseconds=3)
        statement, parameters = compile(td)
        self.assertEquals(statement, "?")
        self.assertEquals(parameters, [TimeDeltaVariable(td)])

    def test_none(self):
        statement, parameters = compile(None)
        self.assertEquals(statement, "NULL")
        self.assertEquals(parameters, [])

    def test_select(self):
        expr = Select([column1, column2])
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1, column2")
        self.assertEquals(parameters, [])

    def test_select_distinct(self):
        expr = Select([column1, column2], Undef, [table1], distinct=True)
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT DISTINCT column1, column2 FROM table1")
        self.assertEquals(parameters, [])

    def test_select_where(self):
        expr = Select([column1, Func1()],
                      Func1(),
                      [table1, Func1()],
                      order_by=[column2, Func1()],
                      group_by=[column3, Func1()],
                      limit=3, offset=4)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1, func1() "
                                     "FROM table1, func1() "
                                     "WHERE func1() "
                                     "ORDER BY column2, func1() "
                                     "GROUP BY column3, func1() "
                                     "LIMIT 3 OFFSET 4")
        self.assertEquals(parameters, [])

    def test_select_join_where(self):
        expr = Select(column1,
                      Func1() == "value1",
                      Join(table1, Func2() == "value2"))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1 FROM "
                                     "JOIN table1 ON func2() = ? "
                                     "WHERE func1() = ?")
        self.assertEquals([variable.get() for variable in parameters],
                          ["value2", "value1"])

    def test_select_auto_table(self):
        expr = Select(Column(column1, table1),
                      Column(column2, table2) == 1),
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT table1.column1 "
                                     "FROM table1, table2 "
                                     "WHERE table2.column2 = ?")
        self.assertEquals(parameters, [Variable(1)])

    def test_select_auto_table_duplicated(self):
        expr = Select(Column(column1, table1),
                      Column(column2, table1) == 1),
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT table1.column1 "
                                     "FROM table1 WHERE table1.column2 = ?")
        self.assertEquals(parameters, [Variable(1)])

    def test_select_auto_table_default(self):
        expr = Select(Column(column1),
                      Column(column2) == 1,
                      default_tables=table1),
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1 "
                                     "FROM table1 "
                                     "WHERE column2 = ?")
        self.assertEquals(parameters, [Variable(1)])

    def test_select_auto_table_default_with_joins(self):
        expr = Select(Column(column1),
                      default_tables=[table1, Join(table2)]),
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1 "
                                     "FROM table1 JOIN table2")
        self.assertEquals(parameters, [])

    def test_select_auto_table_unknown(self):
        statement, parameters = compile(Select(elem1))
        self.assertEquals(statement, "SELECT elem1")

    def test_select_auto_table_sub(self):
        col1 = Column(column1, table1)
        col2 = Column(column2, table2)
        expr = Select(col1, In(elem1, Select(col2, col1 == col2, col2.table)))
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT table1.column1 FROM table1 WHERE "
                          "elem1 IN (SELECT table2.column2 FROM table2 WHERE "
                          "table1.column1 = table2.column2)")

    def test_select_join(self):
        expr = Select([column1, Func1()], Func1(),
                      [table1, Join(table2), Join(table3)])
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1, func1() "
                                     "FROM table1 JOIN table2 JOIN table3 "
                                     "WHERE func1()")
        self.assertEquals(parameters, [])

    def test_select_join_right_left(self):
        expr = Select([column1, Func1()], Func1(),
                      [table1, Join(table2, table3)])
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1, func1() "
                                     "FROM table1, table2 JOIN table3 "
                                     "WHERE func1()")
        self.assertEquals(parameters, [])

    def test_select_with_strings(self):
        expr = Select(column1, "1 = 2", table1, order_by="column1",
                      group_by="column2")
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1 FROM table1 "
                                     "WHERE 1 = 2 ORDER BY column1 "
                                     "GROUP BY column2")
        self.assertEquals(parameters, [])

    def test_select_with_unicode(self):
        expr = Select(column1, u"1 = 2", table1, order_by=u"column1",
                      group_by=[u"column2"])
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT column1 FROM table1 "
                                     "WHERE 1 = 2 ORDER BY column1 "
                                     "GROUP BY column2")
        self.assertEquals(parameters, [])

    def test_select_contexts(self):
        column, where, table, order_by, group_by = track_contexts(5)
        expr = Select(column, where, table,
                      order_by=order_by, group_by=group_by)
        compile(expr)
        self.assertEquals(column.context, COLUMN)
        self.assertEquals(where.context, EXPR)
        self.assertEquals(table.context, TABLE)
        self.assertEquals(order_by.context, EXPR)
        self.assertEquals(group_by.context, EXPR)

    def test_insert(self):
        expr = Insert([column1, Func1()], [elem1, Func1()], Func1())
        statement, parameters = compile(expr)
        self.assertEquals(statement, "INSERT INTO func1() (column1, func1()) "
                                     "VALUES (elem1, func1())")
        self.assertEquals(parameters, [])

    def test_insert_with_columns(self):
        expr = Insert([Column(column1, table1), Column(column2, table1)],
                      [elem1, elem2], table2)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "INSERT INTO table2 (column1, column2) "
                                     "VALUES (elem1, elem2)")
        self.assertEquals(parameters, [])

    def test_insert_auto_table(self):
        expr = Insert(Column(column1, table1), elem1)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "INSERT INTO table1 (column1) "
                                     "VALUES (elem1)")
        self.assertEquals(parameters, [])

    def test_insert_auto_table_default(self):
        expr = Insert(Column(column1), elem1, default_table=table1)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "INSERT INTO table1 (column1) "
                                     "VALUES (elem1)")
        self.assertEquals(parameters, [])

    def test_insert_auto_table_unknown(self):
        expr = Insert(Column(column1), elem1)
        self.assertRaises(NoTableError, compile, expr)

    def test_insert_contexts(self):
        column, value, table = track_contexts(3)
        expr = Insert(column, value, table)
        compile(expr)
        self.assertEquals(column.context, COLUMN_NAME)
        self.assertEquals(value.context, EXPR)
        self.assertEquals(table.context, TABLE)

    def test_update(self):
        expr = Update({column1: elem1, Func1(): Func2()}, table=Func1())
        statement, parameters = compile(expr)
        self.assertTrue(statement in
                        ["UPDATE func1() SET column1=elem1, func1()=func2()",
                         "UPDATE func1() SET func1()=func2(), column1=elem1"])
        self.assertEquals(parameters, [])

    def test_update_with_columns(self):
        expr = Update({Column(column1, table1): elem1}, table=table1)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "UPDATE table1 SET column1=elem1")
        self.assertEquals(parameters, [])

    def test_update_where(self):
        expr = Update({column1: elem1}, Func1(), Func2())
        statement, parameters = compile(expr)
        self.assertEquals(statement,