generative.py

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

import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.sql import table, column, ClauseElement
from sqlalchemy.sql.expression import  _clone
from testlib import *
from sqlalchemy.sql.visitors import *
from sqlalchemy import util
from sqlalchemy.sql import util as sql_util


class TraversalTest(TestBase, AssertsExecutionResults):
    """test ClauseVisitor's traversal, particularly its ability to copy and modify
    a ClauseElement in place."""

    def setUpAll(self):
        global A, B

        # establish two ficticious ClauseElements.
        # define deep equality semantics as well as deep identity semantics.
        class A(ClauseElement):
            def __init__(self, expr):
                self.expr = expr

            def is_other(self, other):
                return other is self

            def __eq__(self, other):
                return other.expr == self.expr

            def __ne__(self, other):
                return other.expr != self.expr

            def __str__(self):
                return "A(%s)" % repr(self.expr)

        class B(ClauseElement):
            def __init__(self, *items):
                self.items = items

            def is_other(self, other):
                if other is not self:
                    return False
                for i1, i2 in zip(self.items, other.items):
                    if i1 is not i2:
                        return False
                return True

            def __eq__(self, other):
                for i1, i2 in zip(self.items, other.items):
                    if i1 != i2:
                        return False
                return True

            def __ne__(self, other):
                for i1, i2 in zip(self.items, other.items):
                    if i1 != i2:
                        return True
                return False

            def _copy_internals(self, clone=_clone):
                self.items = [clone(i) for i in self.items]

            def get_children(self, **kwargs):
                return self.items

            def __str__(self):
                return "B(%s)" % repr([str(i) for i in self.items])

    def test_test_classes(self):
        a1 = A("expr1")
        struct = B(a1, A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3"))
        struct2 = B(a1, A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3"))
        struct3 = B(a1, A("expr2"), B(A("expr1b"), A("expr2bmodified")), A("expr3"))

        assert a1.is_other(a1)
        assert struct.is_other(struct)
        assert struct == struct2
        assert struct != struct3
        assert not struct.is_other(struct2)
        assert not struct.is_other(struct3)

    def test_clone(self):
        struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3"))

        class Vis(ClauseVisitor):
            def visit_a(self, a):
                pass
            def visit_b(self, b):
                pass

        vis = Vis()
        s2 = vis.traverse(struct, clone=True)
        assert struct == s2
        assert not struct.is_other(s2)

    def test_no_clone(self):
        struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3"))

        class Vis(ClauseVisitor):
            def visit_a(self, a):
                pass
            def visit_b(self, b):
                pass

        vis = Vis()
        s2 = vis.traverse(struct, clone=False)
        assert struct == s2
        assert struct.is_other(s2)

    def test_change_in_place(self):
        struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3"))
        struct2 = B(A("expr1"), A("expr2modified"), B(A("expr1b"), A("expr2b")), A("expr3"))
        struct3 = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2bmodified")), A("expr3"))

        class Vis(ClauseVisitor):
            def visit_a(self, a):
                if a.expr == "expr2":
                    a.expr = "expr2modified"
            def visit_b(self, b):
                pass

        vis = Vis()
        s2 = vis.traverse(struct, clone=True)
        assert struct != s2
        assert not struct.is_other(s2)
        assert struct2 == s2

        class Vis2(ClauseVisitor):
            def visit_a(self, a):
                if a.expr == "expr2b":
                    a.expr = "expr2bmodified"
            def visit_b(self, b):
                pass

        vis2 = Vis2()
        s3 = vis2.traverse(struct, clone=True)
        assert struct != s3
        assert struct3 == s3


class ClauseTest(TestBase, AssertsCompiledSQL):
    """test copy-in-place behavior of various ClauseElements."""

    def setUpAll(self):
        global t1, t2
        t1 = table("table1",
            column("col1"),
            column("col2"),
            column("col3"),
            )
        t2 = table("table2",
            column("col1"),
            column("col2"),
            column("col3"),
            )

    def test_binary(self):
        clause = t1.c.col2 == t2.c.col2
        assert str(clause) == ClauseVisitor().traverse(clause, clone=True)

    def test_join(self):
        clause = t1.join(t2, t1.c.col2==t2.c.col2)
        c1 = str(clause)
        assert str(clause) == str(ClauseVisitor().traverse(clause, clone=True))

        class Vis(ClauseVisitor):
            def visit_binary(self, binary):
                binary.right = t2.c.col3

        clause2 = Vis().traverse(clause, clone=True)
        assert c1 == str(clause)
        assert str(clause2) == str(t1.join(t2, t1.c.col2==t2.c.col3))

    def test_text(self):
        clause = text("select * from table where foo=:bar", bindparams=[bindparam('bar')])
        c1 = str(clause)
        class Vis(ClauseVisitor):
            def visit_textclause(self, text):
                text.text = text.text + " SOME MODIFIER=:lala"
                text.bindparams['lala'] = bindparam('lala')

        clause2 = Vis().traverse(clause, clone=True)
        assert c1 == str(clause)
        assert str(clause2) == c1 + " SOME MODIFIER=:lala"
        assert clause.bindparams.keys() == ['bar']
        assert util.Set(clause2.bindparams.keys()) == util.Set(['bar', 'lala'])

    def test_select(self):
        s2 = select([t1])
        s2_assert = str(s2)
        s3_assert = str(select([t1], t1.c.col2==7))
        class Vis(ClauseVisitor):
            def visit_select(self, select):
                select.append_whereclause(t1.c.col2==7)
        s3 = Vis().traverse(s2, clone=True)
        assert str(s3) == s3_assert
        assert str(s2) == s2_assert
        print str(s2)
        print str(s3)
        Vis().traverse(s2)
        assert str(s2) == s3_assert

        print "------------------"

        s4_assert = str(select([t1], and_(t1.c.col2==7, t1.c.col3==9)))
        class Vis(ClauseVisitor):
            def visit_select(self, select):
                select.append_whereclause(t1.c.col3==9)
        s4 = Vis().traverse(s3, clone=True)
        print str(s3)
        print str(s4)
        assert str(s4) == s4_assert
        assert str(s3) == s3_assert

        print "------------------"
        s5_assert = str(select([t1], and_(t1.c.col2==7, t1.c.col1==9)))
        class Vis(ClauseVisitor):
            def visit_binary(self, binary):
                if binary.left is t1.c.col3:
                    binary.left = t1.c.col1
                    binary.right = bindparam("table1_col1", unique=True)
        s5 = Vis().traverse(s4, clone=True)
        print str(s4)
        print str(s5)
        assert str(s5) == s5_assert
        assert str(s4) == s4_assert

    def test_binds(self):
        """test that unique bindparams change their name upon clone() to prevent conflicts"""

        s = select([t1], t1.c.col1==bindparam(None, unique=True)).alias()
        s2 = ClauseVisitor().traverse(s, clone=True).alias()
        s3 = select([s], s.c.col2==s2.c.col2)

        self.assert_compile(s3, "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, "\
        "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :param_1) AS anon_1, "\
        "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 WHERE table1.col1 = :param_2) AS anon_2 "\
        "WHERE anon_1.col2 = anon_2.col2")

        s = select([t1], t1.c.col1==4).alias()
        s2 = ClauseVisitor().traverse(s, clone=True).alias()
        s3 = select([s], s.c.col2==s2.c.col2)
        self.assert_compile(s3, "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, "\
        "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :table1_col1_1) AS anon_1, "\
        "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 WHERE table1.col1 = :table1_col1_2) AS anon_2 "\
        "WHERE anon_1.col2 = anon_2.col2")