(self, connection)
| 156 | |
| 157 | @testing.requires.insert_executemany_returning |
| 158 | def test_splice_horizontally(self, connection): |
| 159 | users = self.tables.users |
| 160 | addresses = self.tables.addresses |
| 161 | |
| 162 | r1 = connection.execute( |
| 163 | users.insert().returning(users.c.user_name, users.c.user_id), |
| 164 | [ |
| 165 | dict(user_id=1, user_name="john"), |
| 166 | dict(user_id=2, user_name="jack"), |
| 167 | ], |
| 168 | ) |
| 169 | |
| 170 | r2 = connection.execute( |
| 171 | addresses.insert().returning( |
| 172 | addresses.c.address_id, |
| 173 | addresses.c.address, |
| 174 | addresses.c.user_id, |
| 175 | ), |
| 176 | [ |
| 177 | dict(address_id=1, user_id=1, address="foo@bar.com"), |
| 178 | dict(address_id=2, user_id=2, address="bar@bat.com"), |
| 179 | ], |
| 180 | ) |
| 181 | |
| 182 | rows = r1.splice_horizontally(r2).all() |
| 183 | eq_( |
| 184 | rows, |
| 185 | [ |
| 186 | ("john", 1, 1, "foo@bar.com", 1), |
| 187 | ("jack", 2, 2, "bar@bat.com", 2), |
| 188 | ], |
| 189 | ) |
| 190 | |
| 191 | eq_(rows[0]._mapping[users.c.user_id], 1) |
| 192 | eq_(rows[0]._mapping[addresses.c.user_id], 1) |
| 193 | eq_(rows[1].address, "bar@bat.com") |
| 194 | |
| 195 | with expect_raises_message( |
| 196 | exc.InvalidRequestError, "Ambiguous column name 'user_id'" |
| 197 | ): |
| 198 | rows[0].user_id |
| 199 | |
| 200 | def test_keys_no_rows(self, connection): |
| 201 | for i in range(2): |
nothing calls this directly
no test coverage detected