Merge remote-tracking branch 'upstream/master'

Author: dchourasia

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

@@ -0,0 +1,51 @@
+import logging
+from pathlib import Path
+
+from jinja2 import Environment, FileSystemLoader
+
+
+def render_template(template_name, context):
+    template_dir = Path(__file__).parent / "templates"
+    env = Environment(
+        loader=FileSystemLoader(str(template_dir)), trim_blocks=True, lstrip_blocks=True
+    )
+    template = env.get_template(template_name)
+    try:
+        return template.render(**context)
+    except Exception as e:
+        logging.warning(
+            f"Failed to render template {template_name} for {context.get('name', 'unknown')}: {e}"
+        )
+        return ""
+
+
+def render_feature_view_code(context):
+    return render_template("feature_view_template.jinja2", context)
+
+
+def render_entity_code(context):
+    return render_template("entity_template.jinja2", context)
+
+
+def render_data_source_code(context):
+    return render_template("data_source_template.jinja2", context)
+
+
+def render_feature_service_code(context):
+    return render_template("feature_service_template.jinja2", context)
+
+
+def render_feature_code(context):
+    return render_template("feature_template.jinja2", context)
+
+
+def render_saved_dataset_code(context):
+    return render_template("saved_dataset_template.jinja2", context)
+
+
+def render_request_source_code(context):
+    return render_template("request_source_template.jinja2", context)
+
+
+def render_push_source_code(context):
+    return render_template("push_source_template.jinja2", context)

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

@@ -3,6 +3,11 @@
 
 from fastapi import APIRouter, Depends, Query
 
+from feast.api.registry.rest.codegen_utils import (
+    render_data_source_code,
+    render_push_source_code,
+    render_request_source_code,
+)
 from feast.api.registry.rest.rest_utils import (
     aggregate_across_projects,
     create_grpc_pagination_params,
@@ -15,6 +20,8 @@
     parse_tags,
 )
 from feast.protos.feast.registry import RegistryServer_pb2
+from feast.type_map import _convert_value_type_str_to_value_type
+from feast.types import from_value_type
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +51,7 @@ def list_data_sources(
         data_sources = response.get("dataSources", [])
 
         result = {
-            "data_sources": data_sources,
+            "dataSources": data_sources,
             "pagination": response.get("pagination", {}),
         }
 
@@ -105,6 +112,49 @@ def get_data_source(
             )
             result["relationships"] = relationships
 
-        return result
+        if result:
+            spec = result.get("spec", result)
+            name = spec.get("name", result.get("name", "default_source"))
+            source_type = spec.get("type", "").upper()
+            if source_type == "REQUEST_SOURCE":
+                schema_fields = []
+                feast_types = set()
+                for field in spec.get("requestDataOptions", {}).get("schema", []):
+                    value_type_enum = _convert_value_type_str_to_value_type(
+                        field.get("valueType", "").upper()
+                    )
+                    feast_type = from_value_type(value_type_enum)
+                    dtype = getattr(feast_type, "__name__", str(feast_type))
+                    schema_fields.append(
+                        f'        Field(name="{field["name"]}", dtype={dtype}),'
+                    )
+                    feast_types.add(dtype)
+                context = dict(
+                    name=name,
+                    schema_lines=schema_fields,
+                    feast_types=list(feast_types),
+                )
+                result["featureDefinition"] = render_request_source_code(context)
+                return result
+            elif source_type == "PUSH_SOURCE" and "batchSource" in spec:
+                batch_source_name = spec["batchSource"].get("name", "batch_source")
+                context = dict(
+                    name=name,
+                    batch_source_name=batch_source_name,
+                )
+                result["featureDefinition"] = render_push_source_code(context)
+                return result
+            else:
+                path = spec.get("path") or spec.get("fileOptions", {}).get("uri", "")
+                timestamp_field = spec.get("timestampField", "event_timestamp")
+                created_timestamp_column = spec.get("createdTimestampColumn", "")
+                context = dict(
+                    name=name,
+                    path=path,
+                    timestamp_field=timestamp_field,
+                    created_timestamp_column=created_timestamp_column,
+                )
+                result["featureDefinition"] = render_data_source_code(context)
+                return result
 
     return router

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

@@ -2,6 +2,7 @@
 
 from fastapi import APIRouter, Depends, Query
 
+from feast.api.registry.rest.codegen_utils import render_entity_code
 from feast.api.registry.rest.rest_utils import (
     aggregate_across_projects,
     create_grpc_pagination_params,
@@ -95,12 +96,44 @@ def get_entity(
 
         result = entity
 
+        relationships = get_object_relationships(
+            grpc_handler, "entity", name, project, allow_cache
+        )
+        ds_list_req = RegistryServer_pb2.ListDataSourcesRequest(
+            project=project,
+            allow_cache=allow_cache,
+        )
+        ds_list_resp = grpc_call(grpc_handler.ListDataSources, ds_list_req)
+        ds_map = {ds["name"]: ds for ds in ds_list_resp.get("dataSources", [])}
+        data_source_objs = []
+        seen_ds_names = set()
+        for rel in relationships:
+            if rel.get("target", {}).get("type") == "dataSource":
+                ds_name = rel["target"]["name"]
+                if ds_name not in seen_ds_names:
+                    ds_obj = ds_map.get(ds_name)
+                    if ds_obj:
+                        data_source_objs.append(ds_obj)
+                        seen_ds_names.add(ds_name)
+        result["dataSources"] = data_source_objs
+
         if include_relationships:
-            relationships = get_object_relationships(
-                grpc_handler, "entity", name, project, allow_cache
-            )
             result["relationships"] = relationships
 
+        if result:
+            spec = result.get("spec", result)
+            name = spec.get("name") or result.get("name") or "default_entity"
+            join_keys = spec.get("joinKeys") or (
+                [spec["joinKey"]] if "joinKey" in spec else []
+            )
+
+            context = {
+                "name": name,
+                "join_keys": join_keys,
+                "description": spec.get("description", ""),
+                "tags": spec.get("tags", {}),
+            }
+            result["featureDefinition"] = render_entity_code(context)
         return result
 
     return router

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

@@ -2,6 +2,7 @@
 
 from fastapi import APIRouter, Depends, Query
 
+from feast.api.registry.rest.codegen_utils import render_feature_service_code
 from feast.api.registry.rest.rest_utils import (
     aggregate_across_projects,
     create_grpc_pagination_params,
@@ -102,6 +103,42 @@ def get_feature_service(
             )
             result["relationships"] = relationships
 
+        if result:
+            spec = result.get("spec", result)
+            name = spec.get("name") or result.get("name") or "default_feature_service"
+            projections = spec.get("features", [])
+            if not isinstance(projections, list):
+                projections = []
+
+            features_exprs = []
+            for proj in projections:
+                if isinstance(proj, dict):
+                    view_name = proj.get("name")
+                    feature_names = proj.get("features", [])
+                else:
+                    view_name = str(proj)
+                    feature_names = []
+
+                if not view_name:
+                    continue
+
+                if feature_names:
+                    feature_list = ", ".join([repr(f) for f in feature_names])
+                    features_exprs.append(f"{view_name}[[{feature_list}]]")
+                else:
+                    features_exprs.append(view_name)
+
+            features_str = ", ".join(features_exprs)
+
+            context = {
+                "name": name,
+                "features": features_str,
+                "tags": spec.get("tags", {}),
+                "description": spec.get("description", ""),
+                "logging_config": spec.get("loggingConfig"),
+            }
+
+            result["featureDefinition"] = render_feature_service_code(context)
         return result
 
     return router

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

@@ -2,6 +2,7 @@
 
 from fastapi import APIRouter, Depends, Query
 
+from feast.api.registry.rest.codegen_utils import render_feature_view_code
 from feast.api.registry.rest.rest_utils import (
     create_grpc_pagination_params,
     create_grpc_sorting_params,
@@ -14,6 +15,8 @@
     parse_tags,
 )
 from feast.registry_server import RegistryServer_pb2
+from feast.type_map import _convert_value_type_str_to_value_type
+from feast.types import from_value_type
 
 
 def _extract_feature_view_from_any(any_feature_view: dict) -> dict:
@@ -32,6 +35,20 @@ def _extract_feature_view_from_any(any_feature_view: dict) -> dict:
     return {}
 
 
+def extract_feast_types_from_fields(fields):
+    types = set()
+    for field in fields:
+        value_type_enum = _convert_value_type_str_to_value_type(
+            field.get("valueType", "").upper()
+        )
+        feast_type = from_value_type(value_type_enum)
+        dtype = (
+            feast_type.__name__ if hasattr(feast_type, "__name__") else str(feast_type)
+        )
+        types.add(dtype)
+    return list(types)
+
+
 def get_feature_view_router(grpc_handler) -> APIRouter:
     router = APIRouter()
 
@@ -116,6 +133,93 @@ def get_any_feature_view(
             )
             result["relationships"] = relationships
 
+        if result and "spec" in result:
+            spec = result["spec"]
+            fv_type = result.get("type", "featureView")
+            features = spec.get("features", [])
+            if not isinstance(features, list):
+                features = []
+            if fv_type == "onDemandFeatureView":
+                class_name = "OnDemandFeatureView"
+                sources = spec.get("sources", {})
+                if sources:
+                    source_exprs = []
+                    for k, v in sources.items():
+                        var_name = v.get("name", k) if isinstance(v, dict) else str(v)
+                        source_exprs.append(f'"{k}": {var_name}')
+                    source_name = "{" + ", ".join(source_exprs) + "}"
+                else:
+                    source_name = "source_feature_view"
+            elif fv_type == "streamFeatureView":
+                class_name = "StreamFeatureView"
+                stream_source = spec.get("streamSource", {})
+                source_name = stream_source.get("name", "stream_source")
+            else:
+                class_name = "FeatureView"
+                source_views = spec.get("source_views") or spec.get("sourceViews")
+                if source_views and isinstance(source_views, list):
+                    source_vars = [sv.get("name", "source_view") for sv in source_views]
+                    source_name = "[" + ", ".join(source_vars) + "]"
+                else:
+                    source = spec.get("source")
+                    if isinstance(source, dict) and source.get("name"):
+                        source_name = source["name"]
+                    else:
+                        batch_source = spec.get("batchSource", {})
+                        source_name = batch_source.get("name", "driver_stats_source")
+
+            # Entities
+            entities = spec.get("entities") or []
+            entities_str = ", ".join(entities)
+
+            # Feature schema
+            schema_lines = []
+            for field in features:
+                value_type_enum = _convert_value_type_str_to_value_type(
+                    field.get("valueType", "").upper()
+                )
+                feast_type = from_value_type(value_type_enum)
+                dtype = getattr(feast_type, "__name__", str(feast_type))
+                desc = field.get("description")
+                desc_str = f', description="{desc}"' if desc else ""
+                schema_lines.append(
+                    f'        Field(name="{field["name"]}", dtype={dtype}{desc_str}),'
+                )
+
+            # Feast types
+            feast_types = extract_feast_types_from_fields(features)
+
+            # Tags
+            tags = spec.get("tags", {})
+            tags_str = f"tags={tags}," if tags else ""
+
+            # TTL
+            ttl = spec.get("ttl")
+            ttl_str = "timedelta(days=1)"
+            if ttl:
+                if isinstance(ttl, int):
+                    ttl_str = f"timedelta(seconds={ttl})"
+                elif isinstance(ttl, str) and ttl.endswith("s") and ttl[:-1].isdigit():
+                    ttl_str = f"timedelta(seconds={int(ttl[:-1])})"
+
+            # Online
+            online = spec.get("online", True)
+
+            # Build context
+            context = dict(
+                class_name=class_name,
+                name=spec.get("name", "example"),
+                entities_str=entities_str,
+                ttl_str=ttl_str,
+                schema_lines=schema_lines,
+                online=online,
+                source_name=source_name,
+                tags_str=tags_str,
+                feast_types=feast_types,
+            )
+
+            result["featureDefinition"] = render_feature_view_code(context)
+
         return result
 
     @router.get("/feature_views")