| 676 | ) |
| 677 | |
| 678 | def apply_feature_view( |
| 679 | self, |
| 680 | feature_view: BaseFeatureView, |
| 681 | project: str, |
| 682 | commit: bool = True, |
| 683 | no_promote: bool = False, |
| 684 | ): |
| 685 | feature_view.ensure_valid() |
| 686 | self._ensure_feature_view_name_is_unique(feature_view, project) |
| 687 | fv_table = self._infer_fv_table(feature_view) |
| 688 | fv_type_str = self._infer_fv_type_string(feature_view) |
| 689 | |
| 690 | is_latest, pin_version = parse_version(feature_view.version) |
| 691 | |
| 692 | if not is_latest: |
| 693 | # Explicit version: check if it exists (pin/revert) or not (forward declaration) |
| 694 | snapshot = self._get_version_snapshot( |
| 695 | feature_view.name, project, pin_version |
| 696 | ) |
| 697 | |
| 698 | if snapshot is not None: |
| 699 | # Version exists → pin/revert to that snapshot |
| 700 | # Check that the user hasn't also modified the definition. |
| 701 | # Compare user's FV (with version="latest") against active FV. |
| 702 | try: |
| 703 | active_fv = self._get_any_feature_view(feature_view.name, project) |
| 704 | user_fv_copy = feature_view.__copy__() |
| 705 | user_fv_copy.version = "latest" |
| 706 | active_fv.version = "latest" |
| 707 | # Clear metadata that differs due to registry state |
| 708 | user_fv_copy.created_timestamp = active_fv.created_timestamp |
| 709 | user_fv_copy.last_updated_timestamp = ( |
| 710 | active_fv.last_updated_timestamp |
| 711 | ) |
| 712 | user_fv_copy.current_version_number = ( |
| 713 | active_fv.current_version_number |
| 714 | ) |
| 715 | if hasattr(active_fv, "materialization_intervals"): |
| 716 | user_fv_copy.materialization_intervals = ( |
| 717 | active_fv.materialization_intervals |
| 718 | ) |
| 719 | if user_fv_copy != active_fv: |
| 720 | raise FeatureViewPinConflict( |
| 721 | feature_view.name, version_tag(pin_version) |
| 722 | ) |
| 723 | except FeatureViewNotFoundException: |
| 724 | pass |
| 725 | |
| 726 | snap_type, snap_proto_bytes = snapshot |
| 727 | proto_class, python_class = self._proto_class_for_type(snap_type) |
| 728 | snap_proto = proto_class.FromString(snap_proto_bytes) |
| 729 | restored_fv = python_class.from_proto(snap_proto) |
| 730 | restored_fv.version = feature_view.version |
| 731 | restored_fv.current_version_number = pin_version |
| 732 | return self._apply_object( |
| 733 | fv_table, |
| 734 | project, |
| 735 | "feature_view_name", |