Source code for feast.base_feature_view

# Copyright 2021 The Feast Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Dict, List, Optional, Type

from google.protobuf.json_format import MessageToJson
from proto import Message

from feast.feature_view_projection import FeatureViewProjection
from feast.field import Field


[docs]class BaseFeatureView(ABC): """ A BaseFeatureView defines a logical group of features. Attributes: name: The unique name of the base feature view. features: The list of features defined as part of this base feature view. description: A human-readable description. tags: A dictionary of key-value pairs to store arbitrary metadata. owner: The owner of the base feature view, typically the email of the primary maintainer. projection: The feature view projection storing modifications to be applied to this base feature view at retrieval time. created_timestamp (optional): The time when the base feature view was created. last_updated_timestamp (optional): The time when the base feature view was last updated. """ name: str features: List[Field] description: str tags: Dict[str, str] owner: str projection: FeatureViewProjection created_timestamp: Optional[datetime] last_updated_timestamp: Optional[datetime] @abstractmethod def __init__( self, *, name: str, features: Optional[List[Field]] = None, description: str = "", tags: Optional[Dict[str, str]] = None, owner: str = "", ): """ Creates a BaseFeatureView object. Args: name: The unique name of the base feature view. features (optional): The list of features defined as part of this base 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 base feature view, typically the email of the primary maintainer. Raises: ValueError: A field mapping conflicts with an Entity or a Feature. """ assert name is not None self.name = name self.features = features or [] self.description = description self.tags = tags or {} self.owner = owner self.projection = FeatureViewProjection.from_definition(self) self.created_timestamp = None self.last_updated_timestamp = None @property @abstractmethod def proto_class(self) -> Type[Message]: pass
[docs] @abstractmethod def to_proto(self) -> Message: pass
[docs] @classmethod @abstractmethod def from_proto(cls, feature_view_proto): pass
@abstractmethod def __copy__(self): """Returns a deep copy of this base feature view.""" pass def __repr__(self): items = (f"{k} = {v}" for k, v in self.__dict__.items()) return f"<{self.__class__.__name__}({', '.join(items)})>" def __str__(self): return str(MessageToJson(self.to_proto())) def __hash__(self): return hash((self.name)) def __getitem__(self, item): assert isinstance(item, list) referenced_features = [] for feature in self.features: if feature.name in item: referenced_features.append(feature) cp = self.__copy__() cp.projection.features = referenced_features return cp def __eq__(self, other): if not isinstance(other, BaseFeatureView): raise TypeError( "Comparisons should only involve BaseFeatureView class objects." ) if ( self.name != other.name or sorted(self.features) != sorted(other.features) or self.projection != other.projection or self.description != other.description or self.tags != other.tags or self.owner != other.owner ): return False return True
[docs] def ensure_valid(self): """ Validates the state of this feature view locally. Raises: ValueError: The feature view is invalid. """ if not self.name: raise ValueError("Feature view needs a name.")
[docs] def with_name(self, name: str): """ Returns a renamed copy of this base feature view. This renamed copy should only be used for query operations and will not modify the underlying base feature view. Args: name: The name to assign to the copy. """ cp = self.__copy__() cp.projection.name_alias = name return cp
[docs] def set_projection(self, feature_view_projection: FeatureViewProjection) -> None: """ Sets the feature view projection of this base feature view to the given projection. Args: feature_view_projection: The feature view projection to be set. Raises: ValueError: The name or features of the projection do not match. """ if feature_view_projection.name != self.name: raise ValueError( f"The projection for the {self.name} FeatureView cannot be applied because it differs in name. " f"The projection is named {feature_view_projection.name} and the name indicates which " "FeatureView the projection is for." ) for feature in feature_view_projection.features: if feature not in self.features: raise ValueError( f"The projection for {self.name} cannot be applied because it contains {feature.name} which the " "FeatureView doesn't have." ) self.projection = feature_view_projection
[docs] def with_projection(self, feature_view_projection: FeatureViewProjection): """ Returns a copy of this base feature view with the feature view projection set to the given projection. Args: feature_view_projection: The feature view projection to assign to the copy. Raises: ValueError: The name or features of the projection do not match. """ if feature_view_projection.name != self.name: raise ValueError( f"The projection for the {self.name} FeatureView cannot be applied because it differs in name. " f"The projection is named {feature_view_projection.name} and the name indicates which " "FeatureView the projection is for." ) for feature in feature_view_projection.features: if feature not in self.features: raise ValueError( f"The projection for {self.name} cannot be applied because it contains {feature.name} which the " "FeatureView doesn't have." ) cp = self.__copy__() cp.projection = feature_view_projection return cp