(
self,
table: Table,
name: str,
project: str,
proto_class: Any,
python_class: Any,
id_field_name: str,
proto_field_name: str,
not_found_exception: Optional[Callable],
)
| 1555 | return rows.rowcount |
| 1556 | |
| 1557 | def _get_object( |
| 1558 | self, |
| 1559 | table: Table, |
| 1560 | name: str, |
| 1561 | project: str, |
| 1562 | proto_class: Any, |
| 1563 | python_class: Any, |
| 1564 | id_field_name: str, |
| 1565 | proto_field_name: str, |
| 1566 | not_found_exception: Optional[Callable], |
| 1567 | ): |
| 1568 | with self.read_engine.begin() as conn: |
| 1569 | stmt = select(table).where( |
| 1570 | getattr(table.c, id_field_name) == name, table.c.project_id == project |
| 1571 | ) |
| 1572 | row = conn.execute(stmt).first() |
| 1573 | if row: |
| 1574 | _proto = proto_class.FromString(row._mapping[proto_field_name]) |
| 1575 | return python_class.from_proto(_proto) |
| 1576 | if not_found_exception: |
| 1577 | raise not_found_exception(name, project) |
| 1578 | else: |
| 1579 | return None |
| 1580 | |
| 1581 | def _list_objects( |
| 1582 | self, |
no test coverage detected