We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
S3 prefixes get broken when running on Windows
**Describe the bug**
On Windows when referencing an S3 data source, the prefix is suffixed with ``\`` instead of ``/``, resulting in query always failing.

**To Reproduce**
Create a ``InferredAssetS3DataConnector`` data connector, and pass any valid prefix (probably needs to be non-empty). Then, attempt to load a file from that connector.

**Expected behavior**
The file gets loaded properly.

**Error**
```
...\lib\site-packages\great_expectations\datasource\data_connector\util.py in list_s3_keys(s3, query_options, iterator_dict, recursive)
    452 
    453     if not any(key in s3_objects_info for key in ["Contents", "CommonPrefixes"]):
--> 454         raise ValueError("S3 query may not have been configured correctly.")
    455 
    456     if "Contents" in s3_objects_info:

ValueError: S3 query may not have been configured correctly.
```

**Fix**
In ``inferred_asset_s3_data_connector.py``, line 82 is:

```
        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)
```

This ensures that the file path is always a folder, by appending a slash to the end. However, on Windows, the slash is ``\``, and thus you end up with a prefix like ``path/to/some/folder\``, which results in an S3 query with no results, hence the above stacktrace.

The fix is to ensure that only the correct slash can be appended to the prefix, regardless of the host OS. For example:

```
        self._prefix = prefix.rstrip('/') + '/'
```

or something to that effect.

**Environment (please complete the following information):**
 - Operating System: Windows
 - Great Expectations Version: 0.14.1


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py`
Content:
```
1 import logging
2 import os
3 from typing import List, Optional
4 
5 try:
6     import boto3
7 except ImportError:
8     boto3 = None
9 
10 from great_expectations.core.batch import BatchDefinition
11 from great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec
12 from great_expectations.datasource.data_connector.asset import Asset
13 from great_expectations.datasource.data_connector.configured_asset_file_path_data_connector import (
14     ConfiguredAssetFilePathDataConnector,
15 )
16 from great_expectations.datasource.data_connector.file_path_data_connector import (
17     FilePathDataConnector,
18 )
19 from great_expectations.datasource.data_connector.util import list_s3_keys
20 from great_expectations.execution_engine import ExecutionEngine
21 
22 logger = logging.getLogger(__name__)
23 
24 
25 class ConfiguredAssetS3DataConnector(ConfiguredAssetFilePathDataConnector):
26     """
27     Extension of ConfiguredAssetFilePathDataConnector used to connect to S3
28 
29     DataConnectors produce identifying information, called "batch_spec" that ExecutionEngines
30     can use to get individual batches of data. They add flexibility in how to obtain data
31     such as with time-based partitioning, downsampling, or other techniques appropriate
32     for the Datasource.
33 
34     The ConfiguredAssetS3DataConnector is one of two classes (InferredAssetS3DataConnector being the
35     other one) designed for connecting to data on S3.
36 
37     A ConfiguredAssetS3DataConnector requires an explicit listing of each DataAsset you want to connect to.
38     This allows more fine-tuning, but also requires more setup.
39     """
40 
41     def __init__(
42         self,
43         name: str,
44         datasource_name: str,
45         bucket: str,
46         assets: dict,
47         execution_engine: Optional[ExecutionEngine] = None,
48         default_regex: Optional[dict] = None,
49         sorters: Optional[list] = None,
50         prefix: str = "",
51         delimiter: str = "/",
52         max_keys: int = 1000,
53         boto3_options: Optional[dict] = None,
54         batch_spec_passthrough: Optional[dict] = None,
55     ):
56         """
57         ConfiguredAssetDataConnector for connecting to S3.
58 
59         Args:
60             name (str): required name for DataConnector
61             datasource_name (str): required name for datasource
62             bucket (str): bucket for S3
63             assets (dict): dict of asset configuration (required for ConfiguredAssetDataConnector)
64             execution_engine (ExecutionEngine): optional reference to ExecutionEngine
65             default_regex (dict): optional regex configuration for filtering data_references
66             sorters (list): optional list of sorters for sorting data_references
67             prefix (str): S3 prefix
68             delimiter (str): S3 delimiter
69             max_keys (int): S3 max_keys (default is 1000)
70             boto3_options (dict): optional boto3 options
71             batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec
72         """
73         logger.debug(f'Constructing ConfiguredAssetS3DataConnector "{name}".')
74 
75         super().__init__(
76             name=name,
77             datasource_name=datasource_name,
78             execution_engine=execution_engine,
79             assets=assets,
80             default_regex=default_regex,
81             sorters=sorters,
82             batch_spec_passthrough=batch_spec_passthrough,
83         )
84         self._bucket = bucket
85         self._prefix = FilePathDataConnector.sanitize_prefix(prefix)
86         self._delimiter = delimiter
87         self._max_keys = max_keys
88 
89         if boto3_options is None:
90             boto3_options = {}
91 
92         try:
93             self._s3 = boto3.client("s3", **boto3_options)
94         except (TypeError, AttributeError):
95             raise ImportError(
96                 "Unable to load boto3 (it is required for ConfiguredAssetS3DataConnector)."
97             )
98 
99     def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:
100         """
101         Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.
102 
103         Args:
104             batch_definition (BatchDefinition): to be used to build batch_spec
105 
106         Returns:
107             BatchSpec built from batch_definition
108         """
109         batch_spec: PathBatchSpec = super().build_batch_spec(
110             batch_definition=batch_definition
111         )
112         return S3BatchSpec(batch_spec)
113 
114     def _get_data_reference_list_for_asset(self, asset: Optional[Asset]) -> List[str]:
115         query_options: dict = {
116             "Bucket": self._bucket,
117             "Prefix": self._prefix,
118             "Delimiter": self._delimiter,
119             "MaxKeys": self._max_keys,
120         }
121         if asset is not None:
122             if asset.bucket:
123                 query_options["Bucket"] = asset.bucket
124             if asset.prefix:
125                 query_options["Prefix"] = asset.prefix
126             if asset.delimiter:
127                 query_options["Delimiter"] = asset.delimiter
128             if asset.max_keys:
129                 query_options["MaxKeys"] = asset.max_keys
130 
131         path_list: List[str] = [
132             key
133             for key in list_s3_keys(
134                 s3=self._s3,
135                 query_options=query_options,
136                 iterator_dict={},
137                 recursive=False,
138             )
139         ]
140         return path_list
141 
142     def _get_full_file_path_for_asset(
143         self, path: str, asset: Optional[Asset] = None
144     ) -> str:
145         # asset isn't used in this method.
146         # It's only kept for compatibility with parent methods.
147         template_arguments: dict = {
148             "bucket": self._bucket,
149             "path": path,
150         }
151         return self.execution_engine.resolve_data_reference(
152             data_connector_name=self.__class__.__name__,
153             template_arguments=template_arguments,
154         )
155
```

Path: `great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py`
Content:
```
1 import logging
2 from typing import List, Optional
3 
4 import great_expectations.exceptions as ge_exceptions
5 from great_expectations.core.batch import BatchDefinition
6 from great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec
7 from great_expectations.datasource.data_connector.file_path_data_connector import (
8     FilePathDataConnector,
9 )
10 
11 try:
12     import boto3
13 except ImportError:
14     boto3 = None
15 
16 from great_expectations.datasource.data_connector.inferred_asset_file_path_data_connector import (
17     InferredAssetFilePathDataConnector,
18 )
19 from great_expectations.datasource.data_connector.util import list_s3_keys
20 from great_expectations.execution_engine import ExecutionEngine
21 
22 logger = logging.getLogger(__name__)
23 
24 INVALID_S3_CHARS = ["*"]
25 
26 
27 class InferredAssetS3DataConnector(InferredAssetFilePathDataConnector):
28     """
29     Extension of InferredAssetFilePathDataConnector used to connect to S3
30 
31     The InferredAssetS3DataConnector is one of two classes (ConfiguredAssetS3DataConnector being the
32     other one) designed for connecting to filesystem-like data, more specifically files on S3. It connects to assets
33     inferred from bucket, prefix, and file name by default_regex.
34 
35     InferredAssetS3DataConnector that operates on S3 buckets and determines
36     the data_asset_name implicitly (e.g., through the combination of the regular expressions pattern and group names)
37 
38     """
39 
40     def __init__(
41         self,
42         name: str,
43         datasource_name: str,
44         bucket: str,
45         execution_engine: Optional[ExecutionEngine] = None,
46         default_regex: Optional[dict] = None,
47         sorters: Optional[list] = None,
48         prefix: str = "",
49         delimiter: str = "/",
50         max_keys: int = 1000,
51         boto3_options: Optional[dict] = None,
52         batch_spec_passthrough: Optional[dict] = None,
53     ):
54         """
55         InferredAssetS3DataConnector for connecting to S3.
56 
57         Args:
58             name (str): required name for data_connector
59             datasource_name (str): required name for datasource
60             bucket (str): bucket for S3
61             execution_engine (ExecutionEngine): optional reference to ExecutionEngine
62             default_regex (dict): optional regex configuration for filtering data_references
63             sorters (list): optional list of sorters for sorting data_references
64             prefix (str): S3 prefix
65             delimiter (str): S3 delimiter
66             max_keys (int): S3 max_keys (default is 1000)
67             boto3_options (dict): optional boto3 options
68             batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec
69         """
70         logger.debug(f'Constructing InferredAssetS3DataConnector "{name}".')
71 
72         super().__init__(
73             name=name,
74             datasource_name=datasource_name,
75             execution_engine=execution_engine,
76             default_regex=default_regex,
77             sorters=sorters,
78             batch_spec_passthrough=batch_spec_passthrough,
79         )
80 
81         self._bucket = bucket
82         self._prefix = FilePathDataConnector.sanitize_prefix(prefix)
83         self._delimiter = delimiter
84         self._max_keys = max_keys
85 
86         if boto3_options is None:
87             boto3_options = {}
88 
89         try:
90             self._s3 = boto3.client("s3", **boto3_options)
91         except (TypeError, AttributeError):
92             raise ImportError(
93                 "Unable to load boto3 (it is required for InferredAssetS3DataConnector)."
94             )
95 
96     def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:
97         """
98         Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.
99 
100         Args:
101             batch_definition (BatchDefinition): to be used to build batch_spec
102 
103         Returns:
104             BatchSpec built from batch_definition
105         """
106         batch_spec: PathBatchSpec = super().build_batch_spec(
107             batch_definition=batch_definition
108         )
109         return S3BatchSpec(batch_spec)
110 
111     def _get_data_reference_list(
112         self, data_asset_name: Optional[str] = None
113     ) -> List[str]:
114         """
115         List objects in the underlying data store to create a list of data_references.
116 
117         This method is used to refresh the cache.
118         """
119         query_options: dict = {
120             "Bucket": self._bucket,
121             "Prefix": self._prefix,
122             "Delimiter": self._delimiter,
123             "MaxKeys": self._max_keys,
124         }
125 
126         path_list: List[str] = [
127             key
128             for key in list_s3_keys(
129                 s3=self._s3,
130                 query_options=query_options,
131                 iterator_dict={},
132                 recursive=True,
133             )
134         ]
135         return path_list
136 
137     def _get_full_file_path(
138         self,
139         path: str,
140         data_asset_name: Optional[str] = None,
141     ) -> str:
142         # data_asset_name isn't used in this method.
143         # It's only kept for compatibility with parent methods.
144         _check_valid_s3_path(path)
145         template_arguments: dict = {
146             "bucket": self._bucket,
147             "path": path,
148         }
149         return self.execution_engine.resolve_data_reference(
150             data_connector_name=self.__class__.__name__,
151             template_arguments=template_arguments,
152         )
153 
154 
155 def _check_valid_s3_path(
156     path: str,
157 ) -> None:
158     """Performs a basic check for validity of the S3 path"""
159     bad_chars: list = [c for c in INVALID_S3_CHARS if c in path]
160     if len(bad_chars) > 0:
161         msg: str = (
162             f"The parsed S3 path={path} contains the invalid characters {bad_chars}."
163             "Please make sure your regex is correct and characters are escaped."
164         )
165         if "*" in bad_chars:
166             msg += "Note: `*` is internally used to replace the regex for `.`."
167         raise ge_exceptions.ParserError(msg)
168
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py b/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py
--- a/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py
+++ b/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py
@@ -13,9 +13,6 @@
 from great_expectations.datasource.data_connector.configured_asset_file_path_data_connector import (
     ConfiguredAssetFilePathDataConnector,
 )
-from great_expectations.datasource.data_connector.file_path_data_connector import (
-    FilePathDataConnector,
-)
 from great_expectations.datasource.data_connector.util import list_s3_keys
 from great_expectations.execution_engine import ExecutionEngine
 
@@ -82,7 +79,7 @@
             batch_spec_passthrough=batch_spec_passthrough,
         )
         self._bucket = bucket
-        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)
+        self._prefix = self.sanitize_prefix_for_s3(prefix)
         self._delimiter = delimiter
         self._max_keys = max_keys
 
@@ -96,6 +93,26 @@
                 "Unable to load boto3 (it is required for ConfiguredAssetS3DataConnector)."
             )
 
+    @staticmethod
+    def sanitize_prefix_for_s3(text: str) -> str:
+        """
+        Takes in a given user-prefix and cleans it to work with file-system traversal methods
+        (i.e. add '/' to the end of a string meant to represent a directory)
+
+        Customized for S3 paths, ignoring the path separator used by the host OS
+        """
+        text = text.strip()
+        if not text:
+            return text
+
+        path_parts = text.split("/")
+        if not path_parts:  # Empty prefix
+            return text
+        elif "." in path_parts[-1]:  # File, not folder
+            return text
+        else:  # Folder, should have trailing /
+            return text.rstrip("/") + "/"
+
     def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:
         """
         Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.
diff --git a/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py b/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py
--- a/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py
+++ b/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py
@@ -4,9 +4,7 @@
 import great_expectations.exceptions as ge_exceptions
 from great_expectations.core.batch import BatchDefinition
 from great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec
-from great_expectations.datasource.data_connector.file_path_data_connector import (
-    FilePathDataConnector,
-)
+from great_expectations.datasource.data_connector import ConfiguredAssetS3DataConnector
 
 try:
     import boto3
@@ -79,7 +77,7 @@
         )
 
         self._bucket = bucket
-        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)
+        self._prefix = ConfiguredAssetS3DataConnector.sanitize_prefix_for_s3(prefix)
         self._delimiter = delimiter
         self._max_keys = max_keys

{"golden_diff": "diff --git a/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py b/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py\n--- a/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py\n+++ b/great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py\n@@ -13,9 +13,6 @@\n from great_expectations.datasource.data_connector.configured_asset_file_path_data_connector import (\n     ConfiguredAssetFilePathDataConnector,\n )\n-from great_expectations.datasource.data_connector.file_path_data_connector import (\n-    FilePathDataConnector,\n-)\n from great_expectations.datasource.data_connector.util import list_s3_keys\n from great_expectations.execution_engine import ExecutionEngine\n \n@@ -82,7 +79,7 @@\n             batch_spec_passthrough=batch_spec_passthrough,\n         )\n         self._bucket = bucket\n-        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)\n+        self._prefix = self.sanitize_prefix_for_s3(prefix)\n         self._delimiter = delimiter\n         self._max_keys = max_keys\n \n@@ -96,6 +93,26 @@\n                 \"Unable to load boto3 (it is required for ConfiguredAssetS3DataConnector).\"\n             )\n \n+    @staticmethod\n+    def sanitize_prefix_for_s3(text: str) -> str:\n+        \"\"\"\n+        Takes in a given user-prefix and cleans it to work with file-system traversal methods\n+        (i.e. add '/' to the end of a string meant to represent a directory)\n+\n+        Customized for S3 paths, ignoring the path separator used by the host OS\n+        \"\"\"\n+        text = text.strip()\n+        if not text:\n+            return text\n+\n+        path_parts = text.split(\"/\")\n+        if not path_parts:  # Empty prefix\n+            return text\n+        elif \".\" in path_parts[-1]:  # File, not folder\n+            return text\n+        else:  # Folder, should have trailing /\n+            return text.rstrip(\"/\") + \"/\"\n+\n     def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:\n         \"\"\"\n         Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.\ndiff --git a/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py b/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py\n--- a/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py\n+++ b/great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py\n@@ -4,9 +4,7 @@\n import great_expectations.exceptions as ge_exceptions\n from great_expectations.core.batch import BatchDefinition\n from great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec\n-from great_expectations.datasource.data_connector.file_path_data_connector import (\n-    FilePathDataConnector,\n-)\n+from great_expectations.datasource.data_connector import ConfiguredAssetS3DataConnector\n \n try:\n     import boto3\n@@ -79,7 +77,7 @@\n         )\n \n         self._bucket = bucket\n-        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)\n+        self._prefix = ConfiguredAssetS3DataConnector.sanitize_prefix_for_s3(prefix)\n         self._delimiter = delimiter\n         self._max_keys = max_keys\n", "issue": "S3 prefixes get broken when running on Windows\n**Describe the bug**\r\nOn Windows when referencing an S3 data source, the prefix is suffixed with ``\\`` instead of ``/``, resulting in query always failing.\r\n\r\n**To Reproduce**\r\nCreate a ``InferredAssetS3DataConnector`` data connector, and pass any valid prefix (probably needs to be non-empty). Then, attempt to load a file from that connector.\r\n\r\n**Expected behavior**\r\nThe file gets loaded properly.\r\n\r\n**Error**\r\n```\r\n...\\lib\\site-packages\\great_expectations\\datasource\\data_connector\\util.py in list_s3_keys(s3, query_options, iterator_dict, recursive)\r\n    452 \r\n    453     if not any(key in s3_objects_info for key in [\"Contents\", \"CommonPrefixes\"]):\r\n--> 454         raise ValueError(\"S3 query may not have been configured correctly.\")\r\n    455 \r\n    456     if \"Contents\" in s3_objects_info:\r\n\r\nValueError: S3 query may not have been configured correctly.\r\n```\r\n\r\n**Fix**\r\nIn ``inferred_asset_s3_data_connector.py``, line 82 is:\r\n\r\n```\r\n        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)\r\n```\r\n\r\nThis ensures that the file path is always a folder, by appending a slash to the end. However, on Windows, the slash is ``\\``, and thus you end up with a prefix like ``path/to/some/folder\\``, which results in an S3 query with no results, hence the above stacktrace.\r\n\r\nThe fix is to ensure that only the correct slash can be appended to the prefix, regardless of the host OS. For example:\r\n\r\n```\r\n        self._prefix = prefix.rstrip('/') + '/'\r\n```\r\n\r\nor something to that effect.\r\n\r\n**Environment (please complete the following information):**\r\n - Operating System: Windows\r\n - Great Expectations Version: 0.14.1\r\n\n", "before_files": [{"content": "import logging\nimport os\nfrom typing import List, Optional\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\nfrom great_expectations.core.batch import BatchDefinition\nfrom great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec\nfrom great_expectations.datasource.data_connector.asset import Asset\nfrom great_expectations.datasource.data_connector.configured_asset_file_path_data_connector import (\n    ConfiguredAssetFilePathDataConnector,\n)\nfrom great_expectations.datasource.data_connector.file_path_data_connector import (\n    FilePathDataConnector,\n)\nfrom great_expectations.datasource.data_connector.util import list_s3_keys\nfrom great_expectations.execution_engine import ExecutionEngine\n\nlogger = logging.getLogger(__name__)\n\n\nclass ConfiguredAssetS3DataConnector(ConfiguredAssetFilePathDataConnector):\n    \"\"\"\n    Extension of ConfiguredAssetFilePathDataConnector used to connect to S3\n\n    DataConnectors produce identifying information, called \"batch_spec\" that ExecutionEngines\n    can use to get individual batches of data. They add flexibility in how to obtain data\n    such as with time-based partitioning, downsampling, or other techniques appropriate\n    for the Datasource.\n\n    The ConfiguredAssetS3DataConnector is one of two classes (InferredAssetS3DataConnector being the\n    other one) designed for connecting to data on S3.\n\n    A ConfiguredAssetS3DataConnector requires an explicit listing of each DataAsset you want to connect to.\n    This allows more fine-tuning, but also requires more setup.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        bucket: str,\n        assets: dict,\n        execution_engine: Optional[ExecutionEngine] = None,\n        default_regex: Optional[dict] = None,\n        sorters: Optional[list] = None,\n        prefix: str = \"\",\n        delimiter: str = \"/\",\n        max_keys: int = 1000,\n        boto3_options: Optional[dict] = None,\n        batch_spec_passthrough: Optional[dict] = None,\n    ):\n        \"\"\"\n        ConfiguredAssetDataConnector for connecting to S3.\n\n        Args:\n            name (str): required name for DataConnector\n            datasource_name (str): required name for datasource\n            bucket (str): bucket for S3\n            assets (dict): dict of asset configuration (required for ConfiguredAssetDataConnector)\n            execution_engine (ExecutionEngine): optional reference to ExecutionEngine\n            default_regex (dict): optional regex configuration for filtering data_references\n            sorters (list): optional list of sorters for sorting data_references\n            prefix (str): S3 prefix\n            delimiter (str): S3 delimiter\n            max_keys (int): S3 max_keys (default is 1000)\n            boto3_options (dict): optional boto3 options\n            batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec\n        \"\"\"\n        logger.debug(f'Constructing ConfiguredAssetS3DataConnector \"{name}\".')\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n            assets=assets,\n            default_regex=default_regex,\n            sorters=sorters,\n            batch_spec_passthrough=batch_spec_passthrough,\n        )\n        self._bucket = bucket\n        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)\n        self._delimiter = delimiter\n        self._max_keys = max_keys\n\n        if boto3_options is None:\n            boto3_options = {}\n\n        try:\n            self._s3 = boto3.client(\"s3\", **boto3_options)\n        except (TypeError, AttributeError):\n            raise ImportError(\n                \"Unable to load boto3 (it is required for ConfiguredAssetS3DataConnector).\"\n            )\n\n    def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:\n        \"\"\"\n        Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            BatchSpec built from batch_definition\n        \"\"\"\n        batch_spec: PathBatchSpec = super().build_batch_spec(\n            batch_definition=batch_definition\n        )\n        return S3BatchSpec(batch_spec)\n\n    def _get_data_reference_list_for_asset(self, asset: Optional[Asset]) -> List[str]:\n        query_options: dict = {\n            \"Bucket\": self._bucket,\n            \"Prefix\": self._prefix,\n            \"Delimiter\": self._delimiter,\n            \"MaxKeys\": self._max_keys,\n        }\n        if asset is not None:\n            if asset.bucket:\n                query_options[\"Bucket\"] = asset.bucket\n            if asset.prefix:\n                query_options[\"Prefix\"] = asset.prefix\n            if asset.delimiter:\n                query_options[\"Delimiter\"] = asset.delimiter\n            if asset.max_keys:\n                query_options[\"MaxKeys\"] = asset.max_keys\n\n        path_list: List[str] = [\n            key\n            for key in list_s3_keys(\n                s3=self._s3,\n                query_options=query_options,\n                iterator_dict={},\n                recursive=False,\n            )\n        ]\n        return path_list\n\n    def _get_full_file_path_for_asset(\n        self, path: str, asset: Optional[Asset] = None\n    ) -> str:\n        # asset isn't used in this method.\n        # It's only kept for compatibility with parent methods.\n        template_arguments: dict = {\n            \"bucket\": self._bucket,\n            \"path\": path,\n        }\n        return self.execution_engine.resolve_data_reference(\n            data_connector_name=self.__class__.__name__,\n            template_arguments=template_arguments,\n        )\n", "path": "great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py"}, {"content": "import logging\nfrom typing import List, Optional\n\nimport great_expectations.exceptions as ge_exceptions\nfrom great_expectations.core.batch import BatchDefinition\nfrom great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec\nfrom great_expectations.datasource.data_connector.file_path_data_connector import (\n    FilePathDataConnector,\n)\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\nfrom great_expectations.datasource.data_connector.inferred_asset_file_path_data_connector import (\n    InferredAssetFilePathDataConnector,\n)\nfrom great_expectations.datasource.data_connector.util import list_s3_keys\nfrom great_expectations.execution_engine import ExecutionEngine\n\nlogger = logging.getLogger(__name__)\n\nINVALID_S3_CHARS = [\"*\"]\n\n\nclass InferredAssetS3DataConnector(InferredAssetFilePathDataConnector):\n    \"\"\"\n    Extension of InferredAssetFilePathDataConnector used to connect to S3\n\n    The InferredAssetS3DataConnector is one of two classes (ConfiguredAssetS3DataConnector being the\n    other one) designed for connecting to filesystem-like data, more specifically files on S3. It connects to assets\n    inferred from bucket, prefix, and file name by default_regex.\n\n    InferredAssetS3DataConnector that operates on S3 buckets and determines\n    the data_asset_name implicitly (e.g., through the combination of the regular expressions pattern and group names)\n\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        bucket: str,\n        execution_engine: Optional[ExecutionEngine] = None,\n        default_regex: Optional[dict] = None,\n        sorters: Optional[list] = None,\n        prefix: str = \"\",\n        delimiter: str = \"/\",\n        max_keys: int = 1000,\n        boto3_options: Optional[dict] = None,\n        batch_spec_passthrough: Optional[dict] = None,\n    ):\n        \"\"\"\n        InferredAssetS3DataConnector for connecting to S3.\n\n        Args:\n            name (str): required name for data_connector\n            datasource_name (str): required name for datasource\n            bucket (str): bucket for S3\n            execution_engine (ExecutionEngine): optional reference to ExecutionEngine\n            default_regex (dict): optional regex configuration for filtering data_references\n            sorters (list): optional list of sorters for sorting data_references\n            prefix (str): S3 prefix\n            delimiter (str): S3 delimiter\n            max_keys (int): S3 max_keys (default is 1000)\n            boto3_options (dict): optional boto3 options\n            batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec\n        \"\"\"\n        logger.debug(f'Constructing InferredAssetS3DataConnector \"{name}\".')\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n            default_regex=default_regex,\n            sorters=sorters,\n            batch_spec_passthrough=batch_spec_passthrough,\n        )\n\n        self._bucket = bucket\n        self._prefix = FilePathDataConnector.sanitize_prefix(prefix)\n        self._delimiter = delimiter\n        self._max_keys = max_keys\n\n        if boto3_options is None:\n            boto3_options = {}\n\n        try:\n            self._s3 = boto3.client(\"s3\", **boto3_options)\n        except (TypeError, AttributeError):\n            raise ImportError(\n                \"Unable to load boto3 (it is required for InferredAssetS3DataConnector).\"\n            )\n\n    def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:\n        \"\"\"\n        Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            BatchSpec built from batch_definition\n        \"\"\"\n        batch_spec: PathBatchSpec = super().build_batch_spec(\n            batch_definition=batch_definition\n        )\n        return S3BatchSpec(batch_spec)\n\n    def _get_data_reference_list(\n        self, data_asset_name: Optional[str] = None\n    ) -> List[str]:\n        \"\"\"\n        List objects in the underlying data store to create a list of data_references.\n\n        This method is used to refresh the cache.\n        \"\"\"\n        query_options: dict = {\n            \"Bucket\": self._bucket,\n            \"Prefix\": self._prefix,\n            \"Delimiter\": self._delimiter,\n            \"MaxKeys\": self._max_keys,\n        }\n\n        path_list: List[str] = [\n            key\n            for key in list_s3_keys(\n                s3=self._s3,\n                query_options=query_options,\n                iterator_dict={},\n                recursive=True,\n            )\n        ]\n        return path_list\n\n    def _get_full_file_path(\n        self,\n        path: str,\n        data_asset_name: Optional[str] = None,\n    ) -> str:\n        # data_asset_name isn't used in this method.\n        # It's only kept for compatibility with parent methods.\n        _check_valid_s3_path(path)\n        template_arguments: dict = {\n            \"bucket\": self._bucket,\n            \"path\": path,\n        }\n        return self.execution_engine.resolve_data_reference(\n            data_connector_name=self.__class__.__name__,\n            template_arguments=template_arguments,\n        )\n\n\ndef _check_valid_s3_path(\n    path: str,\n) -> None:\n    \"\"\"Performs a basic check for validity of the S3 path\"\"\"\n    bad_chars: list = [c for c in INVALID_S3_CHARS if c in path]\n    if len(bad_chars) > 0:\n        msg: str = (\n            f\"The parsed S3 path={path} contains the invalid characters {bad_chars}.\"\n            \"Please make sure your regex is correct and characters are escaped.\"\n        )\n        if \"*\" in bad_chars:\n            msg += \"Note: `*` is internally used to replace the regex for `.`.\"\n        raise ge_exceptions.ParserError(msg)\n", "path": "great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py"}], "after_files": [{"content": "import logging\nimport os\nfrom typing import List, Optional\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\nfrom great_expectations.core.batch import BatchDefinition\nfrom great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec\nfrom great_expectations.datasource.data_connector.asset import Asset\nfrom great_expectations.datasource.data_connector.configured_asset_file_path_data_connector import (\n    ConfiguredAssetFilePathDataConnector,\n)\nfrom great_expectations.datasource.data_connector.util import list_s3_keys\nfrom great_expectations.execution_engine import ExecutionEngine\n\nlogger = logging.getLogger(__name__)\n\n\nclass ConfiguredAssetS3DataConnector(ConfiguredAssetFilePathDataConnector):\n    \"\"\"\n    Extension of ConfiguredAssetFilePathDataConnector used to connect to S3\n\n    DataConnectors produce identifying information, called \"batch_spec\" that ExecutionEngines\n    can use to get individual batches of data. They add flexibility in how to obtain data\n    such as with time-based partitioning, downsampling, or other techniques appropriate\n    for the Datasource.\n\n    The ConfiguredAssetS3DataConnector is one of two classes (InferredAssetS3DataConnector being the\n    other one) designed for connecting to data on S3.\n\n    A ConfiguredAssetS3DataConnector requires an explicit listing of each DataAsset you want to connect to.\n    This allows more fine-tuning, but also requires more setup.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        bucket: str,\n        assets: dict,\n        execution_engine: Optional[ExecutionEngine] = None,\n        default_regex: Optional[dict] = None,\n        sorters: Optional[list] = None,\n        prefix: str = \"\",\n        delimiter: str = \"/\",\n        max_keys: int = 1000,\n        boto3_options: Optional[dict] = None,\n        batch_spec_passthrough: Optional[dict] = None,\n    ):\n        \"\"\"\n        ConfiguredAssetDataConnector for connecting to S3.\n\n        Args:\n            name (str): required name for DataConnector\n            datasource_name (str): required name for datasource\n            bucket (str): bucket for S3\n            assets (dict): dict of asset configuration (required for ConfiguredAssetDataConnector)\n            execution_engine (ExecutionEngine): optional reference to ExecutionEngine\n            default_regex (dict): optional regex configuration for filtering data_references\n            sorters (list): optional list of sorters for sorting data_references\n            prefix (str): S3 prefix\n            delimiter (str): S3 delimiter\n            max_keys (int): S3 max_keys (default is 1000)\n            boto3_options (dict): optional boto3 options\n            batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec\n        \"\"\"\n        logger.debug(f'Constructing ConfiguredAssetS3DataConnector \"{name}\".')\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n            assets=assets,\n            default_regex=default_regex,\n            sorters=sorters,\n            batch_spec_passthrough=batch_spec_passthrough,\n        )\n        self._bucket = bucket\n        self._prefix = self.sanitize_prefix_for_s3(prefix)\n        self._delimiter = delimiter\n        self._max_keys = max_keys\n\n        if boto3_options is None:\n            boto3_options = {}\n\n        try:\n            self._s3 = boto3.client(\"s3\", **boto3_options)\n        except (TypeError, AttributeError):\n            raise ImportError(\n                \"Unable to load boto3 (it is required for ConfiguredAssetS3DataConnector).\"\n            )\n\n    @staticmethod\n    def sanitize_prefix_for_s3(text: str) -> str:\n        \"\"\"\n        Takes in a given user-prefix and cleans it to work with file-system traversal methods\n        (i.e. add '/' to the end of a string meant to represent a directory)\n\n        Customized for S3 paths, ignoring the path separator used by the host OS\n        \"\"\"\n        text = text.strip()\n        if not text:\n            return text\n\n        path_parts = text.split(\"/\")\n        if not path_parts:  # Empty prefix\n            return text\n        elif \".\" in path_parts[-1]:  # File, not folder\n            return text\n        else:  # Folder, should have trailing /\n            return text.rstrip(\"/\") + \"/\"\n\n    def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:\n        \"\"\"\n        Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            BatchSpec built from batch_definition\n        \"\"\"\n        batch_spec: PathBatchSpec = super().build_batch_spec(\n            batch_definition=batch_definition\n        )\n        return S3BatchSpec(batch_spec)\n\n    def _get_data_reference_list_for_asset(self, asset: Optional[Asset]) -> List[str]:\n        query_options: dict = {\n            \"Bucket\": self._bucket,\n            \"Prefix\": self._prefix,\n            \"Delimiter\": self._delimiter,\n            \"MaxKeys\": self._max_keys,\n        }\n        if asset is not None:\n            if asset.bucket:\n                query_options[\"Bucket\"] = asset.bucket\n            if asset.prefix:\n                query_options[\"Prefix\"] = asset.prefix\n            if asset.delimiter:\n                query_options[\"Delimiter\"] = asset.delimiter\n            if asset.max_keys:\n                query_options[\"MaxKeys\"] = asset.max_keys\n\n        path_list: List[str] = [\n            key\n            for key in list_s3_keys(\n                s3=self._s3,\n                query_options=query_options,\n                iterator_dict={},\n                recursive=False,\n            )\n        ]\n        return path_list\n\n    def _get_full_file_path_for_asset(\n        self, path: str, asset: Optional[Asset] = None\n    ) -> str:\n        # asset isn't used in this method.\n        # It's only kept for compatibility with parent methods.\n        template_arguments: dict = {\n            \"bucket\": self._bucket,\n            \"path\": path,\n        }\n        return self.execution_engine.resolve_data_reference(\n            data_connector_name=self.__class__.__name__,\n            template_arguments=template_arguments,\n        )\n", "path": "great_expectations/datasource/data_connector/configured_asset_s3_data_connector.py"}, {"content": "import logging\nfrom typing import List, Optional\n\nimport great_expectations.exceptions as ge_exceptions\nfrom great_expectations.core.batch import BatchDefinition\nfrom great_expectations.core.batch_spec import PathBatchSpec, S3BatchSpec\nfrom great_expectations.datasource.data_connector import ConfiguredAssetS3DataConnector\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\nfrom great_expectations.datasource.data_connector.inferred_asset_file_path_data_connector import (\n    InferredAssetFilePathDataConnector,\n)\nfrom great_expectations.datasource.data_connector.util import list_s3_keys\nfrom great_expectations.execution_engine import ExecutionEngine\n\nlogger = logging.getLogger(__name__)\n\nINVALID_S3_CHARS = [\"*\"]\n\n\nclass InferredAssetS3DataConnector(InferredAssetFilePathDataConnector):\n    \"\"\"\n    Extension of InferredAssetFilePathDataConnector used to connect to S3\n\n    The InferredAssetS3DataConnector is one of two classes (ConfiguredAssetS3DataConnector being the\n    other one) designed for connecting to filesystem-like data, more specifically files on S3. It connects to assets\n    inferred from bucket, prefix, and file name by default_regex.\n\n    InferredAssetS3DataConnector that operates on S3 buckets and determines\n    the data_asset_name implicitly (e.g., through the combination of the regular expressions pattern and group names)\n\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        bucket: str,\n        execution_engine: Optional[ExecutionEngine] = None,\n        default_regex: Optional[dict] = None,\n        sorters: Optional[list] = None,\n        prefix: str = \"\",\n        delimiter: str = \"/\",\n        max_keys: int = 1000,\n        boto3_options: Optional[dict] = None,\n        batch_spec_passthrough: Optional[dict] = None,\n    ):\n        \"\"\"\n        InferredAssetS3DataConnector for connecting to S3.\n\n        Args:\n            name (str): required name for data_connector\n            datasource_name (str): required name for datasource\n            bucket (str): bucket for S3\n            execution_engine (ExecutionEngine): optional reference to ExecutionEngine\n            default_regex (dict): optional regex configuration for filtering data_references\n            sorters (list): optional list of sorters for sorting data_references\n            prefix (str): S3 prefix\n            delimiter (str): S3 delimiter\n            max_keys (int): S3 max_keys (default is 1000)\n            boto3_options (dict): optional boto3 options\n            batch_spec_passthrough (dict): dictionary with keys that will be added directly to batch_spec\n        \"\"\"\n        logger.debug(f'Constructing InferredAssetS3DataConnector \"{name}\".')\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n            default_regex=default_regex,\n            sorters=sorters,\n            batch_spec_passthrough=batch_spec_passthrough,\n        )\n\n        self._bucket = bucket\n        self._prefix = ConfiguredAssetS3DataConnector.sanitize_prefix_for_s3(prefix)\n        self._delimiter = delimiter\n        self._max_keys = max_keys\n\n        if boto3_options is None:\n            boto3_options = {}\n\n        try:\n            self._s3 = boto3.client(\"s3\", **boto3_options)\n        except (TypeError, AttributeError):\n            raise ImportError(\n                \"Unable to load boto3 (it is required for InferredAssetS3DataConnector).\"\n            )\n\n    def build_batch_spec(self, batch_definition: BatchDefinition) -> S3BatchSpec:\n        \"\"\"\n        Build BatchSpec from batch_definition by calling DataConnector's build_batch_spec function.\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            BatchSpec built from batch_definition\n        \"\"\"\n        batch_spec: PathBatchSpec = super().build_batch_spec(\n            batch_definition=batch_definition\n        )\n        return S3BatchSpec(batch_spec)\n\n    def _get_data_reference_list(\n        self, data_asset_name: Optional[str] = None\n    ) -> List[str]:\n        \"\"\"\n        List objects in the underlying data store to create a list of data_references.\n\n        This method is used to refresh the cache.\n        \"\"\"\n        query_options: dict = {\n            \"Bucket\": self._bucket,\n            \"Prefix\": self._prefix,\n            \"Delimiter\": self._delimiter,\n            \"MaxKeys\": self._max_keys,\n        }\n\n        path_list: List[str] = [\n            key\n            for key in list_s3_keys(\n                s3=self._s3,\n                query_options=query_options,\n                iterator_dict={},\n                recursive=True,\n            )\n        ]\n        return path_list\n\n    def _get_full_file_path(\n        self,\n        path: str,\n        data_asset_name: Optional[str] = None,\n    ) -> str:\n        # data_asset_name isn't used in this method.\n        # It's only kept for compatibility with parent methods.\n        _check_valid_s3_path(path)\n        template_arguments: dict = {\n            \"bucket\": self._bucket,\n            \"path\": path,\n        }\n        return self.execution_engine.resolve_data_reference(\n            data_connector_name=self.__class__.__name__,\n            template_arguments=template_arguments,\n        )\n\n\ndef _check_valid_s3_path(\n    path: str,\n) -> None:\n    \"\"\"Performs a basic check for validity of the S3 path\"\"\"\n    bad_chars: list = [c for c in INVALID_S3_CHARS if c in path]\n    if len(bad_chars) > 0:\n        msg: str = (\n            f\"The parsed S3 path={path} contains the invalid characters {bad_chars}.\"\n            \"Please make sure your regex is correct and characters are escaped.\"\n        )\n        if \"*\" in bad_chars:\n            msg += \"Note: `*` is internally used to replace the regex for `.`.\"\n        raise ge_exceptions.ParserError(msg)\n", "path": "great_expectations/datasource/data_connector/inferred_asset_s3_data_connector.py"}]}