(
self,
table: Table,
name: str,
project: str,
id_field_name: str,
not_found_exception: Optional[Callable],
)
| 1530 | ) |
| 1531 | |
| 1532 | def _delete_object( |
| 1533 | self, |
| 1534 | table: Table, |
| 1535 | name: str, |
| 1536 | project: str, |
| 1537 | id_field_name: str, |
| 1538 | not_found_exception: Optional[Callable], |
| 1539 | ): |
| 1540 | with self.write_engine.begin() as conn: |
| 1541 | stmt = delete(table).where( |
| 1542 | getattr(table.c, id_field_name) == name, table.c.project_id == project |
| 1543 | ) |
| 1544 | rows = conn.execute(stmt) |
| 1545 | if rows.rowcount < 1 and not_found_exception: |
| 1546 | raise not_found_exception(name, project) |
| 1547 | self.apply_project( |
| 1548 | self.get_project(name=project, allow_cache=False), commit=True |
| 1549 | ) |
| 1550 | if not self.purge_feast_metadata: |
| 1551 | self._set_last_updated_metadata(_utc_now(), project, conn) |
| 1552 | |
| 1553 | if self.cache_mode == "sync": |
| 1554 | self.refresh() |
| 1555 | return rows.rowcount |
| 1556 | |
| 1557 | def _get_object( |
| 1558 | self, |
no test coverage detected