Add support for third party providers

Signed-off-by: Tsotne Tabidze <tsotne@tecton.ai>

sdk/python/feast/errors.py

@@ -30,3 +30,24 @@ def __init__(self, name, project=None):
 
 class FeastProviderLoginError(Exception):
     """Error class that indicates a user has not authenticated with their provider."""
+    def __init__(self, project, name):
+        super().__init__(f"Feature table {name} does not exist in project {project}")
+
+
+class ProviderNameParsingException(Exception):
+    def __init__(self, provider_name):
+        super().__init__(
+            f"Could not parse provider name '{provider_name}' into module and class names"
+        )
+
+
+class ProviderModuleImportError(Exception):
+    def __init__(self, module_name):
+        super().__init__(f"Could not import provider module '{module_name}'")
+
+
+class ProviderClassImportError(Exception):
+    def __init__(self, module_name, class_name):
+        super().__init__(
+            f"Could not import provider '{class_name}' from module '{module_name}'"
+        )

sdk/python/feast/infra/provider.py

@@ -1,11 +1,13 @@
 import abc
+import importlib
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
 
 import pandas
 import pyarrow
 
+from feast import errors
 from feast.entity import Entity
 from feast.feature_table import FeatureTable
 from feast.feature_view import FeatureView
@@ -144,7 +146,30 @@ def get_provider(config: RepoConfig, repo_path: Path) -> Provider:
 
         return LocalProvider(config, repo_path)
     else:
-        raise ValueError(config)
+        if "." not in config.provider:
+            raise errors.ProviderNameParsingException(config.provider)
+        # Split provider into module and class names by finding the right-most dot.
+        # For example, provider 'foo.bar.MyProvider' will be parsed into 'foo.bar' and 'MyProvider'
+        module_name, class_name = config.provider.rsplit(".", 1)
+
+        # Try importing the module that contains the custom provider
+        try:
+            module = importlib.import_module(module_name)
+        except Exception as e:
+            # The original exception can be anything - either module not found,
+            # or any other kind of error happening during the module import time.
+            # So we should include the original error as well in the stack trace.
+            raise errors.ProviderModuleImportError(module_name) from e
+
+        # Try getting the provider class definition
+        try:
+            ProviderCls = getattr(module, 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.ProviderClassImportError(module_name, class_name) from None
+
+        return ProviderCls(config)
 
 
 def _get_requested_feature_views_to_features_dict(