# Copyright 2019 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.

import asyncio

import itertools

import os

import warnings

from datetime import datetime, timedelta

from pathlib import Path

from typing import (

Any,

Callable,

Dict,

Iterable,

List,

Mapping,

Optional,

Sequence,

Tuple,

Union,

cast,

)

import pandas as pd

import pyarrow as pa

from colorama import Fore, Style

from fastapi.concurrency import run_in_threadpool

from google.protobuf.timestamp_pb2 import Timestamp

from tqdm import tqdm

from feast import feature_server, flags_helper, ui_server, utils

from feast.base_feature_view import BaseFeatureView

from feast.batch_feature_view import BatchFeatureView

from feast.data_source import (

DataSource,

KafkaSource,

KinesisSource,

PushMode,

PushSource,

)

from feast.diff.infra_diff import InfraDiff, diff_infra_protos

from feast.diff.registry_diff import RegistryDiff, apply_diff_to_registry, diff_between

from feast.dqm.errors import ValidationFailed

from feast.entity import Entity

from feast.errors import (

DataFrameSerializationError,

DataSourceRepeatNamesException,

FeatureViewNotFoundException,

PushSourceNotFoundException,

RequestDataNotFoundInEntityDfException,

)

from feast.feast_object import FeastObject

from feast.feature_service import FeatureService

from feast.feature_view import DUMMY_ENTITY, DUMMY_ENTITY_NAME, FeatureView

from feast.field import Field

from feast.inference import (

update_data_sources_with_inferred_event_timestamp_col,

update_feature_views_with_inferred_features_and_entities,

)

from feast.infra.infra_object import Infra

from feast.infra.provider import Provider, RetrievalJob, get_provider

from feast.infra.registry.base_registry import BaseRegistry

from feast.infra.registry.registry import Registry

from feast.infra.registry.sql import SqlRegistry

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.online_response import OnlineResponse

from feast.permissions.permission import Permission

from feast.project import Project

from feast.protos.feast.core.InfraObject_pb2 import Infra as InfraProto

from feast.protos.feast.serving.ServingService_pb2 import (

FieldStatus,

GetOnlineFeaturesResponse,

)

from feast.protos.feast.types.EntityKey_pb2 import EntityKey

from feast.protos.feast.types.Value_pb2 import RepeatedValue, Value

from feast.protos.feast.types.Value_pb2 import Value as ValueProto

from feast.repo_config import RepoConfig, load_repo_config

from feast.repo_contents import RepoContents

from feast.saved_dataset import SavedDataset, SavedDatasetStorage, ValidationReference

from feast.ssl_ca_trust_store_setup import configure_ca_trust_store_env_variables

from feast.stream_feature_view import StreamFeatureView

from feast.transformation.pandas_transformation import PandasTransformation

from feast.transformation.python_transformation import PythonTransformation

from feast.utils import _utc_now

warnings.simplefilter("once", DeprecationWarning)

class FeatureStore:

"""

A FeatureStore object is used to define, create, and retrieve features.

Attributes:

config: The config for the feature store.

repo_path: The path to the feature repo.

_registry: The registry for the feature store.

_provider: The provider for the feature store.

"""

config: RepoConfig

repo_path: Path

_registry: BaseRegistry

_provider: Provider

def __init__(

self,

repo_path: Optional[str] = None,

config: Optional[RepoConfig] = None,

fs_yaml_file: Optional[Path] = None,

):

"""

Creates a FeatureStore object.

Args:

repo_path (optional): Path to the feature repo. Defaults to the current working directory.

config (optional): Configuration object used to configure the feature store.

fs_yaml_file (optional): Path to the `feature_store.yaml` file used to configure the feature store.

At most one of 'fs_yaml_file' and 'config' can be set.

Raises:

ValueError: If both or neither of repo_path and config are specified.

"""

if fs_yaml_file is not None and config is not None:

raise ValueError("You cannot specify both fs_yaml_file and config.")

configure_ca_trust_store_env_variables()

if repo_path:

self.repo_path = Path(repo_path)

else:

self.repo_path = Path(os.getcwd())

# If config is specified, or fs_yaml_file is specified, those take precedence over

# the default feature_store.yaml location under repo_path.

if config is not None:

self.config = config

elif fs_yaml_file is not None:

self.config = load_repo_config(self.repo_path, fs_yaml_file)

else:

self.config = load_repo_config(

self.repo_path, utils.get_default_yaml_file_path(self.repo_path)

)

registry_config = self.config.registry

if registry_config.registry_type == "sql":

self._registry = SqlRegistry(registry_config, self.config.project, None)

elif registry_config.registry_type == "snowflake.registry":

from feast.infra.registry.snowflake import SnowflakeRegistry

self._registry = SnowflakeRegistry(

registry_config, self.config.project, None

)

elif registry_config and registry_config.registry_type == "remote":

from feast.infra.registry.remote import RemoteRegistry

self._registry = RemoteRegistry(

registry_config, self.config.project, None, self.config.auth_config

)

else:

self._registry = Registry(

self.config.project,

registry_config,

repo_path=self.repo_path,

auth_config=self.config.auth_config,

)

self._provider = get_provider(self.config)

def __repr__(self) -> str:

return (

f"FeatureStore(\n"

f" repo_path={self.repo_path!r},\n"

f" config={self.config!r},\n"

f" registry={self._registry!r},\n"

f" provider={self._provider!r}\n"

f")"

)

@property

def registry(self) -> BaseRegistry:

"""Gets the registry of this feature store."""

return self._registry

@property

def project(self) -> str:

"""Gets the project of this feature store."""

return self.config.project

def _get_provider(self) -> Provider:

# TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths

return self._provider

def refresh_registry(self):

"""Fetches and caches a copy of the feature registry in memory.

Explicitly calling this method allows for direct control of the state of the registry cache. Every time this

method is called the complete registry state will be retrieved from the remote registry store backend

(e.g., GCS, S3), and the cache timer will be reset. If refresh_registry() is run before get_online_features()

is called, then get_online_features() will use the cached registry instead of retrieving (and caching) the

registry itself.

Additionally, the TTL for the registry cache can be set to infinity (by setting it to 0), which means that

refresh_registry() will become the only way to update the cached registry. If the TTL is set to a value

greater than 0, then once the cache becomes stale (more time than the TTL has passed), a new cache will be

downloaded synchronously, which may increase latencies if the triggering method is get_online_features().

"""

self._registry.refresh(self.project)

def list_entities(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[Entity]:

"""

Retrieves the list of entities from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

tags: Filter by tags.

Returns:

A list of entities.

"""

return self._list_entities(allow_cache, tags=tags)

def _list_entities(

self,

allow_cache: bool = False,

hide_dummy_entity: bool = True,

tags: Optional[dict[str, str]] = None,

) -> List[Entity]:

all_entities = self._registry.list_entities(

self.project, allow_cache=allow_cache, tags=tags

)

return [

entity

for entity in all_entities

if entity.name != DUMMY_ENTITY_NAME or not hide_dummy_entity

]

def list_feature_services(

self, tags: Optional[dict[str, str]] = None

) -> List[FeatureService]:

"""

Retrieves the list of feature services from the registry.

Args:

tags: Filter by tags.

Returns:

A list of feature services.

"""

return self._registry.list_feature_services(self.project, tags=tags)

def _list_all_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[BaseFeatureView]:

feature_views = []

for fv in self.registry.list_all_feature_views(

self.project, allow_cache=allow_cache, tags=tags

):

if (

isinstance(fv, FeatureView)

and fv.entities

and fv.entities[0] == DUMMY_ENTITY_NAME

):

fv.entities = []

fv.entity_columns = []

feature_views.append(fv)

return feature_views

def list_all_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[BaseFeatureView]:

"""

Retrieves the list of feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

Returns:

A list of feature views.

"""

return self._list_all_feature_views(allow_cache, tags=tags)

def list_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[FeatureView]:

"""

Retrieves the list of feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

tags: Filter by tags.

Returns:

A list of feature views.

"""

return utils._list_feature_views(

self._registry, self.project, allow_cache, tags=tags

)

def list_batch_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[FeatureView]:

"""

Retrieves the list of feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

tags: Filter by tags.

Returns:

A list of feature views.

"""

return self._list_batch_feature_views(allow_cache=allow_cache, tags=tags)

def _list_batch_feature_views(

self,

allow_cache: bool = False,

hide_dummy_entity: bool = True,

tags: Optional[dict[str, str]] = None,

) -> List[FeatureView]:

feature_views = []

for fv in self._registry.list_feature_views(

self.project, allow_cache=allow_cache, tags=tags

):

if (

hide_dummy_entity

and fv.entities

and fv.entities[0] == DUMMY_ENTITY_NAME

):

fv.entities = []

fv.entity_columns = []

feature_views.append(fv)

return feature_views

def _list_stream_feature_views(

self,

allow_cache: bool = False,

hide_dummy_entity: bool = True,

tags: Optional[dict[str, str]] = None,

) -> List[StreamFeatureView]:

stream_feature_views = []

for sfv in self._registry.list_stream_feature_views(

self.project, allow_cache=allow_cache, tags=tags

):

if hide_dummy_entity and sfv.entities[0] == DUMMY_ENTITY_NAME:

sfv.entities = []

sfv.entity_columns = []

stream_feature_views.append(sfv)

return stream_feature_views

def list_on_demand_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[OnDemandFeatureView]:

"""

Retrieves the list of on demand feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

tags: Filter by tags.

Returns:

A list of on demand feature views.

"""

return self._registry.list_on_demand_feature_views(

self.project, allow_cache=allow_cache, tags=tags

)

def list_stream_feature_views(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[StreamFeatureView]:

"""

Retrieves the list of stream feature views from the registry.

Returns:

A list of stream feature views.

"""

return self._list_stream_feature_views(allow_cache, tags=tags)

def list_data_sources(

self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None

) -> List[DataSource]:

"""

Retrieves the list of data sources from the registry.

Args:

allow_cache: Whether to allow returning data sources from a cached registry.

tags: Filter by tags.

Returns:

A list of data sources.

"""

return self._registry.list_data_sources(

self.project, allow_cache=allow_cache, tags=tags

)

def get_entity(self, name: str, allow_registry_cache: bool = False) -> Entity:

"""

Retrieves an entity.

Args:

name: Name of entity.

allow_registry_cache: (Optional) Whether to allow returning this entity from a cached registry

Returns:

The specified entity.

Raises:

EntityNotFoundException: The entity could not be found.

"""

return self._registry.get_entity(

name, self.project, allow_cache=allow_registry_cache

)

def get_feature_service(

self, name: str, allow_cache: bool = False

) -> FeatureService:

"""

Retrieves a feature service.

Args:

name: Name of feature service.

allow_cache: Whether to allow returning feature services from a cached registry.

Returns:

The specified feature service.

Raises:

FeatureServiceNotFoundException: The feature service could not be found.

"""

return self._registry.get_feature_service(name, self.project, allow_cache)

def get_feature_view(

self, name: str, allow_registry_cache: bool = False

) -> FeatureView:

"""

Retrieves a feature view.

Args:

name: Name of feature view.

allow_registry_cache: (Optional) Whether to allow returning this entity from a cached registry

Returns:

The specified feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._get_feature_view(name, allow_registry_cache=allow_registry_cache)

def _get_feature_view(

self,

name: str,

hide_dummy_entity: bool = True,

allow_registry_cache: bool = False,

) -> FeatureView:

feature_view = self._registry.get_feature_view(

name, self.project, allow_cache=allow_registry_cache

)

if hide_dummy_entity and feature_view.entities[0] == DUMMY_ENTITY_NAME:

feature_view.entities = []

return feature_view

def get_stream_feature_view(

self, name: str, allow_registry_cache: bool = False

) -> StreamFeatureView:

"""

Retrieves a stream feature view.

Args:

name: Name of stream feature view.

allow_registry_cache: (Optional) Whether to allow returning this entity from a cached registry

Returns:

The specified stream feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._get_stream_feature_view(

name, allow_registry_cache=allow_registry_cache

)

def _get_stream_feature_view(

self,

name: str,

hide_dummy_entity: bool = True,

allow_registry_cache: bool = False,

) -> StreamFeatureView:

stream_feature_view = self._registry.get_stream_feature_view(

name, self.project, allow_cache=allow_registry_cache

)

if hide_dummy_entity and stream_feature_view.entities[0] == DUMMY_ENTITY_NAME:

stream_feature_view.entities = []

return stream_feature_view

def get_on_demand_feature_view(

self, name: str, allow_registry_cache: bool = False

) -> OnDemandFeatureView:

"""

Retrieves a feature view.

Args:

name: Name of feature view.

allow_registry_cache: (Optional) Whether to allow returning this entity from a cached registry

Returns:

The specified feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._registry.get_on_demand_feature_view(

name, self.project, allow_cache=allow_registry_cache

)

def get_data_source(self, name: str) -> DataSource:

"""

Retrieves the list of data sources from the registry.

Args:

name: Name of the data source.

Returns:

The specified data source.

Raises:

DataSourceObjectNotFoundException: The data source could not be found.

"""

return self._registry.get_data_source(name, self.project)

def delete_feature_view(self, name: str):

"""

Deletes a feature view.

Args:

name: Name of feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._registry.delete_feature_view(name, self.project)

def delete_feature_service(self, name: str):

"""

Deletes a feature service.

Args:

name: Name of feature service.

Raises:

FeatureServiceNotFoundException: The feature view could not be found.

"""

return self._registry.delete_feature_service(name, self.project)

def _should_use_plan(self):

"""Returns True if plan and _apply_diffs should be used, False otherwise."""

# Currently only the local provider with sqlite online store supports plan and _apply_diffs.

return self.config.provider == "local" and (

self.config.online_store and self.config.online_store.type == "sqlite"

)

def _validate_all_feature_views(

self,

views_to_update: List[FeatureView],

odfvs_to_update: List[OnDemandFeatureView],

sfvs_to_update: List[StreamFeatureView],

):

"""Validates all feature views."""

if len(odfvs_to_update) > 0 and not flags_helper.is_test():

warnings.warn(

"On demand feature view is an experimental feature. "

"This API is stable, but the functionality does not scale well for offline retrieval",

RuntimeWarning,

)

_validate_feature_views(

[

*views_to_update,

*odfvs_to_update,

*sfvs_to_update,

]

)

def _make_inferences(

self,

data_sources_to_update: List[DataSource],

entities_to_update: List[Entity],

views_to_update: List[FeatureView],

odfvs_to_update: List[OnDemandFeatureView],

sfvs_to_update: List[StreamFeatureView],

feature_services_to_update: List[FeatureService],

):

"""Makes inferences for entities, feature views, odfvs, and feature services."""

update_data_sources_with_inferred_event_timestamp_col(

data_sources_to_update, self.config

)

update_data_sources_with_inferred_event_timestamp_col(

[view.batch_source for view in views_to_update], self.config

)

update_data_sources_with_inferred_event_timestamp_col(

[view.batch_source for view in sfvs_to_update], self.config

)

# New feature views may reference previously applied entities.

entities = self._list_entities()

provider = self._get_provider()

update_feature_views_with_inferred_features_and_entities(

provider,

views_to_update,

entities + entities_to_update,

self.config,

)

update_feature_views_with_inferred_features_and_entities(

provider,

sfvs_to_update,

entities + entities_to_update,

self.config,

)

# We need to attach the time stamp fields to the underlying data sources

# and cascade the dependencies

update_feature_views_with_inferred_features_and_entities(

provider, odfvs_to_update, entities + entities_to_update, self.config

)

# TODO(kevjumba): Update schema inference

for sfv in sfvs_to_update:

if not sfv.schema:

raise ValueError(

f"schema inference not yet supported for stream feature views. please define schema for stream feature view: {sfv.name}"

)

for odfv in odfvs_to_update:

odfv.infer_features()

odfvs_to_write = [

odfv for odfv in odfvs_to_update if odfv.write_to_online_store

]

# Update to include ODFVs with write to online store

fvs_to_update_map = {

view.name: view

for view in [*views_to_update, *sfvs_to_update, *odfvs_to_write]

}

for feature_service in feature_services_to_update:

feature_service.infer_features(fvs_to_update=fvs_to_update_map)

def _get_feature_views_to_materialize(

self,

feature_views: Optional[List[str]],

) -> List[FeatureView]:

"""

Returns the list of feature views that should be materialized.

If no feature views are specified, all feature views will be returned.

Args:

feature_views: List of names of feature views to materialize.

Raises:

FeatureViewNotFoundException: One of the specified feature views could not be found.

ValueError: One of the specified feature views is not configured for materialization.

"""

feature_views_to_materialize: List[FeatureView] = []

if feature_views is None:

feature_views_to_materialize = utils._list_feature_views(

self._registry, self.project, hide_dummy_entity=False

)

feature_views_to_materialize = [

fv for fv in feature_views_to_materialize if fv.online

]

stream_feature_views_to_materialize = self._list_stream_feature_views(

hide_dummy_entity=False

)

feature_views_to_materialize += [

sfv for sfv in stream_feature_views_to_materialize if sfv.online

]

else:

for name in feature_views:

try:

feature_view = self._get_feature_view(name, hide_dummy_entity=False)

except FeatureViewNotFoundException:

feature_view = self._get_stream_feature_view(

name, hide_dummy_entity=False

)

if not feature_view.online:

raise ValueError(

f"FeatureView {feature_view.name} is not configured to be served online."

)

feature_views_to_materialize.append(feature_view)

return feature_views_to_materialize

def plan(

self, desired_repo_contents: RepoContents

) -> Tuple[RegistryDiff, InfraDiff, Infra]:

"""Dry-run registering objects to metadata store.

The plan method dry-runs registering one or more definitions (e.g., Entity, FeatureView), and produces

a list of all the changes the that would be introduced in the feature repo. The changes computed by the plan

command are for informational purposes, and are not actually applied to the registry.

Args:

desired_repo_contents: The desired repo state.

Raises:

ValueError: The 'objects' parameter could not be parsed properly.

Examples:

Generate a plan adding an Entity and a FeatureView.

>>> from feast import FeatureStore, Entity, FeatureView, Feature, FileSource, RepoConfig

>>> from feast.feature_store import RepoContents

>>> from datetime import timedelta

>>> fs = FeatureStore(repo_path="project/feature_repo")

>>> driver = Entity(name="driver_id", description="driver id")

>>> driver_hourly_stats = FileSource(

... path="data/driver_stats.parquet",

... timestamp_field="event_timestamp",

... created_timestamp_column="created",

... )

>>> driver_hourly_stats_view = FeatureView(

... name="driver_hourly_stats",

... entities=[driver],

... ttl=timedelta(seconds=86400 * 1),

... source=driver_hourly_stats,

... )

>>> registry_diff, infra_diff, new_infra = fs.plan(RepoContents(

... projects=[Project(name="project")],

... data_sources=[driver_hourly_stats],

... feature_views=[driver_hourly_stats_view],

... on_demand_feature_views=list(),

... stream_feature_views=list(),

... entities=[driver],

... feature_services=list(),

... permissions=list())) # register entity and feature view

"""

# Validate and run inference on all the objects to be registered.

self._validate_all_feature_views(

desired_repo_contents.feature_views,

desired_repo_contents.on_demand_feature_views,

desired_repo_contents.stream_feature_views,

)

_validate_data_sources(desired_repo_contents.data_sources)

self._make_inferences(

desired_repo_contents.data_sources,

desired_repo_contents.entities,

desired_repo_contents.feature_views,

desired_repo_contents.on_demand_feature_views,

desired_repo_contents.stream_feature_views,

desired_repo_contents.feature_services,

)

# Compute the desired difference between the current objects in the registry and

# the desired repo state.

registry_diff = diff_between(

self._registry, self.project, desired_repo_contents

)

# Compute the desired difference between the current infra, as stored in the registry,

# and the desired infra.

self._registry.refresh(project=self.project)

current_infra_proto = InfraProto()

current_infra_proto.CopyFrom(self._registry.proto().infra)

desired_registry_proto = desired_repo_contents.to_registry_proto()

new_infra = self._provider.plan_infra(self.config, desired_registry_proto)

new_infra_proto = new_infra.to_proto()

infra_diff = diff_infra_protos(current_infra_proto, new_infra_proto)

return registry_diff, infra_diff, new_infra

def _apply_diffs(

self, registry_diff: RegistryDiff, infra_diff: InfraDiff, new_infra: Infra

):

"""Applies the given diffs to the metadata store and infrastructure.

Args:

registry_diff: The diff between the current registry and the desired registry.

infra_diff: The diff between the current infra and the desired infra.

new_infra: The desired infra.

"""

infra_diff.update()

apply_diff_to_registry(

self._registry, registry_diff, self.project, commit=False

)

self._registry.update_infra(new_infra, self.project, commit=True)

def apply(

self,

objects: Union[

Project,

DataSource,

Entity,

FeatureView,

OnDemandFeatureView,

BatchFeatureView,

StreamFeatureView,

FeatureService,

ValidationReference,

Permission,

List[FeastObject],

],

objects_to_delete: Optional[List[FeastObject]] = None,

partial: bool = True,

):

"""Register objects to metadata store and update related infrastructure.

The apply method registers one or more definitions (e.g., Entity, FeatureView) and registers or updates these

objects in the Feast registry. Once the apply method has updated the infrastructure (e.g., create tables in

an online store), it will commit the updated registry. All operations are idempotent, meaning they can safely

be rerun.

Args:

objects: A single object, or a list of objects that should be registered with the Feature Store.

objects_to_delete: A list of objects to be deleted from the registry and removed from the

provider's infrastructure. This deletion will only be performed if partial is set to False.

partial: If True, apply will only handle the specified objects; if False, apply will also delete

all the objects in objects_to_delete, and tear down any associated cloud resources.

Raises:

ValueError: The 'objects' parameter could not be parsed properly.

Examples:

Register an Entity and a FeatureView.

>>> from feast import FeatureStore, Entity, FeatureView, Feature, FileSource, RepoConfig

>>> from datetime import timedelta

>>> fs = FeatureStore(repo_path="project/feature_repo")

>>> driver = Entity(name="driver_id", description="driver id")

>>> driver_hourly_stats = FileSource(

... path="data/driver_stats.parquet",

... timestamp_field="event_timestamp",

... created_timestamp_column="created",

... )

>>> driver_hourly_stats_view = FeatureView(

... name="driver_hourly_stats",

... entities=[driver],

... ttl=timedelta(seconds=86400 * 1),

... source=driver_hourly_stats,

... )

>>> fs.apply([driver_hourly_stats_view, driver]) # register entity and feature view

"""

# TODO: Add locking

if not isinstance(objects, Iterable):

objects = [objects]

assert isinstance(objects, list)

if not objects_to_delete:

objects_to_delete = []

# Separate all objects into entities, feature services, and different feature view types.

projects_to_update = [ob for ob in objects if isinstance(ob, Project)]

if len(projects_to_update) > 1:

raise ValueError("Only one project can be applied at a time.")

entities_to_update = [ob for ob in objects if isinstance(ob, Entity)]

views_to_update = [

ob

for ob in objects

if (

# BFVs are not handled separately from FVs right now.

(isinstance(ob, FeatureView) or isinstance(ob, BatchFeatureView))

and not isinstance(ob, StreamFeatureView)

)

]

sfvs_to_update = [ob for ob in objects if isinstance(ob, StreamFeatureView)]

odfvs_to_update = [ob for ob in objects if isinstance(ob, OnDemandFeatureView)]

odfvs_with_writes_to_update = [

ob

for ob in objects

if isinstance(ob, OnDemandFeatureView) and ob.write_to_online_store

]

services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)]

data_sources_set_to_update = {

ob for ob in objects if isinstance(ob, DataSource)

}

validation_references_to_update = [

ob for ob in objects if isinstance(ob, ValidationReference)

]

permissions_to_update = [ob for ob in objects if isinstance(ob, Permission)]

batch_sources_to_add: List[DataSource] = []

for data_source in data_sources_set_to_update:

if (

isinstance(data_source, PushSource)

or isinstance(data_source, KafkaSource)

or isinstance(data_source, KinesisSource)

):

assert data_source.batch_source

batch_sources_to_add.append(data_source.batch_source)

for batch_source in batch_sources_to_add:

data_sources_set_to_update.add(batch_source)

for fv in itertools.chain(

views_to_update, sfvs_to_update, odfvs_with_writes_to_update

):

if isinstance(fv, FeatureView):

data_sources_set_to_update.add(fv.batch_source)

if hasattr(fv, "stream_source"):

if fv.stream_source:

data_sources_set_to_update.add(fv.stream_source)

if isinstance(fv, OnDemandFeatureView):

for source_fvp in fv.source_feature_view_projections:

odfv_batch_source: Optional[DataSource] = (

fv.source_feature_view_projections[source_fvp].batch_source

)

if odfv_batch_source is not None:

data_sources_set_to_update.add(odfv_batch_source)

else:

pass

for odfv in odfvs_to_update:

for v in odfv.source_request_sources.values():

data_sources_set_to_update.add(v)

data_sources_to_update = list(data_sources_set_to_update)

# Handle all entityless feature views by using DUMMY_ENTITY as a placeholder entity.

entities_to_update.append(DUMMY_ENTITY)

# Validate all feature views and make inferences.

self._validate_all_feature_views(

views_to_update,

odfvs_to_update,

sfvs_to_update,

)

self._make_inferences(

data_sources_to_update,

entities_to_update,

views_to_update,

odfvs_to_update,

sfvs_to_update,

services_to_update,

)

# Add all objects to the registry and update the provider's infrastructure.

for project in projects_to_update:

self._registry.apply_project(project, commit=False)

for ds in data_sources_to_update:

self._registry.apply_data_source(ds, project=self.project, commit=False)

for view in itertools.chain(views_to_update, odfvs_to_update, sfvs_to_update):

self._registry.apply_feature_view(view, project=self.project, commit=False)

for ent in entities_to_update:

self._registry.apply_entity(ent, project=self.project, commit=False)

for feature_service in services_to_update:

self._registry.apply_feature_service(

feature_service, project=self.project, commit=False

)

for validation_references in validation_references_to_update:

self._registry.apply_validation_reference(

validation_references, project=self.project, commit=False

)

for permission in permissions_to_update:

self._registry.apply_permission(

permission, project=self.project, commit=False

)

entities_to_delete = []

views_to_delete = []

sfvs_to_delete = []

permissions_to_delete = []

if not partial:

# Delete all registry objects that should not exist.

entities_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, Entity)

]

views_to_delete = [

ob

for ob in objects_to_delete

if (

(isinstance(ob, FeatureView) or isinstance(ob, BatchFeatureView))

and not isinstance(ob, StreamFeatureView)

)

]

odfvs_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, OnDemandFeatureView)

]

sfvs_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, StreamFeatureView)

]

services_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, FeatureService)

]

data_sources_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, DataSource)

]

validation_references_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, ValidationReference)

]