# 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, Union
from google.protobuf.json_format import MessageToJson
from google.protobuf.message import Message
from feast.data_source import DataSource
from feast.feature_view_projection import FeatureViewProjection
from feast.field import Field
from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
OnDemandFeatureView as OnDemandFeatureViewProto,
)
from feast.protos.feast.core.StreamFeatureView_pb2 import (
StreamFeatureView as StreamFeatureViewProto,
)
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: The time when the base feature view was created.
last_updated_timestamp: 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 = "",
source: Optional[DataSource] = None,
):
"""
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.
source (optional): The source of data for this group of features. May be a stream source, or a batch source.
If a stream source, the source should contain a batch_source for backfills & batch materialization.
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
if source:
self.source = source
@property
@abstractmethod
def proto_class(self) -> Type[Message]:
pass
@abstractmethod
def to_proto(
self,
) -> Union[FeatureViewProto, OnDemandFeatureViewProto, StreamFeatureViewProto]:
pass
@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""
def __str__(self):
return str(MessageToJson(self.to_proto()))
def __hash__(self):
return hash(self.name)
def __getitem__(self, item):
assert isinstance(item, list)
cp = self.__copy__()
if self.features:
feature_name_to_feature = {
feature.name: feature for feature in self.features
}
referenced_features = []
for feature in item:
if feature not in feature_name_to_feature:
raise ValueError(
f"Feature {feature} does not exist in this feature view."
)
referenced_features.append(feature_name_to_feature[feature])
cp.projection.features = referenced_features
else:
cp.projection.desired_features = item
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
):
# This is meant to ignore the File Source change to Push Source
if isinstance(type(self.source), type(other.source)):
if self.source != other.source:
return False
return False
return True
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.")
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
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
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