diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py
index 2a2f915f335..258db418828 100644
--- a/sdk/python/feast/errors.py
+++ b/sdk/python/feast/errors.py
@@ -50,7 +50,7 @@ def __init__(self, provider_name):
 
 
 class FeastModuleImportError(Exception):
-    def __init__(self, module_name, module_type="provider"):
+    def __init__(self, module_name: str, module_type: str):
         super().__init__(f"Could not import {module_type} module '{module_name}'")
 
 
@@ -85,10 +85,11 @@ def __init__(self, online_store_class_name: str):
         )
 
 
-class FeastOnlineStoreConfigInvalidName(Exception):
-    def __init__(self, online_store_config_class_name: str):
+class FeastClassInvalidName(Exception):
+    def __init__(self, class_name: str, class_type: str):
         super().__init__(
-            f"Online Store Config Class '{online_store_config_class_name}' should end with the string `OnlineStoreConfig`.'"
+            f"Config Class '{class_name}' "
+            f"should end with the string `{class_type}`.'"
         )
 
 
diff --git a/sdk/python/feast/importer.py b/sdk/python/feast/importer.py
new file mode 100644
index 00000000000..5dcd7c71c12
--- /dev/null
+++ b/sdk/python/feast/importer.py
@@ -0,0 +1,28 @@
+import importlib
+
+from feast import errors
+
+
+def get_class_from_type(module_name: str, class_name: str, class_type: str):
+    if not class_name.endswith(class_type):
+        raise errors.FeastClassInvalidName(class_name, class_type)
+
+    # 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.FeastModuleImportError(module_name, class_type) from e
+
+    # Try getting the provider class definition
+    try:
+        _class = 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.FeastClassImportError(
+            module_name, class_name, class_type=class_type
+        ) from None
+    return _class
diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py
index f331d1b7685..a3a151ab223 100644
--- a/sdk/python/feast/infra/offline_stores/bigquery.py
+++ b/sdk/python/feast/infra/offline_stores/bigquery.py
@@ -7,6 +7,8 @@
 import pandas
 import pyarrow
 from jinja2 import BaseLoader, Environment
+from pydantic import StrictStr
+from pydantic.typing import Literal
 from tenacity import retry, stop_after_delay, wait_fixed
 
 from feast import errors
@@ -20,7 +22,7 @@
     _get_requested_feature_views_to_features_dict,
 )
 from feast.registry import Registry
-from feast.repo_config import BigQueryOfflineStoreConfig, RepoConfig
+from feast.repo_config import FeastConfigBaseModel, RepoConfig
 
 try:
     from google.api_core.exceptions import NotFound
@@ -34,6 +36,19 @@
     raise FeastExtrasDependencyImportError("gcp", str(e))
 
 
+class BigQueryOfflineStoreConfig(FeastConfigBaseModel):
+    """ Offline store config for GCP BigQuery """
+
+    type: Literal["bigquery"] = "bigquery"
+    """ Offline store type selector"""
+
+    dataset: StrictStr = "feast"
+    """ (optional) BigQuery Dataset name for temporary tables """
+
+    project_id: Optional[StrictStr] = None
+    """ (optional) GCP project name used for the BigQuery offline store """
+
+
 class BigQueryOfflineStore(OfflineStore):
     @staticmethod
     def pull_latest_from_table_or_query(
diff --git a/sdk/python/feast/infra/offline_stores/file.py b/sdk/python/feast/infra/offline_stores/file.py
index acd12ff9003..70b33726659 100644
--- a/sdk/python/feast/infra/offline_stores/file.py
+++ b/sdk/python/feast/infra/offline_stores/file.py
@@ -4,6 +4,7 @@
 import pandas as pd
 import pyarrow
 import pytz
+from pydantic.typing import Literal
 
 from feast.data_source import DataSource, FileSource
 from feast.errors import FeastJoinKeysDuringMaterialization
@@ -15,7 +16,14 @@
     _run_field_mapping,
 )
 from feast.registry import Registry
-from feast.repo_config import RepoConfig
+from feast.repo_config import FeastConfigBaseModel, RepoConfig
+
+
+class FileOfflineStoreConfig(FeastConfigBaseModel):
+    """ Offline store config for local (file-based) store """
+
+    type: Literal["file"] = "file"
+    """ Offline store type selector"""
 
 
 class FileRetrievalJob(RetrievalJob):
diff --git a/sdk/python/feast/infra/offline_stores/helpers.py b/sdk/python/feast/infra/offline_stores/helpers.py
index af1d1b92123..dff604c7ed1 100644
--- a/sdk/python/feast/infra/offline_stores/helpers.py
+++ b/sdk/python/feast/infra/offline_stores/helpers.py
@@ -1,41 +1,31 @@
-from feast.data_source import BigQuerySource, DataSource, FileSource
-from feast.errors import FeastOfflineStoreUnsupportedDataSource
+import importlib
+from typing import Any
+
+from feast import errors
 from feast.infra.offline_stores.offline_store import OfflineStore
-from feast.repo_config import (
-    BigQueryOfflineStoreConfig,
-    FileOfflineStoreConfig,
-    OfflineStoreConfig,
-)
 
 
-def get_offline_store_from_config(
-    offline_store_config: OfflineStoreConfig,
-) -> OfflineStore:
+def get_offline_store_from_config(offline_store_config: Any,) -> OfflineStore:
     """Get the offline store from offline store config"""
 
-    if isinstance(offline_store_config, FileOfflineStoreConfig):
-        from feast.infra.offline_stores.file import FileOfflineStore
-
-        return FileOfflineStore()
-    elif isinstance(offline_store_config, BigQueryOfflineStoreConfig):
-        from feast.infra.offline_stores.bigquery import BigQueryOfflineStore
-
-        return BigQueryOfflineStore()
-
-    raise ValueError(f"Unsupported offline store config '{offline_store_config}'")
-
-
-def assert_offline_store_supports_data_source(
-    offline_store_config: OfflineStoreConfig, data_source: DataSource
-):
-    if (
-        isinstance(offline_store_config, FileOfflineStoreConfig)
-        and isinstance(data_source, FileSource)
-    ) or (
-        isinstance(offline_store_config, BigQueryOfflineStoreConfig)
-        and isinstance(data_source, BigQuerySource)
-    ):
-        return
-    raise FeastOfflineStoreUnsupportedDataSource(
-        offline_store_config.type, data_source.__class__.__name__
-    )
+    module_name = offline_store_config.__module__
+    qualified_name = type(offline_store_config).__name__
+    store_class_name = qualified_name.replace("Config", "")
+    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.FeastModuleImportError(module_name, "OfflineStore") from e
+
+    # Try getting the provider class definition
+    try:
+        offline_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="OfflineStore"
+        ) from None
+    return offline_store_class()
diff --git a/sdk/python/feast/infra/online_stores/helpers.py b/sdk/python/feast/infra/online_stores/helpers.py
index 71693ce1354..9c42c5ea002 100644
--- a/sdk/python/feast/infra/online_stores/helpers.py
+++ b/sdk/python/feast/infra/online_stores/helpers.py
@@ -22,9 +22,7 @@ def get_online_store_from_config(online_store_config: Any,) -> OnlineStore:
         # 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.FeastModuleImportError(
-            module_name, module_type="OnlineStore"
-        ) from e
+        raise errors.FeastModuleImportError(module_name, "OnlineStore") from e
 
     # Try getting the provider class definition
     try:
diff --git a/sdk/python/feast/infra/online_stores/sqlite.py b/sdk/python/feast/infra/online_stores/sqlite.py
index 7385ae25a94..dbd837c5dfc 100644
--- a/sdk/python/feast/infra/online_stores/sqlite.py
+++ b/sdk/python/feast/infra/online_stores/sqlite.py
@@ -34,7 +34,9 @@
 class SqliteOnlineStoreConfig(FeastConfigBaseModel):
     """ Online store config for local (SQLite-based) store """
 
-    type: Literal["sqlite"] = "sqlite"
+    type: Literal[
+        "sqlite", "feast.infra.online_stores.sqlite.SqliteOnlineStore"
+    ] = "sqlite"
     """ Online store type selector"""
 
     path: StrictStr = "data/online.db"
@@ -51,7 +53,10 @@ class SqliteOnlineStore(OnlineStore):
 
     @staticmethod
     def _get_db_path(config: RepoConfig) -> str:
-        assert config.online_store.type == "sqlite"
+        assert (
+            config.online_store.type == "sqlite"
+            or config.online_store.type.endswith("SqliteOnlineStore")
+        )
 
         if config.repo_path and not Path(config.online_store.path).is_absolute():
             db_path = str(config.repo_path / config.online_store.path)
diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py
index 9fe95132510..905d0fd1dcc 100644
--- a/sdk/python/feast/infra/provider.py
+++ b/sdk/python/feast/infra/provider.py
@@ -1,5 +1,4 @@
 import abc
-import importlib
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
@@ -8,7 +7,7 @@
 import pyarrow
 from tqdm import tqdm
 
-from feast import errors
+from feast import errors, importer
 from feast.entity import Entity
 from feast.feature_table import FeatureTable
 from feast.feature_view import FeatureView
@@ -156,24 +155,9 @@ def get_provider(config: RepoConfig, repo_path: Path) -> Provider:
         # 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.FeastModuleImportError(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.FeastClassImportError(module_name, class_name) from None
-
-        return ProviderCls(config, repo_path)
+        cls = importer.get_class_from_type(module_name, class_name, "Provider")
+
+        return cls(config, repo_path)
 
 
 def _get_requested_feature_views_to_features_dict(
diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py
index 7c72fce9440..5587fb59053 100644
--- a/sdk/python/feast/repo_config.py
+++ b/sdk/python/feast/repo_config.py
@@ -1,24 +1,28 @@
-import importlib
 from pathlib import Path
 from typing import Any
 
 import yaml
 from pydantic import BaseModel, StrictInt, StrictStr, ValidationError, root_validator
 from pydantic.error_wrappers import ErrorWrapper
-from pydantic.typing import Dict, Literal, Optional, Union
+from pydantic.typing import Dict, Optional, Union
 
-from feast import errors
+from feast.importer import get_class_from_type
 from feast.telemetry import log_exceptions
 
-# This dict exists so that:
+# These dict exists so that:
 # - existing values for the online store type in featurestore.yaml files continue to work in a backwards compatible way
 # - first party and third party implementations can use the same class loading code path.
-ONLINE_CONFIG_CLASS_FOR_TYPE = {
+ONLINE_STORE_CLASS_FOR_TYPE = {
     "sqlite": "feast.infra.online_stores.sqlite.SqliteOnlineStore",
     "datastore": "feast.infra.online_stores.datastore.DatastoreOnlineStore",
     "redis": "feast.infra.online_stores.redis.RedisOnlineStore",
 }
 
+OFFLINE_STORE_CLASS_FOR_TYPE = {
+    "file": "feast.infra.offline_stores.file.FileOfflineStore",
+    "bigquery": "feast.infra.offline_stores.bigquery.BigQueryOfflineStore",
+}
+
 
 class FeastBaseModel(BaseModel):
     """ Feast Pydantic Configuration Class """
@@ -36,29 +40,6 @@ class Config:
         extra = "forbid"
 
 
-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 used for the BigQuery offline store """
-
-    project_id: Optional[StrictStr] = None
-    """ (optional) GCP project name used for the BigQuery offline store """
-
-
-OfflineStoreConfig = Union[FileOfflineStoreConfig, BigQueryOfflineStoreConfig]
-
-
 class RegistryConfig(FeastBaseModel):
     """ Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry."""
 
@@ -90,7 +71,7 @@ class RepoConfig(FeastBaseModel):
     online_store: Any
     """ OnlineStoreConfig: Online store configuration (optional depending on provider) """
 
-    offline_store: OfflineStoreConfig = FileOfflineStoreConfig()
+    offline_store: Any
     """ OfflineStoreConfig: Offline store configuration (optional depending on provider) """
 
     repo_path: Optional[Path] = None
@@ -101,6 +82,10 @@ def __init__(self, **data: Any):
             self.online_store = get_online_config_from_type(self.online_store["type"])(
                 **self.online_store
             )
+        if isinstance(self.offline_store, Dict):
+            self.offline_store = get_offline_config_from_type(
+                self.offline_store["type"]
+            )(**self.offline_store)
 
     def get_registry_config(self):
         if isinstance(self.registry, str):
@@ -172,22 +157,13 @@ def _validate_offline_store_config(cls, values):
 
         offline_store_type = values["offline_store"]["type"]
 
-        # Make sure the user hasn't provided the wrong type
-        assert offline_store_type in ["file", "bigquery"]
-
         # Validate the dict to ensure one of the union types match
         try:
-            if offline_store_type == "file":
-                FileOfflineStoreConfig(**values["offline_store"])
-            elif offline_store_type == "bigquery":
-                BigQueryOfflineStoreConfig(**values["offline_store"])
-            else:
-                raise ValidationError(
-                    f"Invalid offline store type {offline_store_type}"
-                )
+            offline_config_class = get_offline_config_from_type(offline_store_type)
+            offline_config_class(**values["offline_store"])
         except ValidationError as e:
             raise ValidationError(
-                [ErrorWrapper(e, loc="offline_store")], model=FileOfflineStoreConfig,
+                [ErrorWrapper(e, loc="offline_store")], model=RepoConfig,
             )
 
         return values
@@ -209,35 +185,25 @@ def __repr__(self) -> str:
 
 
 def get_online_config_from_type(online_store_type: str):
-    if online_store_type in ONLINE_CONFIG_CLASS_FOR_TYPE:
-        online_store_type = ONLINE_CONFIG_CLASS_FOR_TYPE[online_store_type]
-    module_name, class_name = online_store_type.rsplit(".", 1)
-
-    if not class_name.endswith("OnlineStore"):
-        raise errors.FeastOnlineStoreConfigInvalidName(class_name)
-    config_class_name = f"{class_name}Config"
-
-    # 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.FeastModuleImportError(
-            module_name, module_type="OnlineStore"
-        ) from e
-
-    # Try getting the provider class definition
-    try:
-        online_store_config_class = getattr(module, config_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, config_class_name, class_type="OnlineStoreConfig"
-        ) from None
-    return online_store_config_class
+    if online_store_type in ONLINE_STORE_CLASS_FOR_TYPE:
+        online_store_type = ONLINE_STORE_CLASS_FOR_TYPE[online_store_type]
+    else:
+        assert online_store_type.endswith("OnlineStore")
+    module_name, online_store_class_type = online_store_type.rsplit(".", 1)
+    config_class_name = f"{online_store_class_type}Config"
+
+    return get_class_from_type(module_name, config_class_name, config_class_name)
+
+
+def get_offline_config_from_type(offline_store_type: str):
+    if offline_store_type in OFFLINE_STORE_CLASS_FOR_TYPE:
+        offline_store_type = OFFLINE_STORE_CLASS_FOR_TYPE[offline_store_type]
+    else:
+        assert offline_store_type.endswith("OfflineStore")
+    module_name, offline_store_class_type = offline_store_type.rsplit(".", 1)
+    config_class_name = f"{offline_store_class_type}Config"
+
+    return get_class_from_type(module_name, config_class_name, config_class_name)
 
 
 def load_repo_config(repo_path: Path) -> RepoConfig:
diff --git a/sdk/python/feast/repo_operations.py b/sdk/python/feast/repo_operations.py
index 59ff0c60bf7..f4a44a74559 100644
--- a/sdk/python/feast/repo_operations.py
+++ b/sdk/python/feast/repo_operations.py
@@ -17,7 +17,6 @@
     infer_entity_value_type_from_feature_views,
     update_data_sources_with_inferred_event_timestamp_col,
 )
-from feast.infra.offline_stores.helpers import assert_offline_store_supports_data_source
 from feast.infra.provider import get_provider
 from feast.names import adjectives, animals
 from feast.registry import Registry
@@ -156,9 +155,6 @@ 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_offline_store_supports_data_source(
-            repo_config.offline_store, data_source
-        )
         data_source.validate()
 
     update_data_sources_with_inferred_event_timestamp_col(data_sources)
diff --git a/sdk/python/tests/test_cli_local.py b/sdk/python/tests/test_cli_local.py
index 288a2462452..43998190737 100644
--- a/sdk/python/tests/test_cli_local.py
+++ b/sdk/python/tests/test_cli_local.py
@@ -164,18 +164,18 @@ def test_3rd_party_providers() -> None:
         assertpy.assert_that(return_code).is_equal_to(1)
         assertpy.assert_that(output).contains(b"Provider 'feast123' is not implemented")
     # Check with incorrect third-party provider name (with dots)
-    with setup_third_party_provider_repo("feast_foo.provider") as repo_path:
+    with setup_third_party_provider_repo("feast_foo.Provider") as repo_path:
         return_code, output = runner.run_with_output(["apply"], cwd=repo_path)
         assertpy.assert_that(return_code).is_equal_to(1)
         assertpy.assert_that(output).contains(
-            b"Could not import provider module 'feast_foo'"
+            b"Could not import Provider module 'feast_foo'"
         )
     # Check with incorrect third-party provider name (with dots)
     with setup_third_party_provider_repo("foo.FooProvider") as repo_path:
         return_code, output = runner.run_with_output(["apply"], cwd=repo_path)
         assertpy.assert_that(return_code).is_equal_to(1)
         assertpy.assert_that(output).contains(
-            b"Could not import provider 'FooProvider' from module 'foo'"
+            b"Could not import Provider 'FooProvider' from module 'foo'"
         )
     # Check with correct third-party provider name
     with setup_third_party_provider_repo("foo.provider.FooProvider") as repo_path:
diff --git a/sdk/python/tests/test_historical_retrieval.py b/sdk/python/tests/test_historical_retrieval.py
index 656dd589e9c..ccb5030a015 100644
--- a/sdk/python/tests/test_historical_retrieval.py
+++ b/sdk/python/tests/test_historical_retrieval.py
@@ -20,9 +20,9 @@
 from feast.feature import Feature
 from feast.feature_store import FeatureStore
 from feast.feature_view import FeatureView
+from feast.infra.offline_stores.bigquery import BigQueryOfflineStoreConfig
 from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig
 from feast.infra.provider import DEFAULT_ENTITY_DF_EVENT_TIMESTAMP_COL
-from feast.repo_config import BigQueryOfflineStoreConfig
 from feast.value_type import ValueType
 
 np.random.seed(0)
diff --git a/sdk/python/tests/test_repo_config.py b/sdk/python/tests/test_repo_config.py
index 19c8ee4dcc0..b6a6a330119 100644
--- a/sdk/python/tests/test_repo_config.py
+++ b/sdk/python/tests/test_repo_config.py
@@ -26,6 +26,7 @@ def _test_config(config_text, expect_error: Optional[str]):
         if expect_error is not None:
             assert expect_error in str(error)
         else:
+            print(f"error: {error}")
             assert error is None
 
 
@@ -42,6 +43,21 @@ def test_local_config():
     )
 
 
+def test_local_config_with_full_online_class():
+    _test_config(
+        dedent(
+            """
+        project: foo
+        registry: "registry.db"
+        provider: local
+        online_store:
+            type: feast.infra.online_stores.sqlite.SqliteOnlineStore
+        """
+        ),
+        expect_error=None,
+    )
+
+
 def test_gcp_config():
     _test_config(
         dedent(