(
self,
project: str,
feature_view: BaseFeatureView,
metadata_bytes: Optional[bytes],
)
| 1118 | return Infra() |
| 1119 | |
| 1120 | def apply_user_metadata( |
| 1121 | self, |
| 1122 | project: str, |
| 1123 | feature_view: BaseFeatureView, |
| 1124 | metadata_bytes: Optional[bytes], |
| 1125 | ): |
| 1126 | table = self._infer_fv_table(feature_view) |
| 1127 | |
| 1128 | name = feature_view.name |
| 1129 | with self.write_engine.begin() as conn: |
| 1130 | stmt = select(table).where( |
| 1131 | getattr(table.c, "feature_view_name") == name, |
| 1132 | table.c.project_id == project, |
| 1133 | ) |
| 1134 | row = conn.execute(stmt).first() |
| 1135 | update_datetime = _utc_now() |
| 1136 | update_time = int(update_datetime.timestamp()) |
| 1137 | if row: |
| 1138 | values = { |
| 1139 | "user_metadata": metadata_bytes, |
| 1140 | "last_updated_timestamp": update_time, |
| 1141 | } |
| 1142 | update_stmt = ( |
| 1143 | update(table) |
| 1144 | .where( |
| 1145 | getattr(table.c, "feature_view_name") == name, |
| 1146 | table.c.project_id == project, |
| 1147 | ) |
| 1148 | .values( |
| 1149 | values, |
| 1150 | ) |
| 1151 | ) |
| 1152 | conn.execute(update_stmt) |
| 1153 | else: |
| 1154 | raise FeatureViewNotFoundException(feature_view.name, project=project) |
| 1155 | |
| 1156 | def _infer_fv_table(self, feature_view): |
| 1157 | if isinstance(feature_view, LabelView): |
nothing calls this directly
no test coverage detected