test_sql.py 文件源码

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