(self)
| 316 | yield self._populate_data_object(row) |
| 317 | |
| 318 | def delete_all(self): # type: () -> int |
| 319 | query = self._queries.get('delete-all') |
| 320 | if not query: |
| 321 | wheres = [] |
| 322 | if self.schema.owner_column: |
| 323 | wheres.append(f'{self.schema.owner_column}=:{self.schema.owner_column}') |
| 324 | else: |
| 325 | wheres.append('1=1') |
| 326 | query = f'DELETE FROM {self.schema.table_name} WHERE ' + ' AND '.join(wheres) |
| 327 | self._queries['delete-all'] = query |
| 328 | |
| 329 | conn = self.get_connection() |
| 330 | try: |
| 331 | if self.schema.owner_column: |
| 332 | params = {self.schema.owner_column: self.owner} |
| 333 | rs = conn.execute(query, params) |
| 334 | else: |
| 335 | rs = conn.execute(query) |
| 336 | conn.commit() |
| 337 | return rs.rowcount |
| 338 | except Exception as e: |
| 339 | conn.rollback() |
| 340 | raise e |
| 341 | |
| 342 | def delete_by_filter(self, columns, values, multiple_criteria=False): |
| 343 | # type: (Union[str, Sequence[str]], Union[Any, Sequence[Any]], bool) -> int |
no test coverage detected