| 188 | assert cur.fetchall() == [(10, "a", "b", "c")] |
| 189 | |
| 190 | def test_executemany(self, conn): |
| 191 | cur = conn.cursor() |
| 192 | cur.execute(""" |
| 193 | create table test_compose ( |
| 194 | id serial primary key, |
| 195 | foo text, bar text, "ba'z" text) |
| 196 | """) |
| 197 | cur.executemany( |
| 198 | sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format( |
| 199 | sql.Identifier("test_compose"), |
| 200 | sql.SQL(", ").join(map(sql.Identifier, ["foo", "bar", "ba'z"])), |
| 201 | (sql.Placeholder() * 3).join(", "), |
| 202 | ), |
| 203 | [(10, "a", "b", "c"), (20, "d", "e", "f")], |
| 204 | ) |
| 205 | |
| 206 | cur.execute("select * from test_compose") |
| 207 | assert cur.fetchall() == [(10, "a", "b", "c"), (20, "d", "e", "f")] |
| 208 | |
| 209 | @pytest.mark.crdb_skip("copy") |
| 210 | def test_copy(self, conn): |