test_sql.py 文件源码

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')])