diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index a3a151ab223..ddc85bb21df 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -22,7 +22,7 @@ _get_requested_feature_views_to_features_dict, ) from feast.registry import Registry -from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.repo_config import FeastOfflineStoreConfigBaseModel, RepoConfig try: from google.api_core.exceptions import NotFound @@ -36,7 +36,7 @@ raise FeastExtrasDependencyImportError("gcp", str(e)) -class BigQueryOfflineStoreConfig(FeastConfigBaseModel): +class BigQueryOfflineStoreConfig(FeastOfflineStoreConfigBaseModel): """ Offline store config for GCP BigQuery """ type: Literal["bigquery"] = "bigquery" @@ -48,6 +48,9 @@ class BigQueryOfflineStoreConfig(FeastConfigBaseModel): project_id: Optional[StrictStr] = None """ (optional) GCP project name used for the BigQuery offline store """ + def supports_data_source(self, data_source: DataSource) -> bool: + return isinstance(data_source, BigQuerySource) + class BigQueryOfflineStore(OfflineStore): @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/file.py b/sdk/python/feast/infra/offline_stores/file.py index 70b33726659..85125a67c9e 100644 --- a/sdk/python/feast/infra/offline_stores/file.py +++ b/sdk/python/feast/infra/offline_stores/file.py @@ -16,15 +16,18 @@ _run_field_mapping, ) from feast.registry import Registry -from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.repo_config import FeastOfflineStoreConfigBaseModel, RepoConfig -class FileOfflineStoreConfig(FeastConfigBaseModel): +class FileOfflineStoreConfig(FeastOfflineStoreConfigBaseModel): """ Offline store config for local (file-based) store """ type: Literal["file"] = "file" """ Offline store type selector""" + def supports_data_source(self, data_source: DataSource) -> bool: + return isinstance(data_source, FileSource) + class FileRetrievalJob(RetrievalJob): def __init__(self, evaluation_function: Callable): diff --git a/sdk/python/feast/infra/online_stores/datastore.py b/sdk/python/feast/infra/online_stores/datastore.py index c623af1c1f8..f8abf795694 100644 --- a/sdk/python/feast/infra/online_stores/datastore.py +++ b/sdk/python/feast/infra/online_stores/datastore.py @@ -26,7 +26,7 @@ from feast.infra.online_stores.online_store import OnlineStore from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.repo_config import FeastOnlineStoreConfigBaseModel, RepoConfig try: from google.auth.exceptions import DefaultCredentialsError @@ -42,7 +42,7 @@ ] -class DatastoreOnlineStoreConfig(FeastConfigBaseModel): +class DatastoreOnlineStoreConfig(FeastOnlineStoreConfigBaseModel): """ Online store config for GCP Datastore """ type: Literal["datastore"] = "datastore" @@ -60,6 +60,10 @@ class DatastoreOnlineStoreConfig(FeastConfigBaseModel): write_batch_size: Optional[PositiveInt] = 50 """ (optional) Amount of feature rows per batch being written into Datastore""" + def supports_offline_store(self, offline_store: Any) -> bool: + # We're defaulting to supporting all offline stores for now. + return offline_store.type == "bigquery" + class DatastoreOnlineStore(OnlineStore): """ diff --git a/sdk/python/feast/infra/online_stores/redis.py b/sdk/python/feast/infra/online_stores/redis.py index bb85a8e853d..3cdcf4a52ab 100644 --- a/sdk/python/feast/infra/online_stores/redis.py +++ b/sdk/python/feast/infra/online_stores/redis.py @@ -25,7 +25,7 @@ from feast.infra.online_stores.online_store import OnlineStore from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.repo_config import FeastConfigBaseModel +from feast.repo_config import FeastOnlineStoreConfigBaseModel try: from redis import Redis @@ -43,7 +43,7 @@ class RedisType(str, Enum): redis_cluster = "redis_cluster" -class RedisOnlineStoreConfig(FeastConfigBaseModel): +class RedisOnlineStoreConfig(FeastOnlineStoreConfigBaseModel): """Online store config for Redis store""" type: Literal["redis"] = "redis" @@ -56,6 +56,10 @@ class RedisOnlineStoreConfig(FeastConfigBaseModel): """Connection string containing the host, port, and configuration parameters for Redis format: host:port,parameter1,parameter2 eg. redis:6379,db=0 """ + def supports_offline_store(self, offline_store: Any) -> bool: + # We're defaulting to supporting all offline stores for now. + return True + class RedisOnlineStore(OnlineStore): _client: Optional[Union[Redis, RedisCluster]] = None diff --git a/sdk/python/feast/infra/online_stores/sqlite.py b/sdk/python/feast/infra/online_stores/sqlite.py index dbd837c5dfc..d9645d48a96 100644 --- a/sdk/python/feast/infra/online_stores/sqlite.py +++ b/sdk/python/feast/infra/online_stores/sqlite.py @@ -28,10 +28,10 @@ from feast.infra.online_stores.online_store import OnlineStore from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.repo_config import FeastOnlineStoreConfigBaseModel, RepoConfig -class SqliteOnlineStoreConfig(FeastConfigBaseModel): +class SqliteOnlineStoreConfig(FeastOnlineStoreConfigBaseModel): """ Online store config for local (SQLite-based) store """ type: Literal[ @@ -42,6 +42,9 @@ class SqliteOnlineStoreConfig(FeastConfigBaseModel): path: StrictStr = "data/online.db" """ (optional) Path to sqlite db """ + def supports_offline_store(self, offline_store: Any) -> bool: + return offline_store.type == "file" + class SqliteOnlineStore(OnlineStore): """ diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index c680d94d07a..163e28391e8 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -1,3 +1,4 @@ +from abc import abstractmethod from pathlib import Path from typing import Any @@ -6,6 +7,7 @@ from pydantic.error_wrappers import ErrorWrapper from pydantic.typing import Dict, Optional, Union +from feast.data_source import DataSource from feast.importer import get_class_from_type from feast.usage import log_exceptions @@ -32,13 +34,29 @@ class Config: extra = "allow" -class FeastConfigBaseModel(BaseModel): +class FeastOfflineStoreConfigBaseModel(BaseModel): """ Feast Pydantic Configuration Class """ class Config: arbitrary_types_allowed = True extra = "forbid" + @abstractmethod + def supports_data_source(self, data_source: DataSource) -> bool: + ... + + +class FeastOnlineStoreConfigBaseModel(BaseModel): + """ Feast Pydantic Configuration Class """ + + class Config: + arbitrary_types_allowed = True + extra = "forbid" + + @abstractmethod + def supports_offline_store(self, offline_store: Any) -> bool: + ... + class RegistryConfig(FeastBaseModel): """ Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry.""" diff --git a/sdk/python/feast/repo_operations.py b/sdk/python/feast/repo_operations.py index 4f2cb1981d3..61eed3b5a08 100644 --- a/sdk/python/feast/repo_operations.py +++ b/sdk/python/feast/repo_operations.py @@ -155,8 +155,11 @@ def apply_total(repo_config: RepoConfig, repo_path: Path): # Make sure the data source used by this feature view is supported by Feast for data_source in data_sources: + assert repo_config.offline_store.supports_data_source(data_source) data_source.validate() + assert repo_config.online_store.supports_offline_store(repo_config.offline_store) + update_data_sources_with_inferred_event_timestamp_col(data_sources) for view in repo.feature_views: view.infer_features_from_input_source()