Python psycopg2 模块，sql() 实例源码

def test_execute(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.execute(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            (10, 'a', 'b', 'c'))

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])

def test_executemany(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.executemany(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

def test_execute(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.execute(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            (10, 'a', 'b', 'c'))

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])

def test_executemany(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.executemany(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

def test_execute(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.execute(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            (10, 'a', 'b', 'c'))

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])

def test_executemany(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.executemany(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

def test_copy(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)

        s = StringIO("10\ta\tb\tc\n20\td\te\tf\n")
        cur.copy_expert(
            sql.SQL("copy {t} (id, foo, bar, {f}) from stdin").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s)

        s1 = StringIO()
        cur.copy_expert(
            sql.SQL("copy (select {f} from {t} order by id) to stdout").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s1)
        s1.seek(0)
        self.assertEqual(s1.read(), 'c\nf\n')

def test_execute(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.execute(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            (10, 'a', 'b', 'c'))

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])

def test_executemany(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.executemany(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

def test_execute(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.execute(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            (10, 'a', 'b', 'c'))

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])

def test_executemany(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)
        cur.executemany(
            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(
                sql.Identifier('test_compose'),
                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
                (sql.Placeholder() * 3).join(', ')),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

        cur.execute("select * from test_compose")
        self.assertEqual(cur.fetchall(),
            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])

def test_copy(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)

        s = StringIO("10\ta\tb\tc\n20\td\te\tf\n")
        cur.copy_expert(
            sql.SQL("copy {t} (id, foo, bar, {f}) from stdin").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s)

        s1 = StringIO()
        cur.copy_expert(
            sql.SQL("copy (select {f} from {t} order by id) to stdout").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s1)
        s1.seek(0)
        self.assertEqual(s1.read(), 'c\nf\n')

def test_pos(self):
        s = sql.SQL("select {} from {}").format(
            sql.Identifier('field'), sql.Identifier('table'))
        s1 = s.as_string(self.conn)
        self.assertTrue(isinstance(s1, str))
        self.assertEqual(s1, 'select "field" from "table"')

def test_pos_spec(self):
        s = sql.SQL("select {0} from {1}").format(
            sql.Identifier('field'), sql.Identifier('table'))
        s1 = s.as_string(self.conn)
        self.assertTrue(isinstance(s1, str))
        self.assertEqual(s1, 'select "field" from "table"')

        s = sql.SQL("select {1} from {0}").format(
            sql.Identifier('table'), sql.Identifier('field'))
        s1 = s.as_string(self.conn)
        self.assertTrue(isinstance(s1, str))
        self.assertEqual(s1, 'select "field" from "table"')

def test_dict(self):
        s = sql.SQL("select {f} from {t}").format(
            f=sql.Identifier('field'), t=sql.Identifier('table'))
        s1 = s.as_string(self.conn)
        self.assertTrue(isinstance(s1, str))
        self.assertEqual(s1, 'select "field" from "table"')

def test_unicode(self):
        s = sql.SQL("select {0} from {1}").format(
            sql.Identifier('field'), sql.Identifier('table'))
        s1 = s.as_string(self.conn)
        self.assertTrue(isinstance(s1, str))
        self.assertEqual(s1, 'select "field" from "table"')

def test_compose_literal(self):
        s = sql.SQL("select {0};").format(sql.Literal(dt.date(2016, 12, 31)))
        s1 = s.as_string(self.conn)
        self.assertEqual(s1, "select '2016-12-31'::date;")

def test_percent_escape(self):
        s = sql.SQL("42 % {0}").format(sql.Literal(7))
        s1 = s.as_string(self.conn)
        self.assertEqual(s1, "42 % 7")

def test_braces_escape(self):
        s = sql.SQL("{{{0}}}").format(sql.Literal(7))
        self.assertEqual(s.as_string(self.conn), "{7}")
        s = sql.SQL("{{1,{0}}}").format(sql.Literal(7))
        self.assertEqual(s.as_string(self.conn), "{1,7}")

def test_compose_badnargs(self):
        self.assertRaises(IndexError, sql.SQL("select {0};").format)

def test_compose_badnargs_auto(self):
        self.assertRaises(IndexError, sql.SQL("select {};").format)
        self.assertRaises(ValueError, sql.SQL("select {} {1};").format, 10, 20)
        self.assertRaises(ValueError, sql.SQL("select {0} {};").format, 10, 20)

def test_compose_bad_args_type(self):
        self.assertRaises(IndexError, sql.SQL("select {0};").format, a=10)
        self.assertRaises(KeyError, sql.SQL("select {x};").format, 10)

def test_no_modifiers(self):
        self.assertRaises(ValueError, sql.SQL("select {a!r};").format, a=10)
        self.assertRaises(ValueError, sql.SQL("select {a:<};").format, a=10)

def test_must_be_adaptable(self):
        class Foo(object):
            pass

        self.assertRaises(psycopg2.ProgrammingError,
            sql.SQL("select {0};").format(sql.Literal(Foo())).as_string, self.conn)

def test_class(self):
        self.assertTrue(issubclass(sql.Identifier, sql.Composable))

def test_string(self):
        self.assertEqual(sql.Identifier('foo').string, 'foo')

def test_repr(self):
        obj = sql.Identifier("fo'o")
        self.assertEqual(repr(obj), 'Identifier("fo\'o")')
        self.assertEqual(repr(obj), str(obj))

def test_eq(self):
        self.assertTrue(sql.Identifier('foo') == sql.Identifier('foo'))
        self.assertTrue(sql.Identifier('foo') != sql.Identifier('bar'))
        self.assertTrue(sql.Identifier('foo') != 'foo')
        self.assertTrue(sql.Identifier('foo') != sql.SQL('foo'))

def test_as_str(self):
        self.assertEqual(sql.Identifier('foo').as_string(self.conn), '"foo"')
        self.assertEqual(sql.Identifier("fo'o").as_string(self.conn), '"fo\'o"')

def test_join(self):
        self.assertTrue(not hasattr(sql.Identifier('foo'), 'join'))

def test_init(self):
        self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))
        self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))
        self.assertTrue(isinstance(sql.Literal(b'foo'), sql.Literal))
        self.assertTrue(isinstance(sql.Literal(42), sql.Literal))
        self.assertTrue(isinstance(
            sql.Literal(dt.date(2016, 12, 31)), sql.Literal))

def test_wrapped(self):
        self.assertEqual(sql.Literal('foo').wrapped, 'foo')

def test_repr(self):
        self.assertEqual(repr(sql.Literal("foo")), "Literal('foo')")
        self.assertEqual(str(sql.Literal("foo")), "Literal('foo')")
        self.assertQuotedEqual(sql.Literal("foo").as_string(self.conn), "'foo'")
        self.assertEqual(sql.Literal(42).as_string(self.conn), "42")
        self.assertEqual(
            sql.Literal(dt.date(2017, 1, 1)).as_string(self.conn),
            "'2017-01-01'::date")

def test_eq(self):
        self.assertTrue(sql.Literal('foo') == sql.Literal('foo'))
        self.assertTrue(sql.Literal('foo') != sql.Literal('bar'))
        self.assertTrue(sql.Literal('foo') != 'foo')
        self.assertTrue(sql.Literal('foo') != sql.SQL('foo'))

def test_must_be_adaptable(self):
        class Foo(object):
            pass

        self.assertRaises(psycopg2.ProgrammingError,
            sql.Literal(Foo()).as_string, self.conn)

def test_init(self):
        self.assertTrue(isinstance(sql.SQL('foo'), sql.SQL))
        self.assertTrue(isinstance(sql.SQL('foo'), sql.SQL))
        self.assertRaises(TypeError, sql.SQL, 10)
        self.assertRaises(TypeError, sql.SQL, dt.date(2016, 12, 31))

def test_string(self):
        self.assertEqual(sql.SQL('foo').string, 'foo')

def test_repr(self):
        self.assertEqual(repr(sql.SQL("foo")), "SQL('foo')")
        self.assertEqual(str(sql.SQL("foo")), "SQL('foo')")
        self.assertEqual(sql.SQL("foo").as_string(self.conn), "foo")

def test_eq(self):
        self.assertTrue(sql.SQL('foo') == sql.SQL('foo'))
        self.assertTrue(sql.SQL('foo') != sql.SQL('bar'))
        self.assertTrue(sql.SQL('foo') != 'foo')
        self.assertTrue(sql.SQL('foo') != sql.Literal('foo'))

def test_sum(self):
        obj = sql.SQL("foo") + sql.SQL("bar")
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertEqual(obj.as_string(self.conn), "foobar")

def test_multiply(self):
        obj = sql.SQL("foo") * 3
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertEqual(obj.as_string(self.conn), "foofoofoo")

def test_join(self):
        obj = sql.SQL(", ").join(
            [sql.Identifier('foo'), sql.SQL('bar'), sql.Literal(42)])
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertEqual(obj.as_string(self.conn), '"foo", bar, 42')

        obj = sql.SQL(", ").join(
            sql.Composed([sql.Identifier('foo'), sql.SQL('bar'), sql.Literal(42)]))
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertEqual(obj.as_string(self.conn), '"foo", bar, 42')

        obj = sql.SQL(", ").join([])
        self.assertEqual(obj, sql.Composed([]))

def test_class(self):
        self.assertTrue(issubclass(sql.Composed, sql.Composable))

def test_repr(self):
        obj = sql.Composed([sql.Literal("foo"), sql.Identifier("b'ar")])
        self.assertEqual(repr(obj),
            """Composed([Literal('foo'), Identifier("b'ar")])""")
        self.assertEqual(str(obj), repr(obj))

def test_seq(self):
        l = [sql.SQL('foo'), sql.Literal('bar'), sql.Identifier('baz')]
        self.assertEqual(sql.Composed(l).seq, l)

def test_join(self):
        obj = sql.Composed([sql.Literal("foo"), sql.Identifier("b'ar")])
        obj = obj.join(", ")
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertQuotedEqual(obj.as_string(self.conn), "'foo', \"b'ar\"")

def test_sum(self):
        obj = sql.Composed([sql.SQL("foo ")])
        obj = obj + sql.Literal("bar")
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")

def test_sum_inplace(self):
        obj = sql.Composed([sql.SQL("foo ")])
        obj += sql.Literal("bar")
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")

        obj = sql.Composed([sql.SQL("foo ")])
        obj += sql.Composed([sql.Literal("bar")])
        self.assertTrue(isinstance(obj, sql.Composed))
        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")

def test_iter(self):
        obj = sql.Composed([sql.SQL("foo"), sql.SQL('bar')])
        it = iter(obj)
        i = next(it)
        self.assertEqual(i, sql.SQL('foo'))
        i = next(it)
        self.assertEqual(i, sql.SQL('bar'))
        self.assertRaises(StopIteration, it.__next__)

def test_class(self):
        self.assertTrue(issubclass(sql.Placeholder, sql.Composable))