Source code for feast.request_feature_view

import copy
import warnings
from typing import Dict, List, Optional, Type

from feast.base_feature_view import BaseFeatureView
from feast.data_source import RequestSource
from feast.feature_view_projection import FeatureViewProjection
from feast.field import Field
from feast.protos.feast.core.RequestFeatureView_pb2 import (
    RequestFeatureView as RequestFeatureViewProto,
)
from feast.protos.feast.core.RequestFeatureView_pb2 import RequestFeatureViewSpec
from feast.usage import log_exceptions


[docs]class RequestFeatureView(BaseFeatureView): """ [Experimental] A RequestFeatureView defines a logical group of features that should be available as an input to an on demand feature view at request time. Attributes: name: The unique name of the request feature view. request_source: The request source that specifies the schema and features of the request feature view. features: The list of features defined as part of this request feature view. description: A human-readable description. tags: A dictionary of key-value pairs to store arbitrary metadata. owner: The owner of the request feature view, typically the email of the primary maintainer. """ name: str request_source: RequestSource features: List[Field] description: str tags: Dict[str, str] owner: str @log_exceptions def __init__( self, name: str, request_data_source: RequestSource, description: str = "", tags: Optional[Dict[str, str]] = None, owner: str = "", ): """ Creates a RequestFeatureView object. Args: name: The unique name of the request feature view. request_data_source: The request data source that specifies the schema and features of the request feature view. description (optional): A human-readable description. tags (optional): A dictionary of key-value pairs to store arbitrary metadata. owner (optional): The owner of the request feature view, typically the email of the primary maintainer. """ warnings.warn( "Request feature view is deprecated. " "Please use request data source instead", DeprecationWarning, ) if isinstance(request_data_source.schema, Dict): new_features = [ Field(name=name, dtype=dtype) for name, dtype in request_data_source.schema.items() ] else: new_features = request_data_source.schema super().__init__( name=name, features=new_features, description=description, tags=tags, owner=owner, ) self.request_source = request_data_source @property def proto_class(self) -> Type[RequestFeatureViewProto]: return RequestFeatureViewProto
[docs] def to_proto(self) -> RequestFeatureViewProto: """ Converts an request feature view object to its protobuf representation. Returns: A RequestFeatureViewProto protobuf. """ spec = RequestFeatureViewSpec( name=self.name, request_data_source=self.request_source.to_proto(), description=self.description, tags=self.tags, owner=self.owner, ) return RequestFeatureViewProto(spec=spec)
[docs] @classmethod def from_proto(cls, request_feature_view_proto: RequestFeatureViewProto): """ Creates a request feature view from a protobuf representation. Args: request_feature_view_proto: A protobuf representation of an request feature view. Returns: A RequestFeatureView object based on the request feature view protobuf. """ request_feature_view_obj = cls( name=request_feature_view_proto.spec.name, request_data_source=RequestSource.from_proto( request_feature_view_proto.spec.request_data_source ), description=request_feature_view_proto.spec.description, tags=dict(request_feature_view_proto.spec.tags), owner=request_feature_view_proto.spec.owner, ) # FeatureViewProjections are not saved in the RequestFeatureView proto. # Create the default projection. request_feature_view_obj.projection = FeatureViewProjection.from_definition( request_feature_view_obj ) return request_feature_view_obj
def __copy__(self): fv = RequestFeatureView(name=self.name, request_data_source=self.request_source) fv.projection = copy.copy(self.projection) return fv