merge latest from master

Signed-off-by: lblokhin <lenin133@yandex.ru>
feast-dev · feast-ci-bot · Jul 3, 2021 · Apr 27, 2021 · Apr 27, 2021 · May 4, 2021
commit 24c44ee77fbe466e49391bb1c8892fe8902b7587
@@ -15,13 +15,15 @@
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
 
 import mmh3
+from pydantic import PositiveInt, StrictStr
+from pydantic.typing import Literal
 
 from feast import Entity, FeatureTable, FeatureView, utils
 from feast.infra.key_encoding_utils import serialize_entity_key
 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 DynamoDbOnlineStoreConfig, RepoConfig
+from feast.repo_config import FeastConfigBaseModel, RepoConfig
 
 try:
     import boto3
@@ -32,7 +34,23 @@
     raise FeastExtrasDependencyImportError("aws", str(e))
 
 
-class DynamodbOnlineStore(OnlineStore):
+class DynamoDbOnlineStoreConfig(FeastConfigBaseModel):
+    """Online store config for DynamoDB store"""
+
+    type: Literal["dynamodb"] = "dynamodb"
+    """Online store type selector"""
+
+    rcu: Optional[PositiveInt] = 5
+    """ Read capacity unit """
+
+    wcu: Optional[PositiveInt] = 5
+    """ Write capacity unit """
+
+    region_name: Optional[StrictStr] = None
+    """ AWS Region Name """
+
+
+class DynamoDbOnlineStore(OnlineStore):
     def _initialize_dynamodb(self, online_config: DynamoDbOnlineStoreConfig):
         return boto3.resource("dynamodb", region_name=online_config.region_name)
 
@@ -90,8 +108,11 @@ def teardown(
         dynamodb = self._initialize_dynamodb(online_config)
 
         for table_name in tables:
-            table = dynamodb.Table(table_name)
-            table.delete()
+            try:
+                table = dynamodb.Table(table_name)
+                table.delete()
+            except Exception as e:
+                print(str(e))
 
     def online_write_batch(
         self,

@@ -8,13 +8,6 @@
 from feast.infra.online_stores.online_store import OnlineStore
 from feast.protos.feast.storage.Redis_pb2 import RedisKeyV2 as RedisKeyProto
 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
-from feast.repo_config import (
-    DatastoreOnlineStoreConfig,
-    DynamoDbOnlineStoreConfig,
-    OnlineStoreConfig,
-    RedisOnlineStoreConfig,
-    SqliteOnlineStoreConfig,
-)
 
 
 def get_online_store_from_config(online_store_config: Any,) -> OnlineStore:
@@ -31,47 +24,16 @@ def get_online_store_from_config(online_store_config: Any,) -> OnlineStore:
         # So we should include the original error as well in the stack trace.
         raise errors.FeastModuleImportError(module_name, "OnlineStore") from e
 
-        return SqliteOnlineStore()
-    elif isinstance(online_store_config, DatastoreOnlineStoreConfig):
-        from feast.infra.online_stores.datastore import DatastoreOnlineStore
-
-        return DatastoreOnlineStore()
-    elif isinstance(online_store_config, RedisOnlineStoreConfig):
-        from feast.infra.online_stores.redis import RedisOnlineStore
-
-        return RedisOnlineStore()
-    elif isinstance(online_store_config, DynamoDbOnlineStoreConfig):
-        from feast.infra.online_stores.dynamodb import DynamodbOnlineStore
-
-        return DynamodbOnlineStore()
-    raise ValueError(f"Unsupported offline store config '{online_store_config}'")
-
-
-SUPPORTED_SOURCES: Dict[Any, Set[Any]] = {
-    SqliteOnlineStoreConfig: {FileSource},
-    DatastoreOnlineStoreConfig: {BigQuerySource},
-    RedisOnlineStoreConfig: {FileSource, BigQuerySource},
-    DynamoDbOnlineStoreConfig: {FileSource, BigQuerySource},
-}
-
-
-def assert_online_store_supports_data_source(
-    online_store_config: OnlineStoreConfig, data_source: DataSource
-):
-    supported_sources: Set[Any] = SUPPORTED_SOURCES.get(
-        online_store_config.__class__, set()
-    )
-    # This is needed because checking for `in` with Union types breaks mypy.
-    # https://github.com/python/mypy/issues/4954
-    # We can replace this with `data_source.__class__ in SUPPORTED_SOURCES[online_store_config.__class__]`
-    # Once ^ is resolved.
-    if supported_sources:
-        for source in supported_sources:
-            if source == data_source.__class__:
-                return
-    raise FeastOnlineStoreUnsupportedDataSource(
-        online_store_config.type, data_source.__class__.__name__
-    )
+    # Try getting the provider class definition
+    try:
+        online_store_class = getattr(module, store_class_name)
+    except AttributeError:
+        # This can only be one type of error, when class_name attribute does not exist in the module
+        # So we don't have to include the original exception here
+        raise errors.FeastClassImportError(
+            module_name, store_class_name, class_type="OnlineStore"
+        ) from None
+    return online_store_class()
 
 
 def _redis_key(project: str, entity_key: EntityKeyProto):

@@ -16,6 +16,7 @@
     "sqlite": "feast.infra.online_stores.sqlite.SqliteOnlineStore",
     "datastore": "feast.infra.online_stores.datastore.DatastoreOnlineStore",
     "redis": "feast.infra.online_stores.redis.RedisOnlineStore",
+    "dynamodb": "feast.infra.online_stores.dynamodb.DynamoDbOnlineStore",
 }
 
 OFFLINE_STORE_CLASS_FOR_TYPE = {
@@ -40,98 +41,6 @@ class Config:
         extra = "forbid"
 
 
-class SqliteOnlineStoreConfig(FeastBaseModel):
-    """ Online store config for local (SQLite-based) store """
-
-    type: Literal["sqlite"] = "sqlite"
-    """ Online store type selector"""
-
-    path: StrictStr = "data/online.db"
-    """ (optional) Path to sqlite db """
-
-
-class DatastoreOnlineStoreConfig(FeastBaseModel):
-    """ Online store config for GCP Datastore """
-
-    type: Literal["datastore"] = "datastore"
-    """ Online store type selector"""
-
-    project_id: Optional[StrictStr] = None
-    """ (optional) GCP Project Id """
-
-    namespace: Optional[StrictStr] = None
-    """ (optional) Datastore namespace """
-
-    write_concurrency: Optional[PositiveInt] = 40
-    """ (optional) Amount of threads to use when writing batches of feature rows into Datastore"""
-
-    write_batch_size: Optional[PositiveInt] = 50
-    """ (optional) Amount of feature rows per batch being written into Datastore"""
-
-
-class RedisType(str, Enum):
-    redis = "redis"
-    redis_cluster = "redis_cluster"
-
-
-class RedisOnlineStoreConfig(FeastBaseModel):
-    """Online store config for Redis store"""
-
-    type: Literal["redis"] = "redis"
-    """Online store type selector"""
-
-    redis_type: RedisType = RedisType.redis
-    """Redis type: redis or redis_cluster"""
-
-    connection_string: StrictStr = "localhost:6379"
-    """Connection string containing the host, port, and configuration parameters for Redis
-     format: host:port,parameter1,parameter2 eg. redis:6379,db=0 """
-
-
-class DynamoDbOnlineStoreConfig(FeastBaseModel):
-    """Online store config for DynamoDB store"""
-
-    type: Literal["dynamodb"] = "dynamodb"
-    """Online store type selector"""
-
-    rcu: Optional[PositiveInt] = 5
-    """ Read capacity unit """
-
-    wcu: Optional[PositiveInt] = 5
-    """ Write capacity unit """
-
-    region_name: Optional[StrictStr] = None
-    """ AWS Region Name """
-
-
-OnlineStoreConfig = Union[
-    DatastoreOnlineStoreConfig,
-    SqliteOnlineStoreConfig,
-    RedisOnlineStoreConfig,
-    DynamoDbOnlineStoreConfig,
-]
-
-
-class FileOfflineStoreConfig(FeastBaseModel):
-    """ Offline store config for local (file-based) store """
-
-    type: Literal["file"] = "file"
-    """ Offline store type selector"""
-
-
-class BigQueryOfflineStoreConfig(FeastBaseModel):
-    """ Offline store config for GCP BigQuery """
-
-    type: Literal["bigquery"] = "bigquery"
-    """ Offline store type selector"""
-
-    dataset: StrictStr = "feast"
-    """ (optional) BigQuery Dataset name for temporary tables """
-
-
-OfflineStoreConfig = Union[FileOfflineStoreConfig, BigQueryOfflineStoreConfig]
-
-
 class RegistryConfig(FeastBaseModel):
     """ Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry."""
 
@@ -158,7 +67,7 @@ class RepoConfig(FeastBaseModel):
     """
 
     provider: StrictStr
-    """ str: local or gcp or redis or aws """
+    """ str: local or gcp or aws """
 
     online_store: Any
     """ OnlineStoreConfig: Online store configuration (optional depending on provider) """
@@ -218,21 +127,10 @@ def _validate_online_store_config(cls, values):
 
         online_store_type = values["online_store"]["type"]
 
-        # Make sure the user hasn't provided the wrong type
-        assert online_store_type in ["datastore", "sqlite", "redis", "dynamodb"]
-
         # Validate the dict to ensure one of the union types match
         try:
-            if online_store_type == "sqlite":
-                SqliteOnlineStoreConfig(**values["online_store"])
-            elif online_store_type == "datastore":
-                DatastoreOnlineStoreConfig(**values["online_store"])
-            elif online_store_type == "redis":
-                RedisOnlineStoreConfig(**values["online_store"])
-            elif online_store_type == "dynamodb":
-                DynamoDbOnlineStoreConfig(**values["online_store"])
-            else:
-                raise ValueError(f"Invalid online store type {online_store_type}")
+            online_config_class = get_online_config_from_type(online_store_type)
+            online_config_class(**values["online_store"])
         except ValidationError as e:
             raise ValidationError(
                 [ErrorWrapper(e, loc="online_store")], model=RepoConfig,
@@ -257,10 +155,10 @@ def _validate_offline_store_config(cls, values):
         if "type" not in values["offline_store"]:
             if values["provider"] == "local":
                 values["offline_store"]["type"] = "file"
-            elif values["provider"] == "aws":
-                values["offline_store"]["type"] = "file"
             elif values["provider"] == "gcp":
                 values["offline_store"]["type"] = "bigquery"
+            elif values["provider"] == "aws":
+                values["offline_store"]["type"] = "file"
 
         offline_store_type = values["offline_store"]["type"]