fix: Resolve feature view batch source name against registry

POST /api/v1/feature_views takes the batch source as a name, but the
handler turned that name into a DataSourceProto with every other field
left at its default. With `type` unset, DataSource.from_proto rejected
the spec with "Could not identify the source type being added.", so
creating a feature view from the UI always failed.

Resolve the name through the registry instead, so the feature view spec
carries the registered source's type along with its options (path,
timestamp columns, connection settings). Setting `type` alone would not
be enough: each concrete source reads its own options out of the proto,
which would persist a feature view pointing at a gutted source.

An unregistered name now raises FeastObjectNotFoundException, which both
the standalone REST registry server and the UI server already map to a
404 instead of the previous 500.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019AKoXg1p8EkTABGC46PBbw
Signed-off-by: patelchaitany <patelchaitany93@gmail.com>

Author: patelchaitany

sdk/python/feast/api/registry/rest/feature_views.py

@@ -6,6 +6,7 @@
 from fastapi.responses import JSONResponse
 from google.protobuf import timestamp_pb2
 from google.protobuf.duration_pb2 import Duration
+from google.protobuf.json_format import ParseDict
 from pydantic import BaseModel
 
 from feast.api.registry.rest.codegen_utils import render_feature_view_code
@@ -343,9 +344,23 @@ def apply_feature_view(body: ApplyFeatureViewRequestBody):
                 )
             )
 
-        batch_source_proto = (
-            DataSourceProto(name=body.batch_source) if body.batch_source else None
-        )
+        batch_source_proto = None
+        if body.batch_source:
+            # The body carries only the source's name, so read the registered
+            # source back out of the registry. A proto built from the name alone
+            # has no type, and DataSource.from_proto then rejects it with
+            # "Could not identify the source type being added."
+            batch_source_proto = ParseDict(
+                grpc_call(
+                    grpc_handler.GetDataSource,
+                    RegistryServer_pb2.GetDataSourceRequest(
+                        name=body.batch_source,
+                        project=body.project,
+                        allow_cache=False,
+                    ),
+                ),
+                DataSourceProto(),
+            )
 
         ttl = (
             Duration(seconds=body.ttl_seconds) if body.ttl_seconds is not None else None