Source code for feast.request_feature_view

import copy
from typing import Type

from feast.base_feature_view import BaseFeatureView
from feast.data_source import RequestDataSource
from feast.feature import Feature
from feast.feature_view_projection import FeatureViewProjection
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] An RequestFeatureView defines a feature that is available from the inference request. Args: name: Name of the group of features. request_data_source: Request data source that specifies the schema and features """ request_data_source: RequestDataSource @log_exceptions def __init__( self, name: str, request_data_source: RequestDataSource, ): """ Creates an RequestFeatureView object. """ super().__init__( name=name, features=[ Feature(name=name, dtype=dtype) for name, dtype in request_data_source.schema.items() ], ) self.request_data_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_data_source.to_proto() ) 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=RequestDataSource.from_proto( request_feature_view_proto.spec.request_data_source ), ) # 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_data_source ) fv.projection = copy.copy(self.projection) return fv