fix: Add type: ignore comments for mypy on BaseFeatureView attr access

Signed-off-by: RutujaPathade <73137503+RutujaPathade@users.noreply.github.com>
feast-dev · ntkathole · May 23, 2026 · May 13, 2026 · May 8, 2026 · May 15, 2026
commit efd41a6f4d15240b4e23b51ac5b335312e405b2a
diff --git a/sdk/python/feast/api/registry/rest/feature_views.py b/sdk/python/feast/api/registry/rest/feature_views.py
@@ -349,6 +349,160 @@ def apply_feature_view(body: ApplyFeatureViewRequestBody):
             },
         )
 
+    @router.put("/feature_views/{name}/enable")
+    def enable_feature_view(
+        name: str,
+        project: str = Query(...),
+    ):
+        req = RegistryServer_pb2.GetAnyFeatureViewRequest(
+            name=name,
+            project=project,
+        )
+        resp = grpc_call(grpc_handler.GetAnyFeatureView, req)
+        any_fv = resp.any_feature_view
+
+        from feast.feature_view import FeatureView
+        from feast.on_demand_feature_view import OnDemandFeatureView
+        from feast.stream_feature_view import StreamFeatureView
+
+        if any_fv.HasField("feature_view"):
+            fv = FeatureView.from_proto(any_fv.feature_view)
+        elif any_fv.HasField("on_demand_feature_view"):
+            fv = OnDemandFeatureView.from_proto(any_fv.on_demand_feature_view)
+        elif any_fv.HasField("stream_feature_view"):
+            fv = StreamFeatureView.from_proto(any_fv.stream_feature_view)
+        else:
+            return JSONResponse(
+                status_code=404,
+                content={"error": f"Feature view '{name}' not found."},
+            )
+
+        fv.enabled = True
+        apply_req = RegistryServer_pb2.ApplyFeatureViewRequest(
+            project=project, commit=True
+        )
+        if isinstance(fv, StreamFeatureView):
+            apply_req.stream_feature_view.CopyFrom(fv.to_proto())
+        elif isinstance(fv, OnDemandFeatureView):
+            apply_req.on_demand_feature_view.CopyFrom(fv.to_proto())
+        else:
+            apply_req.feature_view.CopyFrom(fv.to_proto())
+        grpc_call(grpc_handler.ApplyFeatureView, apply_req)
+
+        return {"name": name, "project": project, "enabled": True}
+
+    @router.put("/feature_views/{name}/disable")
+    def disable_feature_view(
+        name: str,
+        project: str = Query(...),
+    ):
+        req = RegistryServer_pb2.GetAnyFeatureViewRequest(
+            name=name,
+            project=project,
+        )
+        resp = grpc_call(grpc_handler.GetAnyFeatureView, req)
+        any_fv = resp.any_feature_view
+
+        from feast.feature_view import FeatureView
+        from feast.on_demand_feature_view import OnDemandFeatureView
+        from feast.stream_feature_view import StreamFeatureView
+
+        if any_fv.HasField("feature_view"):
+            fv = FeatureView.from_proto(any_fv.feature_view)
+        elif any_fv.HasField("on_demand_feature_view"):
+            fv = OnDemandFeatureView.from_proto(any_fv.on_demand_feature_view)
+        elif any_fv.HasField("stream_feature_view"):
+            fv = StreamFeatureView.from_proto(any_fv.stream_feature_view)
+        else:
+            return JSONResponse(
+                status_code=404,
+                content={"error": f"Feature view '{name}' not found."},
+            )
+
+        fv.enabled = False
+        apply_req = RegistryServer_pb2.ApplyFeatureViewRequest(
+            project=project, commit=True
+        )
+        if isinstance(fv, StreamFeatureView):
+            apply_req.stream_feature_view.CopyFrom(fv.to_proto())
+        elif isinstance(fv, OnDemandFeatureView):
+            apply_req.on_demand_feature_view.CopyFrom(fv.to_proto())
+        else:
+            apply_req.feature_view.CopyFrom(fv.to_proto())
+        grpc_call(grpc_handler.ApplyFeatureView, apply_req)
+
+        return {"name": name, "project": project, "enabled": False}
+
+    @router.put("/feature_views/{name}/set-state")
+    def set_feature_view_state(
+        name: str,
+        state: str = Query(...),
+        project: str = Query(...),
+    ):
+        from feast.feature_view import (
+            _VALID_STATE_TRANSITIONS,
+            FeatureView,
+            FeatureViewState,
+        )
+        from feast.on_demand_feature_view import OnDemandFeatureView
+        from feast.stream_feature_view import StreamFeatureView
+
+        try:
+            new_state = FeatureViewState[state.upper()]
+        except KeyError:
+            return JSONResponse(
+                status_code=400,
+                content={
+                    "error": f"Invalid state '{state}'. "
+                    f"Valid states: CREATED, GENERATED, MATERIALIZING, AVAILABLE_ONLINE."
+                },
+            )
+
+        req = RegistryServer_pb2.GetAnyFeatureViewRequest(
+            name=name,
+            project=project,
+        )
+        resp = grpc_call(grpc_handler.GetAnyFeatureView, req)
+        any_fv = resp.any_feature_view
+
+        if any_fv.HasField("feature_view"):
+            fv = FeatureView.from_proto(any_fv.feature_view)
+        elif any_fv.HasField("on_demand_feature_view"):
+            fv = OnDemandFeatureView.from_proto(any_fv.on_demand_feature_view)
+        elif any_fv.HasField("stream_feature_view"):
+            fv = StreamFeatureView.from_proto(any_fv.stream_feature_view)
+        else:
+            return JSONResponse(
+                status_code=404,
+                content={"error": f"Feature view '{name}' not found."},
+            )
+
+        if not fv.state.can_transition_to(new_state):
+            current = fv.state.name
+            allowed = _VALID_STATE_TRANSITIONS.get(fv.state, set())
+            allowed_names = ", ".join(sorted(s.name for s in allowed)) or "none"
+            return JSONResponse(
+                status_code=400,
+                content={
+                    "error": f"Invalid state transition: {current} -> {new_state.name}. "
+                    f"Allowed transitions from {current}: {allowed_names}."
+                },
+            )
+
+        fv.state = new_state
+        apply_req = RegistryServer_pb2.ApplyFeatureViewRequest(
+            project=project, commit=True
+        )
+        if isinstance(fv, StreamFeatureView):
+            apply_req.stream_feature_view.CopyFrom(fv.to_proto())
+        elif isinstance(fv, OnDemandFeatureView):
+            apply_req.on_demand_feature_view.CopyFrom(fv.to_proto())
+        else:
+            apply_req.feature_view.CopyFrom(fv.to_proto())
+        grpc_call(grpc_handler.ApplyFeatureView, apply_req)
+
+        return {"name": name, "project": project, "state": new_state.name}
+
     @router.delete("/feature_views/{name}")
     def delete_feature_view(
         name: str,

@@ -806,7 +806,7 @@ def enable_feature_view(self, name: str):
             name: Name of feature view.
         """
         fv = self.registry.get_any_feature_view(name, self.project)
-        fv.enabled = True
+        fv.enabled = True  # type: ignore[attr-defined]
         self.registry.apply_feature_view(fv, self.project)
 
     def disable_feature_view(self, name: str):
@@ -817,7 +817,7 @@ def disable_feature_view(self, name: str):
             name: Name of feature view.
         """
         fv = self.registry.get_any_feature_view(name, self.project)
-        fv.enabled = False
+        fv.enabled = False  # type: ignore[attr-defined]
         self.registry.apply_feature_view(fv, self.project)
 
     def set_feature_view_state(self, name: str, state: FeatureViewState):
@@ -829,11 +829,11 @@ def set_feature_view_state(self, name: str, state: FeatureViewState):
             state: Target state.
         """
         fv = self.registry.get_any_feature_view(name, self.project)
-        if not fv.state.can_transition_to(state):
+        if not fv.state.can_transition_to(state):  # type: ignore[attr-defined]
             raise ValueError(
-                f"Invalid state transition: {fv.state.name} -> {state.name}."
+                f"Invalid state transition: {fv.state.name} -> {state.name}."  # type: ignore[attr-defined]
             )
-        fv.state = state
+        fv.state = state  # type: ignore[attr-defined]
         self.registry.apply_feature_view(fv, self.project)
 
     def delete_feature_service(self, name: str):