Repository: great-expectations/great_expectations
Pull request #7078 (2023-02-08 13:12:59)

[MAINTENANCE] Move experimental `base_directory` from `_FilesystemDataAsset` to `PandasDatasource` and `SparkDatasource`
Changes proposed in this pull request:
- GREAT-1678
- Move `base_directory` from `_FilesystemDataAsset` to `PandasDatasource` and `SparkDatasource`
- Rename `add_pandas` factory function to `add_pandas_filesystem`
- Introduce abstract classes `_PandasDatasource` and `_SparkDatasource`
- Introduce `_DatasourceT` and push methods referencing `self.datasource` down into subclasses for more specific typing information
- Use `TypeVars` and `alias` to kill type ignores at API-level
- Change default serialization method for `by_alias` to `True`
- Update `DataAsset` schemas

### Definition of Done
- [X] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [X] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_checklist) of my own code
- [X] I have commented my code, particularly in hard-to-understand areas
- [X] I have made corresponding changes to the documentation
- [X] I have added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [X] I have run any local integration tests and made sure that nothing is broken.


netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 9daab8cc8baeb6eef63d73a68f491f11338daa71 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63e39fdd25266c0008bdeb1a |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=7078&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations">Click on the image</a> for a new way to code review</b>
</summary>

  - Make big changes easier — review code in small groups of related files
  - Know where to start — see the whole change at a glance
  - Take a code tour — explore the change with an interactive tour
  - Make comments and review — all fully sync’ed with github

    [Try it now!](https://app.codesee.io/r/reviews?pr=7078&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)
</details>

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/great-expectations/great_expectations/7078/b948bf8b/9daab8cc8baeb6eef63d73a68f491f11338daa71.svg)](https://app.codesee.io/r/reviews?pr=7078&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)

Patch:
diff --git a/great_expectations/experimental/datasources/__init__.py b/great_expectations/experimental/datasources/__init__.py
index 0f460b3af..e1bb7f616 100644
--- a/great_expectations/experimental/datasources/__init__.py
+++ b/great_expectations/experimental/datasources/__init__.py
@@ -1,7 +1,7 @@
 import pathlib
 
 from great_expectations.experimental.datasources.pandas_datasource import (
-    PandasDatasource,
+    PandasFilesystemDatasource,
 )
 from great_expectations.experimental.datasources.postgres_datasource import (
     PostgresDatasource,
diff --git a/great_expectations/experimental/datasources/dynamic_pandas.py b/great_expectations/experimental/datasources/dynamic_pandas.py
index b1b6a9b56..8b9fbb797 100644
--- a/great_expectations/experimental/datasources/dynamic_pandas.py
+++ b/great_expectations/experimental/datasources/dynamic_pandas.py
@@ -139,9 +139,9 @@ def _replace_builtins(input_: str | type) -> str | type:
 
 FIELD_SUBSTITUTIONS: Final[Dict[str, Dict[str, _FieldSpec]]] = {
     # CSVAsset
-    "filepath_or_buffer": {"base_directory": _FieldSpec(pathlib.Path, ...)},
+    "filepath_or_buffer": {"_base_directory": _FieldSpec(pathlib.Path, ...)},
     # JSONAsset
-    "path_or_buf": {"base_directory": _FieldSpec(pathlib.Path, ...)},
+    "path_or_buf": {"_base_directory": _FieldSpec(pathlib.Path, ...)},
     # SQLTable
     "schema": {
         "schema_name": _FieldSpec(
@@ -155,7 +155,7 @@ FIELD_SUBSTITUTIONS: Final[Dict[str, Dict[str, _FieldSpec]]] = {
         )
     },
     # misc
-    "filepath": {"base_directory": _FieldSpec(pathlib.Path, ...)},
+    "filepath": {"_base_directory": _FieldSpec(pathlib.Path, ...)},
     "dtype": {"dtype": _FieldSpec(Optional[dict], None)},  # type: ignore[arg-type]
     "dialect": {"dialect": _FieldSpec(Optional[str], None)},  # type: ignore[arg-type]
     "usecols": {"usecols": _FieldSpec(Union[int, str, Sequence[int], None], None)},  # type: ignore[arg-type]
diff --git a/great_expectations/experimental/datasources/file_path_data_asset.py b/great_expectations/experimental/datasources/file_path_data_asset.py
index 2b7206cc1..299727719 100644
--- a/great_expectations/experimental/datasources/file_path_data_asset.py
+++ b/great_expectations/experimental/datasources/file_path_data_asset.py
@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import dataclasses
 import logging
-from typing import ClassVar, Dict, List, Optional, Pattern, Set
+import pathlib
+from pprint import pformat as pf
+from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional, Pattern, Set, Union
 
 import pydantic
 
-import great_expectations.exceptions as ge_exceptions
+import great_expectations.exceptions as gx_exceptions
 from great_expectations.experimental.datasources.data_asset.data_connector.regex_parser import (
     RegExParser,
 )
@@ -16,6 +19,12 @@ from great_expectations.experimental.datasources.interfaces import (
     DataAsset,
 )
 
+if TYPE_CHECKING:
+    from great_expectations.experimental.datasources import (
+        PandasFilesystemDatasource,
+        SparkDatasource,
+    )
+
 logger = logging.getLogger(__name__)
 
 
@@ -31,6 +40,10 @@ class _FilePathDataAsset(DataAsset):
     regex: Pattern
 
     # Internal attributes
+    _datasource: Union[
+        "PandasFilesystemDatasource", "SparkDatasource"
+    ] = pydantic.PrivateAttr()
+
     _unnamed_regex_param_prefix: str = pydantic.PrivateAttr(
         default="batch_request_param_"
     )
@@ -65,6 +78,10 @@ class _FilePathDataAsset(DataAsset):
         )
         self._all_group_names = self._regex_parser.get_all_group_names()
 
+    @property
+    def _base_directory(self) -> pathlib.Path:
+        return self._datasource.base_directory
+
     def batch_request_options_template(
         self,
     ) -> BatchRequestOptions:
@@ -80,12 +97,46 @@ class _FilePathDataAsset(DataAsset):
                     option in self._all_group_name_to_group_index_mapping
                     and not isinstance(value, str)
                 ):
-                    raise ge_exceptions.InvalidBatchRequestError(
+                    raise gx_exceptions.InvalidBatchRequestError(
                         f"All regex matching options must be strings. The value of '{option}' is "
                         f"not a string: {value}"
                     )
 
-        return super().build_batch_request(options)
+        if options is not None and not self._valid_batch_request_options(options):
+            allowed_keys = set(self.batch_request_options_template().keys())
+            actual_keys = set(options.keys())
+            raise gx_exceptions.InvalidBatchRequestError(
+                "Batch request options should only contain keys from the following set:\n"
+                f"{allowed_keys}\nbut your specified keys contain\n"
+                f"{actual_keys.difference(allowed_keys)}\nwhich is not valid.\n"
+            )
+        return BatchRequest(
+            datasource_name=self.datasource.name,
+            data_asset_name=self.name,
+            options=options or {},
+        )
+
+    def _validate_batch_request(self, batch_request: BatchRequest) -> None:
+        """Validates the batch_request has the correct form.
+
+        Args:
+            batch_request: A batch request object to be validated.
+        """
+        if not (
+            batch_request.datasource_name == self.datasource.name
+            and batch_request.data_asset_name == self.name
+            and self._valid_batch_request_options(batch_request.options)
+        ):
+            expect_batch_request_form = BatchRequest(
+                datasource_name=self.datasource.name,
+                data_asset_name=self.name,
+                options=self.batch_request_options_template(),
+            )
+            raise gx_exceptions.InvalidBatchRequestError(
+                "BatchRequest should have form:\n"
+                f"{pf(dataclasses.asdict(expect_batch_request_form))}\n"
+                f"but actually has form:\n{pf(dataclasses.asdict(batch_request))}\n"
+            )
 
     def test_connection(self) -> None:
         """Test the connection for the DataAsset.
diff --git a/great_expectations/experimental/datasources/filesystem_data_asset.py b/great_expectations/experimental/datasources/filesystem_data_asset.py
index 9d53520b4..e12b57486 100644
--- a/great_expectations/experimental/datasources/filesystem_data_asset.py
+++ b/great_expectations/experimental/datasources/filesystem_data_asset.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import copy
 import logging
 import pathlib
-from typing import TYPE_CHECKING, ClassVar, List, Optional, Set, Tuple
+from typing import TYPE_CHECKING, List, Optional, Set, Tuple
 
 import great_expectations.exceptions as ge_exceptions
 from great_expectations.core.batch_spec import PathBatchSpec
@@ -22,20 +22,15 @@ if TYPE_CHECKING:
         PandasExecutionEngine,
         SparkDFExecutionEngine,
     )
+    from great_expectations.experimental.datasources import (
+        PandasFilesystemDatasource,
+        SparkDatasource,
+    )
 
 logger = logging.getLogger(__name__)
 
 
 class _FilesystemDataAsset(_FilePathDataAsset):
-    _EXCLUDE_FROM_READER_OPTIONS: ClassVar[
-        Set[str]
-    ] = _FilePathDataAsset._EXCLUDE_FROM_READER_OPTIONS | {
-        "base_directory",
-    }
-
-    # Filesystem specific attributes
-    base_directory: pathlib.Path
-
     def _get_reader_method(self) -> str:
         raise NotImplementedError(
             """One needs to explicitly provide "reader_method" for Filesystem DataAsset extensions as temporary \
@@ -54,20 +49,17 @@ to use as its "include" directive for Filesystem style DataAsset processing."""
         Raises:
             TestConnectionError: If the connection test fails.
         """
-        if not self.base_directory.exists():
-            raise TestConnectionError(
-                f"Path: {self.base_directory.resolve()} does not exist."
-            )
+        datasource: PandasFilesystemDatasource | SparkDatasource = self.datasource
 
         success = False
-        for filepath in self.base_directory.iterdir():
+        for filepath in datasource.base_directory.iterdir():
             if self.regex.match(filepath.name):
                 # if one file in the path matches the regex, we consider this asset valid
                 success = True
                 break
         if not success:
             raise TestConnectionError(
-                f"No file at path: {self.base_directory} matched the regex: {self.regex.pattern}"
+                f"No file at path: {datasource.base_directory.resolve()} matched the regex: {self.regex.pattern}"
             )
 
     def _fully_specified_batch_requests_with_path(
@@ -84,11 +76,12 @@ to use as its "include" directive for Filesystem style DataAsset processing."""
             This list will be empty if no files exist on disk that correspond to the input
             batch request.
         """
-        batch_requests_with_path: List[Tuple[BatchRequest, pathlib.Path]] = []
+        datasource: PandasFilesystemDatasource | SparkDatasource = self.datasource
 
-        all_files: List[pathlib.Path] = list(
-            pathlib.Path(self.base_directory).iterdir()
-        )
+        base_directory: pathlib.Path = datasource.base_directory
+        all_files: List[pathlib.Path] = list(pathlib.Path(base_directory).iterdir())
+
+        batch_requests_with_path: List[Tuple[BatchRequest, pathlib.Path]] = []
 
         file_name: pathlib.Path
         for file_name in all_files:
@@ -112,14 +105,14 @@ to use as its "include" directive for Filesystem style DataAsset processing."""
                     batch_requests_with_path.append(
                         (
                             BatchRequest(
-                                datasource_name=self.datasource.name,
+                                datasource_name=datasource.name,
                                 data_asset_name=self.name,
                                 options=match_options,
                             ),
-                            self.base_directory / file_name,
+                            base_directory / file_name,
                         )
                     )
-                    logger.debug(f"Matching path: {self.base_directory / file_name}")
+                    logger.debug(f"Matching path: {base_directory / file_name}")
         if not batch_requests_with_path:
             logger.warning(
                 f"Batch request {batch_request} corresponds to no data files."
diff --git a/great_expectations/experimental/datasources/interfaces.py b/great_expectations/experimental/datasources/interfaces.py
index 111bde863..a94c1a6ea 100644
--- a/great_expectations/experimental/datasources/interfaces.py
+++ b/great_expectations/experimental/datasources/interfaces.py
@@ -24,7 +24,6 @@ from pydantic import Field, StrictBool, StrictInt, root_validator, validate_argu
 from pydantic import dataclasses as pydantic_dc
 from typing_extensions import TypeAlias, TypeGuard
 
-import great_expectations.exceptions as ge_exceptions
 from great_expectations.core.id_dict import BatchSpec  # noqa: TCH001
 from great_expectations.experimental.datasources.constants import _FIELDS_ALWAYS_SET
 from great_expectations.experimental.datasources.experimental_base_model import (
@@ -39,7 +38,11 @@ logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
     # TODO: When the Batch class is moved out of the experimental directory, we should try to import the annotations
     #       from core.batch so we no longer need to call Batch.update_forward_refs() before instantiation.
-    from great_expectations.core.batch import BatchData, BatchDefinition, BatchMarkers
+    from great_expectations.core.batch import (
+        BatchData,
+        BatchDefinition,
+        BatchMarkers,
+    )
 
 try:
     import pyspark
@@ -77,7 +80,7 @@ class BatchSorter:
     reverse: bool = False
 
 
-BatchSortersDefinition: TypeAlias = Union[List[BatchSorter], List[str]]
+BatchSortersDefinition: TypeAlias = List[Union[BatchSorter, str]]
 
 
 def _is_batch_sorter_list(
@@ -94,11 +97,11 @@ def _is_str_sorter_list(sorters: BatchSortersDefinition) -> TypeGuard[list[str]]
     return False
 
 
-def _batch_sorter_from_list(sorters: BatchSortersDefinition) -> List[BatchSorter]:
+def _batch_sorter_from_list(sorters: BatchSortersDefinition) -> list[BatchSorter]:
     if _is_batch_sorter_list(sorters):
         return sorters
 
-    # mypy doesn't successfully type-narrow sorters to a List[str] here, so we use
+    # mypy doesn't successfully type-narrow sorters to a list[str] here, so we use
     # another TypeGuard. We could cast instead which may be slightly faster.
     sring_valued_sorter: str
     if _is_str_sorter_list(sorters):
@@ -130,7 +133,11 @@ def _batch_sorter_from_str(sort_key: str) -> BatchSorter:
         return BatchSorter(key=sort_key, reverse=False)
 
 
-class DataAsset(ExperimentalBaseModel):
+# It would be best to bind this to Datasource, but we can't now due to circular dependencies
+_DatasourceT = TypeVar("_DatasourceT")
+
+
+class DataAsset(ExperimentalBaseModel, Generic[_DatasourceT]):
     # To subclass a DataAsset one must define `type` as a Class literal explicitly on the sublass
     # as well as implementing the methods in the `Abstract Methods` section below.
     # Some examples:
@@ -139,23 +146,16 @@ class DataAsset(ExperimentalBaseModel):
     # * type: Literal["csv"] = "csv"
     name: str
     type: str
+
     order_by: List[BatchSorter] = Field(default_factory=list)
 
     # non-field private attributes
-    _datasource: Datasource = pydantic.PrivateAttr()
+    _datasource: _DatasourceT = pydantic.PrivateAttr()
 
     @property
-    def datasource(self) -> Datasource:
+    def datasource(self) -> _DatasourceT:
         return self._datasource
 
-    @datasource.setter
-    # TODO (kilo): remove setter and add custom init for DataAsset to inject datasource in constructor??
-    # This setter is non-functional: https://github.com/pydantic/pydantic/issues/3395
-    # There is some related discussion linked from that ticket which may be a workaround.
-    def datasource(self, ds: Datasource):
-        assert isinstance(ds, Datasource)
-        self._datasource = ds
-
     def test_connection(self) -> None:
         """Test the connection for the DataAsset.
 
@@ -199,18 +199,8 @@ class DataAsset(ExperimentalBaseModel):
             A BatchRequest object that can be used to obtain a batch list from a Datasource by calling the
             get_batch_list_from_batch_request method.
         """
-        if options is not None and not self._valid_batch_request_options(options):
-            allowed_keys = set(self.batch_request_options_template().keys())
-            actual_keys = set(options.keys())
-            raise ge_exceptions.InvalidBatchRequestError(
-                "Batch request options should only contain keys from the following set:\n"
-                f"{allowed_keys}\nbut your specified keys contain\n"
-                f"{actual_keys.difference(allowed_keys)}\nwhich is not valid.\n"
-            )
-        return BatchRequest(
-            datasource_name=self.datasource.name,
-            data_asset_name=self.name,
-            options=options or {},
+        raise NotImplementedError(
+            """One must implement "build_batch_request" on a DataAsset subclass."""
         )
 
     def _valid_batch_request_options(self, options: BatchRequestOptions) -> bool:
@@ -224,21 +214,9 @@ class DataAsset(ExperimentalBaseModel):
         Args:
             batch_request: A batch request object to be validated.
         """
-        if not (
-            batch_request.datasource_name == self.datasource.name
-            and batch_request.data_asset_name == self.name
-            and self._valid_batch_request_options(batch_request.options)
-        ):
-            expect_batch_request_form = BatchRequest(
-                datasource_name=self.datasource.name,
-                data_asset_name=self.name,
-                options=self.batch_request_options_template(),
-            )
-            raise ge_exceptions.InvalidBatchRequestError(
-                "BatchRequest should have form:\n"
-                f"{pf(dataclasses.asdict(expect_batch_request_form))}\n"
-                f"but actually has form:\n{pf(dataclasses.asdict(batch_request))}\n"
-            )
+        raise NotImplementedError(
+            """One must implement "_validate_batch_request" on a DataAsset subclass."""
+        )
 
     # Sorter methods
     @pydantic.validator("order_by", pre=True, each_item=True)
@@ -335,6 +313,7 @@ class Datasource(
         "type",
         "execution_engine",
         "assets",
+        "base_directory",
     }
     # Setting this in a Datasource subclass will override the execution engine type.
     # The primary use case is to inject an execution engine for testing.
@@ -450,7 +429,7 @@ class Datasource(
             test_assets: If assets have been passed to the Datasource, an attempt can be made to test them as well.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
         raise NotImplementedError(
             """One needs to implement "test_connection" on a Datasource subclass."""
diff --git a/great_expectations/experimental/datasources/pandas_datasource.py b/great_expectations/experimental/datasources/pandas_datasource.py
index ec8a79b03..27fdc8e2d 100644
--- a/great_expectations/experimental/datasources/pandas_datasource.py
+++ b/great_expectations/experimental/datasources/pandas_datasource.py
@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import logging
+import pathlib
 import re
 from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional, Type, Union
 
+import pydantic
 from typing_extensions import Literal
 
-from great_expectations.alias_types import PathStr  # noqa: TCH001
 from great_expectations.experimental.datasources.dynamic_pandas import (
     _generate_pandas_data_asset_models,
 )
@@ -17,6 +18,8 @@ from great_expectations.experimental.datasources.interfaces import (
     BatchSortersDefinition,
     DataAsset,
     Datasource,
+    TestConnectionError,
+    _batch_sorter_from_list,
 )
 
 if TYPE_CHECKING:
@@ -72,18 +75,18 @@ except KeyError as key_err:
     ParquetAsset = _FilesystemDataAsset
 
 
-class PandasDatasource(Datasource):
+class _PandasDatasource(Datasource):
     # class attributes
     asset_types: ClassVar[List[Type[DataAsset]]] = list(_ASSET_MODELS.values())
 
     # instance attributes
-    type: Literal["pandas"] = "pandas"
-    name: str
+    type: str = pydantic.Field("_pandas")
     assets: Dict[
         str,
         _FilesystemDataAsset,
     ] = {}
 
+    # Abstract Methods
     @property
     def execution_engine_type(self) -> Type[PandasExecutionEngine]:
         """Return the PandasExecutionEngine unless the override is set"""
@@ -93,6 +96,33 @@ class PandasDatasource(Datasource):
 
         return PandasExecutionEngine
 
+    def test_connection(self, test_assets: bool = True) -> None:
+        """Test the connection for the _PandasDatasource.
+
+        Args:
+            test_assets: If assets have been passed to the _PandasDatasource,
+                         an attempt can be made to test them as well.
+
+        Raises:
+            TestConnectionError: If the connection test fails.
+        """
+        raise NotImplementedError(
+            """One needs to implement "test_connection" on a _PandasDatasource subclass."""
+        )
+
+    # End Abstract Methods
+
+
+class PandasFilesystemDatasource(_PandasDatasource):
+    # instance attributes
+    type: Literal["pandas_filesystem"] = "pandas_filesystem"
+    name: str
+    base_directory: pathlib.Path
+    assets: Dict[
+        str,
+        _FilesystemDataAsset,
+    ] = {}
+
     def test_connection(self, test_assets: bool = True) -> None:
         """Test the connection for the PandasDatasource.
 
@@ -100,9 +130,13 @@ class PandasDatasource(Datasource):
             test_assets: If assets have been passed to the PandasDatasource, whether to test them as well.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
-        # Only self.assets can be tested for PandasDatasource
+        if not self.base_directory.exists():
+            raise TestConnectionError(
+                f"Path: {self.base_directory.resolve()} does not exist."
+            )
+
         if self.assets and test_assets:
             for asset in self.assets.values():
                 asset.test_connection()
@@ -110,8 +144,7 @@ class PandasDatasource(Datasource):
     def add_csv_asset(
         self,
         name: str,
-        base_directory: PathStr,
-        regex: Union[str, re.Pattern],
+        regex: Union[re.Pattern, str],
         order_by: Optional[BatchSortersDefinition] = None,
         **kwargs,  # TODO: update signature to have specific keys & types
     ) -> CSVAsset:  # type: ignore[valid-type]
@@ -119,16 +152,16 @@ class PandasDatasource(Datasource):
 
         Args:
             name: The name of the csv asset
-            base_directory: base directory path, relative to which CSV file paths will be collected
             regex: regex pattern that matches csv filenames that is used to label the batches
-            order_by: sorting directive via either List[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
+            order_by: sorting directive via either list[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
             kwargs: Extra keyword arguments should correspond to ``pandas.read_csv`` keyword args
         """
+        if isinstance(regex, str):
+            regex = re.compile(regex)
         asset = CSVAsset(
             name=name,
-            base_directory=base_directory,  # type: ignore[arg-type]
-            regex=regex,  # type: ignore[arg-type]  # type: ignore[arg-type]
-            order_by=order_by or [],  # type: ignore[arg-type]
+            regex=regex,
+            order_by=_batch_sorter_from_list(order_by or []),
             **kwargs,
         )
         return self.add_asset(asset)
@@ -136,7 +169,6 @@ class PandasDatasource(Datasource):
     def add_excel_asset(
         self,
         name: str,
-        base_directory: PathStr,
         regex: Union[str, re.Pattern],
         order_by: Optional[BatchSortersDefinition] = None,
         **kwargs,  # TODO: update signature to have specific keys & types
@@ -145,16 +177,16 @@ class PandasDatasource(Datasource):
 
         Args:
             name: The name of the csv asset
-            base_directory: base directory path, relative to which CSV file paths will be collected
             regex: regex pattern that matches csv filenames that is used to label the batches
-            order_by: sorting directive via either List[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
+            order_by: sorting directive via either list[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
             kwargs: Extra keyword arguments should correspond to ``pandas.read_excel`` keyword args
         """
+        if isinstance(regex, str):
+            regex = re.compile(regex)
         asset = ExcelAsset(
             name=name,
-            base_directory=base_directory,  # type: ignore[arg-type]
-            regex=regex,  # type: ignore[arg-type]
-            order_by=order_by or [],  # type: ignore[arg-type]
+            regex=regex,
+            order_by=_batch_sorter_from_list(order_by or []),
             **kwargs,
         )
         return self.add_asset(asset)
@@ -162,7 +194,6 @@ class PandasDatasource(Datasource):
     def add_json_asset(
         self,
         name: str,
-        base_directory: PathStr,
         regex: Union[str, re.Pattern],
         order_by: Optional[BatchSortersDefinition] = None,
         **kwargs,  # TODO: update signature to have specific keys & types
@@ -171,16 +202,16 @@ class PandasDatasource(Datasource):
 
         Args:
             name: The name of the csv asset
-            base_directory: base directory path, relative to which CSV file paths will be collected
             regex: regex pattern that matches csv filenames that is used to label the batches
-            order_by: sorting directive via either List[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
+            order_by: sorting directive via either list[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
             kwargs: Extra keyword arguments should correspond to ``pandas.read_json`` keyword args
         """
+        if isinstance(regex, str):
+            regex = re.compile(regex)
         asset = JSONAsset(
             name=name,
-            base_directory=base_directory,  # type: ignore[arg-type]
-            regex=regex,  # type: ignore[arg-type]
-            order_by=order_by or [],  # type: ignore[arg-type]
+            regex=regex,
+            order_by=_batch_sorter_from_list(order_by or []),
             **kwargs,
         )
         return self.add_asset(asset)
@@ -188,7 +219,6 @@ class PandasDatasource(Datasource):
     def add_parquet_asset(
         self,
         name: str,
-        base_directory: PathStr,
         regex: Union[str, re.Pattern],
         order_by: Optional[BatchSortersDefinition] = None,
         **kwargs,  # TODO: update signature to have specific keys & types
@@ -197,16 +227,16 @@ class PandasDatasource(Datasource):
 
         Args:
             name: The name of the csv asset
-            base_directory: base directory path, relative to which CSV file paths will be collected
             regex: regex pattern that matches csv filenames that is used to label the batches
-            order_by: sorting directive via either List[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
+            order_by: sorting directive via either list[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
             kwargs: Extra keyword arguments should correspond to ``pandas.read_parquet`` keyword args
         """
+        if isinstance(regex, str):
+            regex = re.compile(regex)
         asset = ParquetAsset(
             name=name,
-            base_directory=base_directory,  # type: ignore[arg-type]
-            regex=regex,  # type: ignore[arg-type]
-            order_by=order_by or [],  # type: ignore[arg-type]
+            regex=regex,
+            order_by=_batch_sorter_from_list(order_by or []),
             **kwargs,
         )
         return self.add_asset(asset)
diff --git a/great_expectations/experimental/datasources/schemas/CSVAsset.json b/great_expectations/experimental/datasources/schemas/CSVAsset.json
index 5b6e79c6c..86f71aa4a 100644
--- a/great_expectations/experimental/datasources/schemas/CSVAsset.json
+++ b/great_expectations/experimental/datasources/schemas/CSVAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "sep": {
             "title": "Sep",
             "default": "NO_DEFAULT"
@@ -245,8 +240,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/schemas/CSVSparkAsset.json b/great_expectations/experimental/datasources/schemas/CSVSparkAsset.json
index 54c78599e..0ab736de2 100644
--- a/great_expectations/experimental/datasources/schemas/CSVSparkAsset.json
+++ b/great_expectations/experimental/datasources/schemas/CSVSparkAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "header": {
             "title": "Header",
             "default": false,
@@ -45,8 +40,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "definitions": {
         "BatchSorter": {
diff --git a/great_expectations/experimental/datasources/schemas/ClipboardAsset.json b/great_expectations/experimental/datasources/schemas/ClipboardAsset.json
index cac2b30d6..20bdec938 100644
--- a/great_expectations/experimental/datasources/schemas/ClipboardAsset.json
+++ b/great_expectations/experimental/datasources/schemas/ClipboardAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "sep": {
             "title": "Sep",
             "default": "\\s+"
@@ -43,7 +38,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "kwargs"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/ExcelAsset.json b/great_expectations/experimental/datasources/schemas/ExcelAsset.json
index bb5aefbff..d2b73039f 100644
--- a/great_expectations/experimental/datasources/schemas/ExcelAsset.json
+++ b/great_expectations/experimental/datasources/schemas/ExcelAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "io": {
             "title": "Io"
         },
@@ -150,7 +145,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "io"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/FeatherAsset.json b/great_expectations/experimental/datasources/schemas/FeatherAsset.json
index 2141bf515..7721fe2bc 100644
--- a/great_expectations/experimental/datasources/schemas/FeatherAsset.json
+++ b/great_expectations/experimental/datasources/schemas/FeatherAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "path": {
             "title": "Path"
         },
@@ -51,7 +46,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "path"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/GBQAsset.json b/great_expectations/experimental/datasources/schemas/GBQAsset.json
index d49e88f02..7980e7079 100644
--- a/great_expectations/experimental/datasources/schemas/GBQAsset.json
+++ b/great_expectations/experimental/datasources/schemas/GBQAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "query": {
             "title": "Query",
             "type": "string"
@@ -92,7 +87,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "query"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/HDFAsset.json b/great_expectations/experimental/datasources/schemas/HDFAsset.json
index 06b1d9738..143993db8 100644
--- a/great_expectations/experimental/datasources/schemas/HDFAsset.json
+++ b/great_expectations/experimental/datasources/schemas/HDFAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "key": {
             "title": "Key"
         },
@@ -74,7 +69,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "kwargs"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/HTMLAsset.json b/great_expectations/experimental/datasources/schemas/HTMLAsset.json
index 79d4fcdc4..fc042995a 100644
--- a/great_expectations/experimental/datasources/schemas/HTMLAsset.json
+++ b/great_expectations/experimental/datasources/schemas/HTMLAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "match": {
             "title": "Match",
             "default": ".+",
@@ -137,8 +132,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/schemas/JSONAsset.json b/great_expectations/experimental/datasources/schemas/JSONAsset.json
index a76b5760a..62df9208e 100644
--- a/great_expectations/experimental/datasources/schemas/JSONAsset.json
+++ b/great_expectations/experimental/datasources/schemas/JSONAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "orient": {
             "title": "Orient"
         },
@@ -117,8 +112,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/schemas/ORCAsset.json b/great_expectations/experimental/datasources/schemas/ORCAsset.json
index c5eb18d02..178e838a9 100644
--- a/great_expectations/experimental/datasources/schemas/ORCAsset.json
+++ b/great_expectations/experimental/datasources/schemas/ORCAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "columns": {
             "title": "Columns",
             "type": "array",
@@ -46,7 +41,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "kwargs"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/ParquetAsset.json b/great_expectations/experimental/datasources/schemas/ParquetAsset.json
index 6d857e3db..96aee4d1b 100644
--- a/great_expectations/experimental/datasources/schemas/ParquetAsset.json
+++ b/great_expectations/experimental/datasources/schemas/ParquetAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "path": {
             "title": "Path"
         },
@@ -59,7 +54,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "path",
         "kwargs"
     ],
diff --git a/great_expectations/experimental/datasources/schemas/PickleAsset.json b/great_expectations/experimental/datasources/schemas/PickleAsset.json
index e3d3c7665..a9d3b6ea1 100644
--- a/great_expectations/experimental/datasources/schemas/PickleAsset.json
+++ b/great_expectations/experimental/datasources/schemas/PickleAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "compression": {
             "title": "Compression",
             "default": "infer",
@@ -51,8 +46,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/schemas/SPSSAsset.json b/great_expectations/experimental/datasources/schemas/SPSSAsset.json
index 10042099e..a6febb8e8 100644
--- a/great_expectations/experimental/datasources/schemas/SPSSAsset.json
+++ b/great_expectations/experimental/datasources/schemas/SPSSAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "path": {
             "title": "Path",
             "anyOf": [
@@ -70,7 +65,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "path"
     ],
     "additionalProperties": false,
diff --git a/great_expectations/experimental/datasources/schemas/SQLQueryAsset.json b/great_expectations/experimental/datasources/schemas/SQLQueryAsset.json
index a9acd4d35..21fd1ea91 100644
--- a/great_expectations/experimental/datasources/schemas/SQLQueryAsset.json
+++ b/great_expectations/experimental/datasources/schemas/SQLQueryAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "sql": {
             "title": "Sql"
         },
@@ -64,7 +59,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "sql",
         "con"
     ],
diff --git a/great_expectations/experimental/datasources/schemas/SQLTableAsset.json b/great_expectations/experimental/datasources/schemas/SQLTableAsset.json
index da1b78a57..e2302b543 100644
--- a/great_expectations/experimental/datasources/schemas/SQLTableAsset.json
+++ b/great_expectations/experimental/datasources/schemas/SQLTableAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "table_name": {
             "title": "Table Name",
             "type": "string"
@@ -77,7 +72,6 @@
     "required": [
         "name",
         "regex",
-        "base_directory",
         "table_name",
         "con"
     ],
diff --git a/great_expectations/experimental/datasources/schemas/STATAAsset.json b/great_expectations/experimental/datasources/schemas/STATAAsset.json
index a9ec772cc..1e576a229 100644
--- a/great_expectations/experimental/datasources/schemas/STATAAsset.json
+++ b/great_expectations/experimental/datasources/schemas/STATAAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "convert_dates": {
             "title": "Convert Dates",
             "default": true,
@@ -105,8 +100,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/schemas/XMLAsset.json b/great_expectations/experimental/datasources/schemas/XMLAsset.json
index 998355c87..224a09821 100644
--- a/great_expectations/experimental/datasources/schemas/XMLAsset.json
+++ b/great_expectations/experimental/datasources/schemas/XMLAsset.json
@@ -27,11 +27,6 @@
             "type": "string",
             "format": "regex"
         },
-        "base_directory": {
-            "title": "Base Directory",
-            "type": "string",
-            "format": "path"
-        },
         "xpath": {
             "title": "Xpath",
             "default": "./*",
@@ -96,8 +91,7 @@
     },
     "required": [
         "name",
-        "regex",
-        "base_directory"
+        "regex"
     ],
     "additionalProperties": false,
     "definitions": {
diff --git a/great_expectations/experimental/datasources/sources.py b/great_expectations/experimental/datasources/sources.py
index 4a65fbb8f..d7ac75b21 100644
--- a/great_expectations/experimental/datasources/sources.py
+++ b/great_expectations/experimental/datasources/sources.py
@@ -56,10 +56,10 @@ class _SourceFactories:
         Example
         -------
 
-        An `.add_pandas()` pandas factory method will be added to `context.sources`.
+        An `.add_pandas_filesystem()` pandas_filesystem factory method will be added to `context.sources`.
 
-        >>> class PandasDatasource(Datasource):
-        >>>     type: str = 'pandas'
+        >>> class PandasFilesystemDatasource(_PandasDatasource):
+        >>>     type: str = 'pandas_filesystem'
         >>>     asset_types = [FileAsset]
         >>>     execution_engine: PandasExecutionEngine
         """
diff --git a/great_expectations/experimental/datasources/spark_datasource.py b/great_expectations/experimental/datasources/spark_datasource.py
index c46e20cdc..ce1f93c10 100644
--- a/great_expectations/experimental/datasources/spark_datasource.py
+++ b/great_expectations/experimental/datasources/spark_datasource.py
@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import logging
+import pathlib
 import re
 from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional, Type, Union
 
+import pydantic
 from typing_extensions import Literal
 
-from great_expectations.alias_types import PathStr  # noqa: TCH001
 from great_expectations.experimental.datasources.filesystem_data_asset import (
     _FilesystemDataAsset,
 )
@@ -14,6 +15,8 @@ from great_expectations.experimental.datasources.interfaces import (
     BatchSortersDefinition,
     DataAsset,
     Datasource,
+    TestConnectionError,
+    _batch_sorter_from_list,
 )
 
 if TYPE_CHECKING:
@@ -41,15 +44,18 @@ class CSVSparkAsset(_FilesystemDataAsset):
         return {"header", "inferSchema"}
 
 
-class SparkDatasource(Datasource):
+class _SparkDatasource(Datasource):
     # class attributes
     asset_types: ClassVar[List[Type[DataAsset]]] = [CSVSparkAsset]
 
     # instance attributes
-    type: Literal["spark"] = "spark"
-    name: str
-    assets: Dict[str, CSVSparkAsset] = {}
+    type: str = pydantic.Field("_spark")
+    assets: Dict[
+        str,
+        CSVSparkAsset,
+    ] = {}
 
+    # Abstract Methods
     @property
     def execution_engine_type(self) -> Type[SparkDFExecutionEngine]:
         """Return the SparkDFExecutionEngine unless the override is set"""
@@ -59,6 +65,30 @@ class SparkDatasource(Datasource):
 
         return SparkDFExecutionEngine
 
+    def test_connection(self, test_assets: bool = True) -> None:
+        """Test the connection for the _SparkDatasource.
+
+        Args:
+            test_assets: If assets have been passed to the _SparkDatasource,
+                         an attempt can be made to test them as well.
+
+        Raises:
+            TestConnectionError: If the connection test fails.
+        """
+        raise NotImplementedError(
+            """One needs to implement "test_connection" on a _SparkDatasource subclass."""
+        )
+
+    # End Abstract Methods
+
+
+class SparkDatasource(_SparkDatasource):
+    # instance attributes
+    type: Literal["spark"] = "spark"
+    name: str
+    base_directory: pathlib.Path
+    assets: Dict[str, CSVSparkAsset] = {}
+
     def test_connection(self, test_assets: bool = True) -> None:
         """Test the connection for the SparkDatasource.
 
@@ -68,7 +98,11 @@ class SparkDatasource(Datasource):
         Raises:
             TestConnectionError: If the connection test fails.
         """
-        # Only self.assets can be tested for PandasDatasource
+        if not self.base_directory.exists():
+            raise TestConnectionError(
+                f"Path: {self.base_directory.resolve()} does not exist."
+            )
+
         if self.assets and test_assets:
             for asset in self.assets.values():
                 asset.test_connection()
@@ -76,7 +110,6 @@ class SparkDatasource(Datasource):
     def add_csv_asset(
         self,
         name: str,
-        base_directory: PathStr,
         regex: Union[str, re.Pattern],
         header: bool = False,
         infer_schema: bool = False,
@@ -86,18 +119,18 @@ class SparkDatasource(Datasource):
 
         Args:
             name: The name of the csv asset
-            base_directory: base directory path, relative to which CSV file paths will be collected
             regex: regex pattern that matches csv filenames that is used to label the batches
             header: boolean (default False) indicating whether or not first line of CSV file is header line
             infer_schema: boolean (default False) instructing Spark to attempt to infer schema of CSV file heuristically
-            order_by: sorting directive via either List[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
+            order_by: sorting directive via either list[BatchSorter] or "{+|-}key" syntax: +/- (a/de)scending; + default
         """
+        if isinstance(regex, str):
+            regex = re.compile(regex)
         asset = CSVSparkAsset(
             name=name,
-            base_directory=base_directory,  # type: ignore[arg-type]  # str will be coerced to Path
-            regex=regex,  # type: ignore[arg-type]  # str with will coerced to Pattern
+            regex=regex,
             header=header,
             inferSchema=infer_schema,
-            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            order_by=_batch_sorter_from_list(order_by or []),
         )
         return self.add_asset(asset)
diff --git a/great_expectations/experimental/datasources/sql_datasource.py b/great_expectations/experimental/datasources/sql_datasource.py
index 7b663cef2..bc159cfda 100644
--- a/great_expectations/experimental/datasources/sql_datasource.py
+++ b/great_expectations/experimental/datasources/sql_datasource.py
@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import copy
+import dataclasses
 import itertools
 from datetime import datetime
+from pprint import pformat as pf
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -24,6 +26,7 @@ import pydantic
 from pydantic import dataclasses as pydantic_dc
 from typing_extensions import Literal
 
+import great_expectations.exceptions as gx_exceptions
 from great_expectations.core.batch_spec import SqlAlchemyDatasourceBatchSpec
 from great_expectations.experimental.datasources.interfaces import (
     Batch,
@@ -33,6 +36,7 @@ from great_expectations.experimental.datasources.interfaces import (
     DataAsset,
     Datasource,
     TestConnectionError,
+    _batch_sorter_from_list,
 )
 
 SQLALCHEMY_IMPORTED = False
@@ -92,10 +96,8 @@ class ColumnSplitter:
         Raises:
             TestConnectionError: If the connection test fails.
         """
-        # A TypeVar for Datasource would get rid of this assertion,
-        # but circularity between Datasource <-> DataAsset is an issue.
-        assert isinstance(table_asset.datasource, SQLDatasource)
-        engine: sqlalchemy.engine.Engine = table_asset.datasource.get_engine()
+        datasource: SQLDatasource = table_asset.datasource
+        engine: sqlalchemy.engine.Engine = datasource.get_engine()
         inspector: sqlalchemy.engine.Inspector = sqlalchemy.inspect(engine)
 
         columns: list[dict[str, Any]] = inspector.get_columns(
@@ -184,10 +186,13 @@ def _get_sql_datetime_range(
 
 class _SQLAsset(DataAsset, Generic[_ColumnSplitterT]):
     # Instance fields
-    type: Literal["_sqlasset"] = "_sqlasset"
+    type: str = pydantic.Field("_sql_asset")
     column_splitter: Optional[_ColumnSplitterT] = None
     name: str
 
+    # Internal attributes
+    _datasource: "SQLDatasource" = pydantic.PrivateAttr()
+
     def batch_request_options_template(
         self,
     ) -> BatchRequestOptions:
@@ -344,6 +349,56 @@ class _SQLAsset(DataAsset, Generic[_ColumnSplitterT]):
         self.sort_batches(batch_list)
         return batch_list
 
+    def build_batch_request(
+        self, options: Optional[BatchRequestOptions] = None
+    ) -> BatchRequest:
+        """A batch request that can be used to obtain batches for this DataAsset.
+
+        Args:
+            options: A dict that can be used to limit the number of batches returned from the asset.
+                The dict structure depends on the asset type. A template of the dict can be obtained by
+                calling batch_request_options_template.
+
+        Returns:
+            A BatchRequest object that can be used to obtain a batch list from a Datasource by calling the
+            get_batch_list_from_batch_request method.
+        """
+        if options is not None and not self._valid_batch_request_options(options):
+            allowed_keys = set(self.batch_request_options_template().keys())
+            actual_keys = set(options.keys())
+            raise gx_exceptions.InvalidBatchRequestError(
+                "Batch request options should only contain keys from the following set:\n"
+                f"{allowed_keys}\nbut your specified keys contain\n"
+                f"{actual_keys.difference(allowed_keys)}\nwhich is not valid.\n"
+            )
+        return BatchRequest(
+            datasource_name=self.datasource.name,
+            data_asset_name=self.name,
+            options=options or {},
+        )
+
+    def _validate_batch_request(self, batch_request: BatchRequest) -> None:
+        """Validates the batch_request has the correct form.
+
+        Args:
+            batch_request: A batch request object to be validated.
+        """
+        if not (
+            batch_request.datasource_name == self.datasource.name
+            and batch_request.data_asset_name == self.name
+            and self._valid_batch_request_options(batch_request.options)
+        ):
+            expect_batch_request_form = BatchRequest(
+                datasource_name=self.datasource.name,
+                data_asset_name=self.name,
+                options=self.batch_request_options_template(),
+            )
+            raise gx_exceptions.InvalidBatchRequestError(
+                "BatchRequest should have form:\n"
+                f"{pf(dataclasses.asdict(expect_batch_request_form))}\n"
+                f"but actually has form:\n{pf(dataclasses.asdict(batch_request))}\n"
+            )
+
     def _create_batch_spec_kwargs(self) -> dict[str, Any]:
         """Creates batch_spec_kwargs used to instantiate a SqlAlchemyDatasourceBatchSpec
 
@@ -365,7 +420,7 @@ class _SQLAsset(DataAsset, Generic[_ColumnSplitterT]):
 
 class QueryAsset(_SQLAsset):
     # Instance fields
-    type: Literal["query"] = "query"  # type: ignore[assignment]
+    type: Literal["query"] = "query"
     query: str
     column_splitter: Optional[SqlYearMonthSplitter] = None
 
@@ -416,7 +471,7 @@ class QueryAsset(_SQLAsset):
 
 class TableAsset(_SQLAsset):
     # Instance fields
-    type: Literal["table"] = "table"  # type: ignore[assignment]
+    type: Literal["table"] = "table"
     table_name: str
     schema_name: Optional[str] = None
     column_splitter: Optional[SqlYearMonthSplitter] = None
@@ -435,10 +490,8 @@ class TableAsset(_SQLAsset):
         Raises:
             TestConnectionError: If the connection test fails.
         """
-        # A TypeVar for Datasource would get rid of this assertion,
-        # but circularity between Datasource <-> DataAsset is an issue.
-        assert isinstance(self.datasource, SQLDatasource)
-        engine: sqlalchemy.engine.Engine = self.datasource.get_engine()
+        datasource: SQLDatasource = self.datasource
+        engine: sqlalchemy.engine.Engine = datasource.get_engine()
         inspector: sqlalchemy.engine.Inspector = sqlalchemy.inspect(engine)
 
         if self.schema_name and self.schema_name not in inspector.get_schema_names():
@@ -593,7 +646,7 @@ class SQLDatasource(Datasource):
             name=name,
             table_name=table_name,
             schema_name=schema_name,
-            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            order_by=_batch_sorter_from_list(order_by or []),
             # see DataAsset._parse_order_by_sorter()
         )
         return self.add_asset(asset)
@@ -617,6 +670,6 @@ class SQLDatasource(Datasource):
         asset = QueryAsset(
             name=name,
             query=query,
-            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            order_by=_batch_sorter_from_list(order_by or []),
         )
         return self.add_asset(asset)
diff --git a/great_expectations/experimental/datasources/sqlite_datasource.py b/great_expectations/experimental/datasources/sqlite_datasource.py
index 6c5b7d392..b8321df9d 100644
--- a/great_expectations/experimental/datasources/sqlite_datasource.py
+++ b/great_expectations/experimental/datasources/sqlite_datasource.py
@@ -6,13 +6,11 @@ from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional, Type, TypeVar,
 import pydantic
 from pydantic import dataclasses as pydantic_dc
 
-from great_expectations.experimental.datasources.interfaces import (
-    DataAsset,  # noqa: TCH001
-)
-
 if TYPE_CHECKING:
     import sqlalchemy
 
+    from great_expectations.experimental.datasources.interfaces import DataAsset
+
 from typing_extensions import Literal
 
 from great_expectations.experimental.datasources.sql_datasource import (
@@ -23,6 +21,7 @@ from great_expectations.experimental.datasources.sql_datasource import (
     SQLDatasourceError,
     SqlYearMonthSplitter,
     TableAsset,
+    _batch_sorter_from_list,
     _query_for_year_and_month,
     _SQLAsset,
 )
@@ -188,7 +187,7 @@ class SqliteDatasource(SQLDatasource):
             name=name,
             table_name=table_name,
             schema_name=schema_name,
-            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            order_by=_batch_sorter_from_list(order_by or []),
             # see TableAsset._parse_order_by_sorter()
         )
         asset._datasource = self
@@ -214,7 +213,7 @@ class SqliteDatasource(SQLDatasource):
         asset = SqliteQueryAsset(
             name=name,
             query=query,
-            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            order_by=_batch_sorter_from_list(order_by or []),
             # see TableAsset._parse_order_by_sorter()
         )
         asset._datasource = self
diff --git a/tasks.py b/tasks.py
index 33f21bbb3..bd54e1a68 100644
--- a/tasks.py
+++ b/tasks.py
@@ -499,7 +499,7 @@ def type_schema(
     from great_expectations.experimental.datasources import _PANDAS_SCHEMA_VERSION
     from great_expectations.experimental.datasources.interfaces import Datasource
     from great_expectations.experimental.datasources.pandas_datasource import (
-        PandasDatasource,
+        PandasFilesystemDatasource,
     )
     from great_expectations.experimental.datasources.sources import _SourceFactories
 
@@ -525,7 +525,8 @@ def type_schema(
             model = _SourceFactories.type_lookup[name]
 
             if (
-                model in {*PandasDatasource.asset_types, PandasDatasource}
+                model
+                in {*PandasFilesystemDatasource.asset_types, PandasFilesystemDatasource}
                 and _PANDAS_SCHEMA_VERSION != pandas.__version__
             ):
                 print(


Test patch:
diff --git a/tests/experimental/datasources/integration/conftest.py b/tests/experimental/datasources/integration/conftest.py
index 1c88459dd..329451ad5 100644
--- a/tests/experimental/datasources/integration/conftest.py
+++ b/tests/experimental/datasources/integration/conftest.py
@@ -25,10 +25,12 @@ def pandas_data(
     csv_path = (
         pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
     )
-    pandas_ds = context.sources.add_pandas(name="my_pandas")
+    pandas_ds = context.sources.add_pandas_filesystem(
+        name="my_pandas",
+        base_directory=csv_path,
+    )
     asset = pandas_ds.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2})\.csv",
         order_by=["year", "month"],
     )
@@ -81,10 +83,12 @@ def spark_data(
     csv_path = (
         pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
     )
-    spark_ds = context.sources.add_spark(name="my_spark")
+    spark_ds = context.sources.add_spark(
+        name="my_spark",
+        base_directory=csv_path,
+    )
     asset = spark_ds.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2})\.csv",
         order_by=["year", "month"],
         header=True,
@@ -103,10 +107,12 @@ def multibatch_pandas_data(
     csv_path = (
         pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
     )
-    pandas_ds = context.sources.add_pandas(name="my_pandas")
+    pandas_ds = context.sources.add_pandas_filesystem(
+        name="my_pandas",
+        base_directory=csv_path,
+    )
     asset = pandas_ds.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2})\.csv",
         order_by=["year", "month"],
     )
@@ -141,10 +147,12 @@ def multibatch_spark_data(
     csv_path = (
         pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
     )
-    spark_ds = context.sources.add_spark(name="my_spark")
+    spark_ds = context.sources.add_spark(
+        name="my_spark",
+        base_directory=csv_path,
+    )
     asset = spark_ds.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2})\.csv",
         order_by=["year", "month"],
         header=True,
diff --git a/tests/experimental/datasources/test_config.py b/tests/experimental/datasources/test_config.py
index 855102359..fc345a42e 100644
--- a/tests/experimental/datasources/test_config.py
+++ b/tests/experimental/datasources/test_config.py
@@ -72,13 +72,13 @@ PG_COMPLEX_CONFIG_DICT = {
             },
         },
         "my_pandas_ds": {
-            "type": "pandas",
+            "type": "pandas_filesystem",
             "name": "my_pandas_ds",
+            "base_directory": __file__,
             "assets": {
                 "my_csv_asset": {
                     "name": "my_csv_asset",
                     "type": "csv",
-                    "base_directory": __file__,
                     "regex": r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
                     "sep": "|",
                     "names": ["col1", "col2"],
@@ -86,7 +86,6 @@ PG_COMPLEX_CONFIG_DICT = {
                 "my_json_asset": {
                     "name": "my_json_asset",
                     "type": "json",
-                    "base_directory": __file__,
                     "regex": r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).json",
                     "orient": "records",
                 },
@@ -152,7 +151,7 @@ class TestExcludeUnsetAssetFields:
     """
 
     def test_from_datasource(self, asset_dict: dict):
-        ds_mapping = {"csv": "pandas", "json": "pandas"}
+        ds_mapping = {"csv": "pandas_filesystem", "json": "pandas_filesystem"}
 
         ds_type_: str = ds_mapping[asset_dict["type"]]
         ds_class = _SourceFactories.type_lookup[ds_type_]
@@ -161,12 +160,15 @@ class TestExcludeUnsetAssetFields:
         asset_dict.update(
             {
                 "name": "my_asset",
-                "base_directory": pathlib.Path(__file__),
                 "regex": re.compile(r"sample_(?P<year>\d{4})-(?P<month>\d{2}).csv"),
             }
         )
         asset_name = asset_dict["name"]
-        ds_dict = {"name": "my_ds", "assets": {asset_name: asset_dict}}
+        ds_dict = {
+            "name": "my_ds",
+            "base_directory": pathlib.Path(__file__),
+            "assets": {asset_name: asset_dict},
+        }
         datasource: Datasource = ds_class.parse_obj(ds_dict)
         assert asset_dict == datasource.dict()["assets"][asset_name]
 
@@ -178,14 +180,14 @@ class TestExcludeUnsetAssetFields:
         asset_dict.update(
             {
                 "name": "my_asset",
-                "base_directory": pathlib.Path(__file__),
                 "regex": re.compile(r"sample_(?P<year>\d{4})-(?P<month>\d{2}).csv"),
             }
         )
         asset_name = asset_dict["name"]
         ds_dict = {
             "name": "my_ds",
-            "type": "pandas",
+            "type": "pandas_filesystem",
+            "base_directory": pathlib.Path(__file__),
             "assets": {asset_name: asset_dict},
         }
         gx_config = GxConfig.parse_obj({"xdatasources": {"my_ds": ds_dict}})
diff --git a/tests/experimental/datasources/test_pandas_datasource.py b/tests/experimental/datasources/test_pandas_datasource.py
index 5f826f326..860e9493e 100644
--- a/tests/experimental/datasources/test_pandas_datasource.py
+++ b/tests/experimental/datasources/test_pandas_datasource.py
@@ -18,7 +18,7 @@ from great_expectations.experimental.datasources.interfaces import TestConnectio
 from great_expectations.experimental.datasources.pandas_datasource import (
     CSVAsset,
     JSONAsset,
-    PandasDatasource,
+    PandasFilesystemDatasource,
     _FilesystemDataAsset,
 )
 
@@ -40,8 +40,19 @@ pytestmark = [
 
 
 @pytest.fixture
-def pandas_datasource() -> PandasDatasource:
-    return PandasDatasource(name="pandas_datasource")
+def pandas_filesystem_datasource() -> PandasFilesystemDatasource:
+    base_directory_rel_path = pathlib.Path(
+        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
+    )
+    base_directory_abs_path = (
+        pathlib.Path(__file__)
+        .parent.joinpath(base_directory_rel_path)
+        .resolve(strict=True)
+    )
+    return PandasFilesystemDatasource(
+        name="pandas_filesystem_datasource",
+        base_directory=base_directory_abs_path,
+    )
 
 
 @pytest.fixture
@@ -141,16 +152,17 @@ class TestDynamicPandasAssets:
         assert type_name
 
         asset_class_names: set[str] = {
-            t.__name__.lower().split("asset")[0] for t in PandasDatasource.asset_types
+            t.__name__.lower().split("asset")[0]
+            for t in PandasFilesystemDatasource.asset_types
         }
         print(asset_class_names)
 
         assert type_name in asset_class_names
 
-    @pytest.mark.parametrize("asset_class", PandasDatasource.asset_types)
+    @pytest.mark.parametrize("asset_class", PandasFilesystemDatasource.asset_types)
     def test_minimal_validation(self, asset_class: Type[_FilesystemDataAsset]):
         """
-        These parametrized tests ensures that every `PandasDatasource` asset model does some minimal
+        These parametrized tests ensures that every `PandasFilesystemDatasource` asset model does some minimal
         validation, and doesn't accept arbitrary keyword arguments.
         This is also a proxy for testing that the dynamic pydantic model creation was successful.
         """
@@ -189,7 +201,6 @@ class TestDynamicPandasAssets:
         """
         kwargs: dict[str, Any] = {
             "name": "test",
-            "base_directory": pathlib.Path(__file__),
             "regex": re.compile(r"yellow_tripdata_sample_(\d{4})-(\d{2})"),
         }
         kwargs.update(extra_kwargs)
@@ -213,10 +224,12 @@ class TestDynamicPandasAssets:
         extra_kwargs: dict,
     ):
         batch_request = (
-            empty_data_context.sources.add_pandas("my_pandas")
+            empty_data_context.sources.add_pandas_filesystem(
+                "my_pandas",
+                base_directory=csv_path,
+            )
             .add_csv_asset(
                 "my_csv",
-                base_directory=csv_path,
                 regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
                 **extra_kwargs,
             )
@@ -233,21 +246,21 @@ class TestDynamicPandasAssets:
 
 
 @pytest.mark.unit
-def test_construct_pandas_datasource(pandas_datasource: PandasDatasource):
-    assert pandas_datasource.name == "pandas_datasource"
+def test_construct_pandas_filesystem_datasource(
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
+):
+    assert pandas_filesystem_datasource.name == "pandas_filesystem_datasource"
 
 
 @pytest.mark.unit
 def test_add_csv_asset_to_datasource(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",
     )
     assert asset.name == "csv_asset"  # type: ignore[attr-defined]
-    assert asset.base_directory == csv_path  # type: ignore[attr-defined]
     assert asset.regex.match("random string") is None  # type: ignore[attr-defined]
     assert asset.regex.match("yellow_tripdata_sample_11D1-22.csv") is None  # type: ignore[attr-defined]
     m1 = asset.regex.match("yellow_tripdata_sample_1111-22.csv")  # type: ignore[attr-defined]
@@ -255,15 +268,13 @@ def test_add_csv_asset_to_datasource(
 
 
 @pytest.mark.unit
-def test_construct_csv_asset_directly(csv_path: pathlib.Path):
+def test_construct_csv_asset_directly():
     # noinspection PyTypeChecker
-    asset = CSVAsset(  # type: ignore[call-arg]
+    asset = CSVAsset(
         name="csv_asset",
-        base_directory=csv_path,
-        regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",  # type: ignore[arg-type]
+        regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",
     )
     assert asset.name == "csv_asset"
-    assert asset.base_directory == csv_path
     assert asset.regex.match("random string") is None
     assert asset.regex.match("yellow_tripdata_sample_11D1-22.csv") is None
     m1 = asset.regex.match("yellow_tripdata_sample_1111-22.csv")
@@ -272,11 +283,10 @@ def test_construct_csv_asset_directly(csv_path: pathlib.Path):
 
 @pytest.mark.unit
 def test_csv_asset_with_regex_unnamed_parameters(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",
     )
     options = asset.batch_request_options_template()  # type: ignore[attr-defined]
@@ -285,11 +295,10 @@ def test_csv_asset_with_regex_unnamed_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_regex_named_parameters(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
     )
     options = asset.batch_request_options_template()  # type: ignore[attr-defined]
@@ -298,11 +307,10 @@ def test_csv_asset_with_regex_named_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_some_regex_named_parameters(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(?P<month>\d{2}).csv",
     )
     options = asset.batch_request_options_template()  # type: ignore[attr-defined]
@@ -311,11 +319,10 @@ def test_csv_asset_with_some_regex_named_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_non_string_regex_named_parameters(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(?P<month>\d{2}).csv",
     )
     with pytest.raises(ge_exceptions.InvalidBatchRequestError):
@@ -325,36 +332,39 @@ def test_csv_asset_with_non_string_regex_named_parameters(
 
 @pytest.mark.unit
 def test_get_batch_list_from_fully_specified_batch_request(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
     )
     request = asset.build_batch_request({"year": "2018", "month": "04"})  # type: ignore[attr-defined]
     batches = asset.get_batch_list_from_batch_request(request)  # type: ignore[attr-defined]
     assert len(batches) == 1
     batch = batches[0]
-    assert batch.batch_request.datasource_name == pandas_datasource.name
+    assert batch.batch_request.datasource_name == pandas_filesystem_datasource.name
     assert batch.batch_request.data_asset_name == asset.name  # type: ignore[attr-defined]
     assert batch.batch_request.options == {"year": "2018", "month": "04"}
     assert batch.metadata == {
         "year": "2018",
         "month": "04",
-        "base_directory": asset.base_directory / "yellow_tripdata_sample_2018-04.csv",  # type: ignore[attr-defined]
+        "base_directory": pandas_filesystem_datasource.base_directory
+        / "yellow_tripdata_sample_2018-04.csv",
     }
-    assert batch.id == "pandas_datasource-csv_asset-year_2018-month_04"
+    assert batch.id == "pandas_filesystem_datasource-csv_asset-year_2018-month_04"
 
 
 @pytest.mark.unit
 def test_get_batch_list_from_partially_specified_batch_request(
-    pandas_datasource: PandasDatasource, csv_path: pathlib.Path
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
 ):
     # Verify test directory has files that don't match what we will query for
     file_name: PathStr
     all_files: list[str] = [
-        file_name.stem for file_name in list(pathlib.Path(csv_path).iterdir())
+        file_name.stem
+        for file_name in list(
+            pathlib.Path(pandas_filesystem_datasource.base_directory).iterdir()
+        )
     ]
     # assert there are files that are not csv files
     assert any([not file_name.endswith("csv") for file_name in all_files])
@@ -364,9 +374,8 @@ def test_get_batch_list_from_partially_specified_batch_request(
     ]
     assert len(files_for_2018) == 12
 
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
     )
     request = asset.build_batch_request({"year": "2018"})  # type: ignore[attr-defined]
@@ -417,8 +426,7 @@ def test_get_batch_list_from_partially_specified_batch_request(
     ],
 )
 def test_pandas_sorter(
-    pandas_datasource: PandasDatasource,
-    csv_path: pathlib.Path,
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
     order_by: BatchSortersDefinition,
 ):
     # Verify test directory has files we expect
@@ -426,7 +434,10 @@ def test_pandas_sorter(
     months = [format(m, "02d") for m in range(1, 13)]
     file_name: PathStr
     all_files: list[str] = [
-        file_name.stem for file_name in list(pathlib.Path(csv_path).iterdir())
+        file_name.stem
+        for file_name in list(
+            pathlib.Path(pandas_filesystem_datasource.base_directory).iterdir()
+        )
     ]
     # assert there are 12 files for each year
     for year in years:
@@ -437,9 +448,8 @@ def test_pandas_sorter(
         ]
         assert len(files_for_year) == 12
 
-    asset = pandas_datasource.add_csv_asset(
+    asset = pandas_filesystem_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
         order_by=order_by,
     )
@@ -475,54 +485,41 @@ def test_pandas_sorter(
             assert metadata[key2] == range2
 
 
-def bad_base_directory_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
-    base_directory = pathlib.Path("/this/path/is/not/here")
-    regex = re.compile(r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
-    test_connection_error = TestConnectionError(
-        f"Path: {base_directory.resolve()} does not exist."
-    )
-    return base_directory, regex, test_connection_error
-
-
-def bad_regex_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
-    relative_path = pathlib.Path(
-        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
-    )
-    base_directory = (
-        pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
-    )
+def bad_regex_config(csv_path: pathlib.Path) -> tuple[re.Pattern, TestConnectionError]:
     regex = re.compile(r"green_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
     test_connection_error = TestConnectionError(
-        f"No file at path: {base_directory.resolve()} matched the regex: {regex.pattern}"
+        f"No file at path: {csv_path.resolve()} matched the regex: {regex.pattern}"
     )
-    return base_directory, regex, test_connection_error
+    return regex, test_connection_error
 
 
-@pytest.fixture(params=[bad_base_directory_config, bad_regex_config])
+@pytest.fixture(params=[bad_regex_config])
 def datasource_test_connection_error_messages(
-    pandas_datasource: PandasDatasource, request
-) -> tuple[PandasDatasource, TestConnectionError]:
-    base_directory, regex, test_connection_error = request.param()
+    csv_path: pathlib.Path,
+    pandas_filesystem_datasource: PandasFilesystemDatasource,
+    request,
+) -> tuple[PandasFilesystemDatasource, TestConnectionError]:
+    regex, test_connection_error = request.param(csv_path=csv_path)
     csv_asset = CSVAsset(  # type: ignore[call-arg]
         name="csv_asset",
-        base_directory=base_directory,
         regex=regex,
     )
-    pandas_datasource.assets = {"csv_asset": csv_asset}
-    return pandas_datasource, test_connection_error
+    csv_asset._datasource = pandas_filesystem_datasource
+    pandas_filesystem_datasource.assets = {"csv_asset": csv_asset}
+    return pandas_filesystem_datasource, test_connection_error
 
 
 @pytest.mark.unit
 def test_test_connection_failures(
     datasource_test_connection_error_messages: tuple[
-        PandasDatasource, TestConnectionError
+        PandasFilesystemDatasource, TestConnectionError
     ]
 ):
     (
-        pandas_datasource,
+        pandas_filesystem_datasource,
         test_connection_error,
     ) = datasource_test_connection_error_messages
 
     with pytest.raises(type(test_connection_error)) as e:
-        pandas_datasource.test_connection()
+        pandas_filesystem_datasource.test_connection()
     assert str(e.value) == str(test_connection_error)
diff --git a/tests/experimental/datasources/test_schemas.py b/tests/experimental/datasources/test_schemas.py
index 565f91978..7a466a38f 100644
--- a/tests/experimental/datasources/test_schemas.py
+++ b/tests/experimental/datasources/test_schemas.py
@@ -40,7 +40,7 @@ def test_vcs_schemas_match(zep_ds_or_asset_model: Type[pydantic.BaseModel]):
     """
 
     def _sort_required_lists(schema_as_dict: dict) -> None:
-        """Someties "required" lists come unsorted, causing misleading assertion failures; this corrects the issue.
+        """Sometimes "required" lists come unsorted, causing misleading assertion failures; this corrects the issue.
 
         Args:
             schema_as_dict: source dictionary (will be modified "in-situ")
@@ -62,7 +62,6 @@ def test_vcs_schemas_match(zep_ds_or_asset_model: Type[pydantic.BaseModel]):
     print(f"pandas version: {PANDAS_VERSION}\n")
 
     schema_path = _SCHEMAS_DIR.joinpath(f"{zep_ds_or_asset_model.__name__}.json")
-    print(schema_path.relative_to(schema_path.cwd()))
 
     # TODO: remove this logic and make this fail once all json schemas are working
     if schema_path.name in (
@@ -70,7 +69,12 @@ def test_vcs_schemas_match(zep_ds_or_asset_model: Type[pydantic.BaseModel]):
         "SqliteTableAsset.json",
         "SqliteQueryAsset.json",
         "SASAsset.json",
-        "PandasDatasource.json",
+        "SQLDatasource.json",
+        "PostgresDatasource.json",
+        "PandasFilesystemDatasource.json",
+        "SparkDatasource.json",
+        "_PandasDatasource.json",
+        "_SparkDatasource.json",
     ):
         pytest.xfail(f"{schema_path.name} does not exist")
 
diff --git a/tests/experimental/datasources/test_spark_datasource.py b/tests/experimental/datasources/test_spark_datasource.py
index a708608e0..8042e75d3 100644
--- a/tests/experimental/datasources/test_spark_datasource.py
+++ b/tests/experimental/datasources/test_spark_datasource.py
@@ -27,7 +27,18 @@ def spark_datasource(test_backends) -> SparkDatasource:
     if "SparkDFDataset" not in test_backends:
         pytest.skip("No spark backend selected.")
 
-    return SparkDatasource(name="spark_datasource")
+    base_directory_rel_path = pathlib.Path(
+        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
+    )
+    base_directory_abs_path = (
+        pathlib.Path(__file__)
+        .parent.joinpath(base_directory_rel_path)
+        .resolve(strict=True)
+    )
+    return SparkDatasource(
+        name="spark_datasource",
+        base_directory=base_directory_abs_path,
+    )
 
 
 @pytest.fixture
@@ -48,17 +59,15 @@ def test_construct_spark_datasource(spark_datasource: SparkDatasource):
 
 @pytest.mark.unit
 def test_add_csv_asset_to_datasource(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",
         header=True,
         infer_schema=True,
     )
     assert asset.name == "csv_asset"
-    assert asset.base_directory == csv_path
     assert asset.regex.match("random string") is None
     assert asset.regex.match("yellow_tripdata_sample_11D1-22.csv") is None
     m1 = asset.regex.match("yellow_tripdata_sample_1111-22.csv")
@@ -66,15 +75,13 @@ def test_add_csv_asset_to_datasource(
 
 
 @pytest.mark.unit
-def test_construct_csv_asset_directly(csv_path: pathlib.Path):
+def test_construct_csv_asset_directly():
     # noinspection PyTypeChecker
     asset = CSVSparkAsset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",  # Ignoring IDE warning (type declarations are consistent).
     )
     assert asset.name == "csv_asset"
-    assert asset.base_directory == csv_path
     assert asset.regex.match("random string") is None
     assert asset.regex.match("yellow_tripdata_sample_11D1-22.csv") is None
     m1 = asset.regex.match("yellow_tripdata_sample_1111-22.csv")
@@ -83,11 +90,10 @@ def test_construct_csv_asset_directly(csv_path: pathlib.Path):
 
 @pytest.mark.unit
 def test_csv_asset_with_regex_unnamed_parameters(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -98,11 +104,10 @@ def test_csv_asset_with_regex_unnamed_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_regex_named_parameters(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -113,11 +118,10 @@ def test_csv_asset_with_regex_named_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_some_regex_named_parameters(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(?P<month>\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -128,11 +132,10 @@ def test_csv_asset_with_some_regex_named_parameters(
 
 @pytest.mark.unit
 def test_csv_asset_with_non_string_regex_named_parameters(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(\d{4})-(?P<month>\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -145,11 +148,10 @@ def test_csv_asset_with_non_string_regex_named_parameters(
 @pytest.mark.unit
 @pytest.mark.xfail(reason="temp xfail for release 0.15.47")
 def test_get_batch_list_from_fully_specified_batch_request(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -164,7 +166,8 @@ def test_get_batch_list_from_fully_specified_batch_request(
     assert batch.metadata == {
         "year": "2018",
         "month": "04",
-        "base_directory": asset.base_directory / "yellow_tripdata_sample_2018-04.csv",
+        "base_directory": spark_datasource.base_directory
+        / "yellow_tripdata_sample_2018-04.csv",
     }
     assert batch.id == "spark_datasource-csv_asset-year_2018-month_04"
 
@@ -172,12 +175,13 @@ def test_get_batch_list_from_fully_specified_batch_request(
 @pytest.mark.unit
 @pytest.mark.xfail(reason="temp xfail for release 0.15.47")
 def test_get_batch_list_from_partially_specified_batch_request(
-    spark_datasource: SparkDatasource, csv_path: pathlib.Path
+    spark_datasource: SparkDatasource,
 ):
     # Verify test directory has files that don't match what we will query for
     file_name: PathStr
     all_files: List[str] = [
-        file_name.stem for file_name in list(pathlib.Path(csv_path).iterdir())
+        file_name.stem
+        for file_name in list(pathlib.Path(spark_datasource.base_directory).iterdir())
     ]
     # assert there are files that are not csv files
     assert any([not file_name.endswith("csv") for file_name in all_files])
@@ -189,7 +193,6 @@ def test_get_batch_list_from_partially_specified_batch_request(
 
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
         header=True,
         infer_schema=True,
@@ -243,7 +246,6 @@ def test_get_batch_list_from_partially_specified_batch_request(
 )
 def test_spark_sorter(
     spark_datasource: SparkDatasource,
-    csv_path: pathlib.Path,
     order_by: BatchSortersDefinition,
 ):
     # Verify test directory has files we expect
@@ -251,7 +253,8 @@ def test_spark_sorter(
     months = [format(m, "02d") for m in range(1, 13)]
     file_name: PathStr
     all_files: List[str] = [
-        file_name.stem for file_name in list(pathlib.Path(csv_path).iterdir())
+        file_name.stem
+        for file_name in list(pathlib.Path(spark_datasource.base_directory).iterdir())
     ]
     # assert there are 12 files for each year
     for year in years:
@@ -264,7 +267,6 @@ def test_spark_sorter(
 
     asset = spark_datasource.add_csv_asset(
         name="csv_asset",
-        base_directory=csv_path,
         regex=r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv",
         order_by=order_by,
     )
@@ -300,39 +302,24 @@ def test_spark_sorter(
             assert metadata[key2] == range2
 
 
-def bad_base_directory_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
-    base_directory = pathlib.Path("/this/path/is/not/here")
-    regex = re.compile(r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
-    test_connection_error = TestConnectionError(
-        f"Path: {base_directory.resolve()} does not exist."
-    )
-    return base_directory, regex, test_connection_error
-
-
-def bad_regex_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
-    relative_path = pathlib.Path(
-        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
-    )
-    base_directory = (
-        pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
-    )
+def bad_regex_config(csv_path: pathlib.Path) -> tuple[re.Pattern, TestConnectionError]:
     regex = re.compile(r"green_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
     test_connection_error = TestConnectionError(
-        f"No file at path: {base_directory.resolve()} matched the regex: {regex.pattern}",
+        f"No file at path: {csv_path.resolve()} matched the regex: {regex.pattern}",
     )
-    return base_directory, regex, test_connection_error
+    return regex, test_connection_error
 
 
-@pytest.fixture(params=[bad_base_directory_config, bad_regex_config])
+@pytest.fixture(params=[bad_regex_config])
 def datasource_test_connection_error_messages(
-    spark_datasource: SparkDatasource, request
+    csv_path: pathlib.Path, spark_datasource: SparkDatasource, request
 ) -> tuple[SparkDatasource, TestConnectionError]:
-    base_directory, regex, test_connection_error = request.param()
+    regex, test_connection_error = request.param(csv_path=csv_path)
     csv_spark_asset = CSVSparkAsset(
         name="csv_spark_asset",
-        base_directory=base_directory,
         regex=regex,
     )
+    csv_spark_asset._datasource = spark_datasource
     spark_datasource.assets = {"csv_spark_asset": csv_spark_asset}
     return spark_datasource, test_connection_error