correct redis provider

Signed-off-by: qooba <dev@qooba.net>
feast-dev · feast-ci-bot · Jun 9, 2021 · Apr 25, 2021 · Apr 25, 2021 · Apr 26, 2021
commit 435c8566f03b37b178f1fc25722eb6858260143d
diff --git a/.github/workflows/pr_integration_tests.yml b/.github/workflows/pr_integration_tests.yml
@@ -18,6 +18,9 @@ jobs:
       matrix:
         python-version: [ 3.7, 3.8, 3.9 ]
         os: [ ubuntu-latest ]
+    env:
+      OS: ${{ matrix.os }}
+      PYTHON: ${{ matrix.python-version }}
     services:
       redis:
         image: redis
@@ -28,9 +31,6 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
-    env:
-      OS: ${{ matrix.os }}
-      PYTHON: ${{ matrix.python-version }}
     steps:
       - uses: actions/checkout@v2
         with:
@@ -76,4 +76,3 @@ jobs:
           env_vars: OS,PYTHON
           fail_ci_if_error: true
           verbose: true
-
@@ -146,7 +146,7 @@ def get_provider(config: RepoConfig, repo_path: Path) -> Provider:
 
             return GcpProvider(config)
         elif config.provider == "redis":
-            from feast.infra.redis_provider import RedisProvider
+            from feast.infra.redis import RedisProvider
 
             return RedisProvider(config)
         elif config.provider == "local":

@@ -13,7 +13,7 @@
 from feast import FeatureTable, utils
 from feast.entity import Entity
 from feast.feature_view import FeatureView
-from feast.infra.offline_stores.helpers import get_offline_store_from_sources
+from feast.infra.offline_stores.helpers import get_offline_store_from_config
 from feast.infra.provider import (
     Provider,
     RetrievalJob,
@@ -36,13 +36,11 @@ class RedisProvider(Provider):
 
     def __init__(self, config: RepoConfig):
         assert isinstance(config.online_store, RedisOnlineStoreConfig)
-        if config and config.online_store:
-            if config.online_store.redis_type:
-                self._redis_type = config.online_store.redis_type
-            if config.online_store.redis_connection_string:
-                self._redis_connection_string = (
-                    config.online_store.redis_connection_string
-                )
+        if config.online_store.redis_type:
+            self._redis_type = config.online_store.redis_type
+        if config.online_store.redis_connection_string:
+            self._redis_connection_string = config.online_store.redis_connection_string
+        self.offline_store = get_offline_store_from_config(config.offline_store)
 
     def update_infra(
         self,
@@ -173,8 +171,7 @@ def materialize_single_feature_view(
         start_date = utils.make_tzaware(start_date)
         end_date = utils.make_tzaware(end_date)
 
-        offline_store = get_offline_store_from_sources([feature_view.input])
-        table = offline_store.pull_latest_from_table_or_query(
+        table = self.offline_store.pull_latest_from_table_or_query(
             data_source=feature_view.input,
             join_key_columns=join_key_columns,
             feature_name_columns=feature_name_columns,
@@ -239,19 +236,16 @@ def _get_client(self):
             kwargs["port"] = startup_nodes[0]["port"]
             return Redis(**kwargs)
 
-    @staticmethod
     def get_historical_features(
+        self,
         config: RepoConfig,
         feature_views: List[FeatureView],
         feature_refs: List[str],
         entity_df: Union[pd.DataFrame, str],
         registry: Registry,
         project: str,
     ) -> RetrievalJob:
-        offline_store = get_offline_store_from_sources(
-            [feature_view.input for feature_view in feature_views]
-        )
-        return offline_store.get_historical_features(
+        return self.offline_store.get_historical_features(
             config=config,
             feature_views=feature_views,
             feature_refs=feature_refs,

@@ -166,16 +166,16 @@ def _validate_online_store_config(cls, values):
                 values["online_store"]["type"] = "datastore"
             elif values["provider"] == "redis":
                 values["online_store"]["type"] = "redis"
-                values["online_store"]["redis_connection_string"] = os.environ.get(
-                    "REDIS_CONNECTION_STRING"
-                )
-
 
+        if values["online_store"]["type"] == "redis":
+            values["online_store"]["redis_connection_string"] = os.environ.get(
+                "REDIS_CONNECTION_STRING"
+            )
 
         online_store_type = values["online_store"]["type"]
 
         # Make sure the user hasn't provided the wrong type
-        assert online_store_type in ["datastore", "sqlite"]
+        assert online_store_type in ["datastore", "sqlite", "redis"]
 
         # Validate the dict to ensure one of the union types match
         try:
@@ -209,7 +209,7 @@ def _validate_offline_store_config(cls, values):
 
         # Set the default type
         if "type" not in values["offline_store"]:
-            if values["provider"] == "local":
+            if values["provider"] == "local" or values["provider"] == "redis":
                 values["offline_store"]["type"] = "file"
             elif values["provider"] == "gcp":
                 values["offline_store"]["type"] = "bigquery"