(
self,
name: str,
project: str,
version_number: int,
fv_type: str,
proto_bytes: bytes,
)
| 1224 | return result or 0 |
| 1225 | |
| 1226 | def _save_version_snapshot( |
| 1227 | self, |
| 1228 | name: str, |
| 1229 | project: str, |
| 1230 | version_number: int, |
| 1231 | fv_type: str, |
| 1232 | proto_bytes: bytes, |
| 1233 | ): |
| 1234 | now = int(_utc_now().timestamp()) |
| 1235 | vid = generate_version_id() |
| 1236 | with self.write_engine.begin() as conn: |
| 1237 | stmt = insert(feature_view_version_history).values( |
| 1238 | feature_view_name=name, |
| 1239 | project_id=project, |
| 1240 | version_number=version_number, |
| 1241 | feature_view_type=fv_type, |
| 1242 | feature_view_proto=proto_bytes, |
| 1243 | created_timestamp=now, |
| 1244 | description="", |
| 1245 | version_id=vid, |
| 1246 | ) |
| 1247 | conn.execute(stmt) |
| 1248 | |
| 1249 | def _get_version_snapshot( |
| 1250 | self, name: str, project: str, version_number: int |
no test coverage detected