We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Asset Name is not populated in Data Docs for SimpleSqlalchemy data source
**Describe the bug**
When using a SimpleSqlalchemy data source, the Asset Name column is not populated in the Validations index of the data docs. SImilarly, the Data Asset entry in the corresponding Validation result is None. 

**To Reproduce**
Steps to reproduce the behavior:
1. Configure a SimpleSqlalchemy datasource
```yaml
datasources:
  db:
    credentials: ${my_postgres_db}
    class_name: SimpleSqlalchemyDatasource
    introspection:
      whole_table: {}
``` 
2. Configure an expectation suite
3. Configure a SimpleCheckpoint  
```yaml
name: my_simple_checkpoint
config_version: 1.0
class_name: SimpleCheckpoint
expectation_suite_name: ge_validations_store.warning
validations:
  - batch_request:
      datasource_name: db
      data_connector_name: whole_table
      data_asset_name: ge_validations_store
      partition_request:
        index: -1
```
4. Run checkpoint my_simple_checkpoint
5. The generated data docs do not contain properly populated data asset entries

**Expected behavior**
The generated data docs should be populated the same way when using the v3 API as when using the v2 API

**Environment (please complete the following information):**
 - Operating System: Linux, MacOS
 - Great Expectations Version: develop branch


--- 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_sql_data_connector.py`
Content:
```
1 from typing import Dict, List, Optional
2 
3 from great_expectations.core.batch import (
4     BatchDefinition,
5     BatchRequest,
6     PartitionDefinition,
7 )
8 from great_expectations.datasource.data_connector.data_connector import DataConnector
9 from great_expectations.datasource.data_connector.util import (
10     batch_definition_matches_batch_request,
11 )
12 from great_expectations.execution_engine import ExecutionEngine
13 
14 try:
15     import sqlalchemy as sa
16 except ImportError:
17     sa = None
18 
19 
20 class ConfiguredAssetSqlDataConnector(DataConnector):
21     """
22     A DataConnector that requires explicit listing of SQL tables you want to connect to.
23 
24     Args:
25         name (str): The name of this DataConnector
26         datasource_name (str): The name of the Datasource that contains it
27         execution_engine (ExecutionEngine): An ExecutionEngine
28         data_assets (str): data_assets
29     """
30 
31     def __init__(
32         self,
33         name: str,
34         datasource_name: str,
35         execution_engine: Optional[ExecutionEngine] = None,
36         data_assets: Optional[Dict[str, dict]] = None,
37     ):
38         if data_assets is None:
39             data_assets = {}
40         self._data_assets = data_assets
41 
42         super().__init__(
43             name=name,
44             datasource_name=datasource_name,
45             execution_engine=execution_engine,
46         )
47 
48     @property
49     def data_assets(self) -> Dict[str, dict]:
50         return self._data_assets
51 
52     def add_data_asset(
53         self,
54         name: str,
55         config: dict,
56     ):
57         """
58         Add data_asset to DataConnector using data_asset name as key, and data_asset configuration as value.
59         """
60         self._data_assets[name] = config
61 
62     def _get_partition_definition_list_from_data_asset_config(
63         self,
64         data_asset_name,
65         data_asset_config,
66     ):
67         if "table_name" in data_asset_config:
68             table_name = data_asset_config["table_name"]
69         else:
70             table_name = data_asset_name
71 
72         if "splitter_method" in data_asset_config:
73             splitter_fn = getattr(self, data_asset_config["splitter_method"])
74             split_query = splitter_fn(
75                 table_name=table_name, **data_asset_config["splitter_kwargs"]
76             )
77 
78             rows = self._execution_engine.engine.execute(split_query).fetchall()
79 
80             # Zip up split parameters with column names
81             column_names = self._get_column_names_from_splitter_kwargs(
82                 data_asset_config["splitter_kwargs"]
83             )
84             partition_definition_list = [dict(zip(column_names, row)) for row in rows]
85 
86         else:
87             partition_definition_list = [{}]
88 
89         return partition_definition_list
90 
91     def _refresh_data_references_cache(self):
92         self._data_references_cache = {}
93 
94         for data_asset_name in self.data_assets:
95             data_asset = self.data_assets[data_asset_name]
96             partition_definition_list = (
97                 self._get_partition_definition_list_from_data_asset_config(
98                     data_asset_name,
99                     data_asset,
100                 )
101             )
102 
103             # TODO Abe 20201029 : Apply sorters to partition_definition_list here
104             # TODO Will 20201102 : add sorting code here
105             self._data_references_cache[data_asset_name] = partition_definition_list
106 
107     def _get_column_names_from_splitter_kwargs(self, splitter_kwargs) -> List[str]:
108         column_names: List[str] = []
109 
110         if "column_names" in splitter_kwargs:
111             column_names = splitter_kwargs["column_names"]
112         elif "column_name" in splitter_kwargs:
113             column_names = [splitter_kwargs["column_name"]]
114 
115         return column_names
116 
117     def get_available_data_asset_names(self):
118         """
119         Return the list of asset names known by this DataConnector.
120 
121         Returns:
122             A list of available names
123         """
124         return list(self.data_assets.keys())
125 
126     def get_unmatched_data_references(self) -> List[str]:
127         """
128         Returns the list of data_references unmatched by configuration by looping through items in _data_references_cache
129         and returning data_reference that do not have an associated data_asset.
130 
131         Returns:
132             list of data_references that are not matched by configuration.
133         """
134         return []
135 
136     def get_batch_definition_list_from_batch_request(self, batch_request: BatchRequest):
137         self._validate_batch_request(batch_request=batch_request)
138 
139         if len(self._data_references_cache) == 0:
140             self._refresh_data_references_cache()
141 
142         batch_definition_list: List[BatchDefinition] = []
143         try:
144             sub_cache = self._data_references_cache[batch_request.data_asset_name]
145         except KeyError as e:
146             raise KeyError(
147                 f"data_asset_name {batch_request.data_asset_name} is not recognized."
148             )
149 
150         for partition_definition in sub_cache:
151             batch_definition: BatchDefinition = BatchDefinition(
152                 datasource_name=self.datasource_name,
153                 data_connector_name=self.name,
154                 data_asset_name=batch_request.data_asset_name,
155                 partition_definition=PartitionDefinition(partition_definition),
156             )
157             if batch_definition_matches_batch_request(batch_definition, batch_request):
158                 batch_definition_list.append(batch_definition)
159 
160         return batch_definition_list
161 
162     def _get_data_reference_list_from_cache_by_data_asset_name(
163         self, data_asset_name: str
164     ) -> List[str]:
165         return self._data_references_cache[data_asset_name]
166 
167     def _map_data_reference_to_batch_definition_list(
168         self, data_reference, data_asset_name: Optional[str] = None  #: Any,
169     ) -> Optional[List[BatchDefinition]]:
170         # Note: This is a bit hacky, but it works. In sql_data_connectors, data references *are* dictionaries,
171         # allowing us to invoke `PartitionDefinition(data_reference)`
172         return [
173             BatchDefinition(
174                 datasource_name=self.datasource_name,
175                 data_connector_name=self.name,
176                 data_asset_name=data_asset_name,
177                 partition_definition=PartitionDefinition(data_reference),
178             )
179         ]
180 
181     def _generate_batch_spec_parameters_from_batch_definition(
182         self, batch_definition: BatchDefinition
183     ) -> Dict:
184         """
185         Build BatchSpec parameters from batch_definition with the following components:
186             1. data_asset_name from batch_definition
187             2. partition_definition from batch_definition
188             3. data_asset from data_connector
189 
190         Args:
191             batch_definition (BatchDefinition): to be used to build batch_spec
192 
193         Returns:
194             dict built from batch_definition
195         """
196         data_asset_name: str = batch_definition.data_asset_name
197         return {
198             "table_name": data_asset_name,
199             "partition_definition": batch_definition.partition_definition,
200             **self.data_assets[data_asset_name],
201         }
202 
203     # Splitter methods for listing partitions
204 
205     def _split_on_whole_table(
206         self,
207         table_name: str,
208     ):
209         """'Split' by returning the whole table
210 
211         Note: the table_name parameter is a required to keep the signature of this method consistent with other methods.
212         """
213 
214         return sa.select([sa.true()])
215 
216     def _split_on_column_value(
217         self,
218         table_name: str,
219         column_name: str,
220     ):
221         """Split using the values in the named column"""
222         # query = f"SELECT DISTINCT(\"{self.column_name}\") FROM {self.table_name}"
223 
224         return sa.select([sa.func.distinct(sa.column(column_name))]).select_from(
225             sa.text(table_name)
226         )
227 
228     def _split_on_converted_datetime(
229         self,
230         table_name: str,
231         column_name: str,
232         date_format_string: str = "%Y-%m-%d",
233     ):
234         """Convert the values in the named column to the given date_format, and split on that"""
235         # query = f"SELECT DISTINCT( strftime(\"{date_format_string}\", \"{self.column_name}\")) as my_var FROM {self.table_name}"
236 
237         return sa.select(
238             [
239                 sa.func.distinct(
240                     sa.func.strftime(
241                         date_format_string,
242                         sa.column(column_name),
243                     )
244                 )
245             ]
246         ).select_from(sa.text(table_name))
247 
248     def _split_on_divided_integer(
249         self, table_name: str, column_name: str, divisor: int
250     ):
251         """Divide the values in the named column by `divisor`, and split on that"""
252         # query = f"SELECT DISTINCT(\"{self.column_name}\" / {divisor}) AS my_var FROM {self.table_name}"
253 
254         return sa.select(
255             [sa.func.distinct(sa.cast(sa.column(column_name) / divisor, sa.Integer))]
256         ).select_from(sa.text(table_name))
257 
258     def _split_on_mod_integer(self, table_name: str, column_name: str, mod: int):
259         """Divide the values in the named column by `divisor`, and split on that"""
260         # query = f"SELECT DISTINCT(\"{self.column_name}\" / {divisor}) AS my_var FROM {self.table_name}"
261 
262         return sa.select(
263             [sa.func.distinct(sa.cast(sa.column(column_name) % mod, sa.Integer))]
264         ).select_from(sa.text(table_name))
265 
266     def _split_on_multi_column_values(
267         self,
268         table_name: str,
269         column_names: List[str],
270     ):
271         """Split on the joint values in the named columns"""
272         # query = f"SELECT DISTINCT(\"{self.column_name}\") FROM {self.table_name}"
273 
274         return (
275             sa.select([sa.column(column_name) for column_name in column_names])
276             .distinct()
277             .select_from(sa.text(table_name))
278         )
279 
280     def _split_on_hashed_column(
281         self,
282         table_name: str,
283         column_name: str,
284         hash_digits: int,
285     ):
286         """Note: this method is experimental. It does not work with all SQL dialects."""
287         # query = f"SELECT MD5(\"{self.column_name}\") = {matching_hash}) AS hashed_var FROM {self.table_name}"
288 
289         return sa.select([sa.func.md5(sa.column(column_name))]).select_from(
290             sa.text(table_name)
291         )
292
```
--- 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_sql_data_connector.py b/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py
--- a/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py
+++ b/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py
@@ -195,6 +195,7 @@
         """
         data_asset_name: str = batch_definition.data_asset_name
         return {
+            "data_asset_name": data_asset_name,
             "table_name": data_asset_name,
             "partition_definition": batch_definition.partition_definition,
             **self.data_assets[data_asset_name],

{"golden_diff": "diff --git a/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py b/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py\n--- a/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py\n+++ b/great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py\n@@ -195,6 +195,7 @@\n         \"\"\"\n         data_asset_name: str = batch_definition.data_asset_name\n         return {\n+            \"data_asset_name\": data_asset_name,\n             \"table_name\": data_asset_name,\n             \"partition_definition\": batch_definition.partition_definition,\n             **self.data_assets[data_asset_name],\n", "issue": "Asset Name is not populated in Data Docs for SimpleSqlalchemy data source\n**Describe the bug**\r\nWhen using a SimpleSqlalchemy data source, the Asset Name column is not populated in the Validations index of the data docs. SImilarly, the Data Asset entry in the corresponding Validation result is None. \r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\n1. Configure a SimpleSqlalchemy datasource\r\n```yaml\r\ndatasources:\r\n  db:\r\n    credentials: ${my_postgres_db}\r\n    class_name: SimpleSqlalchemyDatasource\r\n    introspection:\r\n      whole_table: {}\r\n``` \r\n2. Configure an expectation suite\r\n3. Configure a SimpleCheckpoint  \r\n```yaml\r\nname: my_simple_checkpoint\r\nconfig_version: 1.0\r\nclass_name: SimpleCheckpoint\r\nexpectation_suite_name: ge_validations_store.warning\r\nvalidations:\r\n  - batch_request:\r\n      datasource_name: db\r\n      data_connector_name: whole_table\r\n      data_asset_name: ge_validations_store\r\n      partition_request:\r\n        index: -1\r\n```\r\n4. Run checkpoint my_simple_checkpoint\r\n5. The generated data docs do not contain properly populated data asset entries\r\n\r\n**Expected behavior**\r\nThe generated data docs should be populated the same way when using the v3 API as when using the v2 API\r\n\r\n**Environment (please complete the following information):**\r\n - Operating System: Linux, MacOS\r\n - Great Expectations Version: develop branch\r\n\n", "before_files": [{"content": "from typing import Dict, List, Optional\n\nfrom great_expectations.core.batch import (\n    BatchDefinition,\n    BatchRequest,\n    PartitionDefinition,\n)\nfrom great_expectations.datasource.data_connector.data_connector import DataConnector\nfrom great_expectations.datasource.data_connector.util import (\n    batch_definition_matches_batch_request,\n)\nfrom great_expectations.execution_engine import ExecutionEngine\n\ntry:\n    import sqlalchemy as sa\nexcept ImportError:\n    sa = None\n\n\nclass ConfiguredAssetSqlDataConnector(DataConnector):\n    \"\"\"\n    A DataConnector that requires explicit listing of SQL tables you want to connect to.\n\n    Args:\n        name (str): The name of this DataConnector\n        datasource_name (str): The name of the Datasource that contains it\n        execution_engine (ExecutionEngine): An ExecutionEngine\n        data_assets (str): data_assets\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        execution_engine: Optional[ExecutionEngine] = None,\n        data_assets: Optional[Dict[str, dict]] = None,\n    ):\n        if data_assets is None:\n            data_assets = {}\n        self._data_assets = data_assets\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n        )\n\n    @property\n    def data_assets(self) -> Dict[str, dict]:\n        return self._data_assets\n\n    def add_data_asset(\n        self,\n        name: str,\n        config: dict,\n    ):\n        \"\"\"\n        Add data_asset to DataConnector using data_asset name as key, and data_asset configuration as value.\n        \"\"\"\n        self._data_assets[name] = config\n\n    def _get_partition_definition_list_from_data_asset_config(\n        self,\n        data_asset_name,\n        data_asset_config,\n    ):\n        if \"table_name\" in data_asset_config:\n            table_name = data_asset_config[\"table_name\"]\n        else:\n            table_name = data_asset_name\n\n        if \"splitter_method\" in data_asset_config:\n            splitter_fn = getattr(self, data_asset_config[\"splitter_method\"])\n            split_query = splitter_fn(\n                table_name=table_name, **data_asset_config[\"splitter_kwargs\"]\n            )\n\n            rows = self._execution_engine.engine.execute(split_query).fetchall()\n\n            # Zip up split parameters with column names\n            column_names = self._get_column_names_from_splitter_kwargs(\n                data_asset_config[\"splitter_kwargs\"]\n            )\n            partition_definition_list = [dict(zip(column_names, row)) for row in rows]\n\n        else:\n            partition_definition_list = [{}]\n\n        return partition_definition_list\n\n    def _refresh_data_references_cache(self):\n        self._data_references_cache = {}\n\n        for data_asset_name in self.data_assets:\n            data_asset = self.data_assets[data_asset_name]\n            partition_definition_list = (\n                self._get_partition_definition_list_from_data_asset_config(\n                    data_asset_name,\n                    data_asset,\n                )\n            )\n\n            # TODO Abe 20201029 : Apply sorters to partition_definition_list here\n            # TODO Will 20201102 : add sorting code here\n            self._data_references_cache[data_asset_name] = partition_definition_list\n\n    def _get_column_names_from_splitter_kwargs(self, splitter_kwargs) -> List[str]:\n        column_names: List[str] = []\n\n        if \"column_names\" in splitter_kwargs:\n            column_names = splitter_kwargs[\"column_names\"]\n        elif \"column_name\" in splitter_kwargs:\n            column_names = [splitter_kwargs[\"column_name\"]]\n\n        return column_names\n\n    def get_available_data_asset_names(self):\n        \"\"\"\n        Return the list of asset names known by this DataConnector.\n\n        Returns:\n            A list of available names\n        \"\"\"\n        return list(self.data_assets.keys())\n\n    def get_unmatched_data_references(self) -> List[str]:\n        \"\"\"\n        Returns the list of data_references unmatched by configuration by looping through items in _data_references_cache\n        and returning data_reference that do not have an associated data_asset.\n\n        Returns:\n            list of data_references that are not matched by configuration.\n        \"\"\"\n        return []\n\n    def get_batch_definition_list_from_batch_request(self, batch_request: BatchRequest):\n        self._validate_batch_request(batch_request=batch_request)\n\n        if len(self._data_references_cache) == 0:\n            self._refresh_data_references_cache()\n\n        batch_definition_list: List[BatchDefinition] = []\n        try:\n            sub_cache = self._data_references_cache[batch_request.data_asset_name]\n        except KeyError as e:\n            raise KeyError(\n                f\"data_asset_name {batch_request.data_asset_name} is not recognized.\"\n            )\n\n        for partition_definition in sub_cache:\n            batch_definition: BatchDefinition = BatchDefinition(\n                datasource_name=self.datasource_name,\n                data_connector_name=self.name,\n                data_asset_name=batch_request.data_asset_name,\n                partition_definition=PartitionDefinition(partition_definition),\n            )\n            if batch_definition_matches_batch_request(batch_definition, batch_request):\n                batch_definition_list.append(batch_definition)\n\n        return batch_definition_list\n\n    def _get_data_reference_list_from_cache_by_data_asset_name(\n        self, data_asset_name: str\n    ) -> List[str]:\n        return self._data_references_cache[data_asset_name]\n\n    def _map_data_reference_to_batch_definition_list(\n        self, data_reference, data_asset_name: Optional[str] = None  #: Any,\n    ) -> Optional[List[BatchDefinition]]:\n        # Note: This is a bit hacky, but it works. In sql_data_connectors, data references *are* dictionaries,\n        # allowing us to invoke `PartitionDefinition(data_reference)`\n        return [\n            BatchDefinition(\n                datasource_name=self.datasource_name,\n                data_connector_name=self.name,\n                data_asset_name=data_asset_name,\n                partition_definition=PartitionDefinition(data_reference),\n            )\n        ]\n\n    def _generate_batch_spec_parameters_from_batch_definition(\n        self, batch_definition: BatchDefinition\n    ) -> Dict:\n        \"\"\"\n        Build BatchSpec parameters from batch_definition with the following components:\n            1. data_asset_name from batch_definition\n            2. partition_definition from batch_definition\n            3. data_asset from data_connector\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            dict built from batch_definition\n        \"\"\"\n        data_asset_name: str = batch_definition.data_asset_name\n        return {\n            \"table_name\": data_asset_name,\n            \"partition_definition\": batch_definition.partition_definition,\n            **self.data_assets[data_asset_name],\n        }\n\n    # Splitter methods for listing partitions\n\n    def _split_on_whole_table(\n        self,\n        table_name: str,\n    ):\n        \"\"\"'Split' by returning the whole table\n\n        Note: the table_name parameter is a required to keep the signature of this method consistent with other methods.\n        \"\"\"\n\n        return sa.select([sa.true()])\n\n    def _split_on_column_value(\n        self,\n        table_name: str,\n        column_name: str,\n    ):\n        \"\"\"Split using the values in the named column\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\") FROM {self.table_name}\"\n\n        return sa.select([sa.func.distinct(sa.column(column_name))]).select_from(\n            sa.text(table_name)\n        )\n\n    def _split_on_converted_datetime(\n        self,\n        table_name: str,\n        column_name: str,\n        date_format_string: str = \"%Y-%m-%d\",\n    ):\n        \"\"\"Convert the values in the named column to the given date_format, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT( strftime(\\\"{date_format_string}\\\", \\\"{self.column_name}\\\")) as my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [\n                sa.func.distinct(\n                    sa.func.strftime(\n                        date_format_string,\n                        sa.column(column_name),\n                    )\n                )\n            ]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_divided_integer(\n        self, table_name: str, column_name: str, divisor: int\n    ):\n        \"\"\"Divide the values in the named column by `divisor`, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\" / {divisor}) AS my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [sa.func.distinct(sa.cast(sa.column(column_name) / divisor, sa.Integer))]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_mod_integer(self, table_name: str, column_name: str, mod: int):\n        \"\"\"Divide the values in the named column by `divisor`, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\" / {divisor}) AS my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [sa.func.distinct(sa.cast(sa.column(column_name) % mod, sa.Integer))]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_multi_column_values(\n        self,\n        table_name: str,\n        column_names: List[str],\n    ):\n        \"\"\"Split on the joint values in the named columns\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\") FROM {self.table_name}\"\n\n        return (\n            sa.select([sa.column(column_name) for column_name in column_names])\n            .distinct()\n            .select_from(sa.text(table_name))\n        )\n\n    def _split_on_hashed_column(\n        self,\n        table_name: str,\n        column_name: str,\n        hash_digits: int,\n    ):\n        \"\"\"Note: this method is experimental. It does not work with all SQL dialects.\"\"\"\n        # query = f\"SELECT MD5(\\\"{self.column_name}\\\") = {matching_hash}) AS hashed_var FROM {self.table_name}\"\n\n        return sa.select([sa.func.md5(sa.column(column_name))]).select_from(\n            sa.text(table_name)\n        )\n", "path": "great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py"}], "after_files": [{"content": "from typing import Dict, List, Optional\n\nfrom great_expectations.core.batch import (\n    BatchDefinition,\n    BatchRequest,\n    PartitionDefinition,\n)\nfrom great_expectations.datasource.data_connector.data_connector import DataConnector\nfrom great_expectations.datasource.data_connector.util import (\n    batch_definition_matches_batch_request,\n)\nfrom great_expectations.execution_engine import ExecutionEngine\n\ntry:\n    import sqlalchemy as sa\nexcept ImportError:\n    sa = None\n\n\nclass ConfiguredAssetSqlDataConnector(DataConnector):\n    \"\"\"\n    A DataConnector that requires explicit listing of SQL tables you want to connect to.\n\n    Args:\n        name (str): The name of this DataConnector\n        datasource_name (str): The name of the Datasource that contains it\n        execution_engine (ExecutionEngine): An ExecutionEngine\n        data_assets (str): data_assets\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str,\n        datasource_name: str,\n        execution_engine: Optional[ExecutionEngine] = None,\n        data_assets: Optional[Dict[str, dict]] = None,\n    ):\n        if data_assets is None:\n            data_assets = {}\n        self._data_assets = data_assets\n\n        super().__init__(\n            name=name,\n            datasource_name=datasource_name,\n            execution_engine=execution_engine,\n        )\n\n    @property\n    def data_assets(self) -> Dict[str, dict]:\n        return self._data_assets\n\n    def add_data_asset(\n        self,\n        name: str,\n        config: dict,\n    ):\n        \"\"\"\n        Add data_asset to DataConnector using data_asset name as key, and data_asset configuration as value.\n        \"\"\"\n        self._data_assets[name] = config\n\n    def _get_partition_definition_list_from_data_asset_config(\n        self,\n        data_asset_name,\n        data_asset_config,\n    ):\n        if \"table_name\" in data_asset_config:\n            table_name = data_asset_config[\"table_name\"]\n        else:\n            table_name = data_asset_name\n\n        if \"splitter_method\" in data_asset_config:\n            splitter_fn = getattr(self, data_asset_config[\"splitter_method\"])\n            split_query = splitter_fn(\n                table_name=table_name, **data_asset_config[\"splitter_kwargs\"]\n            )\n\n            rows = self._execution_engine.engine.execute(split_query).fetchall()\n\n            # Zip up split parameters with column names\n            column_names = self._get_column_names_from_splitter_kwargs(\n                data_asset_config[\"splitter_kwargs\"]\n            )\n            partition_definition_list = [dict(zip(column_names, row)) for row in rows]\n\n        else:\n            partition_definition_list = [{}]\n\n        return partition_definition_list\n\n    def _refresh_data_references_cache(self):\n        self._data_references_cache = {}\n\n        for data_asset_name in self.data_assets:\n            data_asset = self.data_assets[data_asset_name]\n            partition_definition_list = (\n                self._get_partition_definition_list_from_data_asset_config(\n                    data_asset_name,\n                    data_asset,\n                )\n            )\n\n            # TODO Abe 20201029 : Apply sorters to partition_definition_list here\n            # TODO Will 20201102 : add sorting code here\n            self._data_references_cache[data_asset_name] = partition_definition_list\n\n    def _get_column_names_from_splitter_kwargs(self, splitter_kwargs) -> List[str]:\n        column_names: List[str] = []\n\n        if \"column_names\" in splitter_kwargs:\n            column_names = splitter_kwargs[\"column_names\"]\n        elif \"column_name\" in splitter_kwargs:\n            column_names = [splitter_kwargs[\"column_name\"]]\n\n        return column_names\n\n    def get_available_data_asset_names(self):\n        \"\"\"\n        Return the list of asset names known by this DataConnector.\n\n        Returns:\n            A list of available names\n        \"\"\"\n        return list(self.data_assets.keys())\n\n    def get_unmatched_data_references(self) -> List[str]:\n        \"\"\"\n        Returns the list of data_references unmatched by configuration by looping through items in _data_references_cache\n        and returning data_reference that do not have an associated data_asset.\n\n        Returns:\n            list of data_references that are not matched by configuration.\n        \"\"\"\n        return []\n\n    def get_batch_definition_list_from_batch_request(self, batch_request: BatchRequest):\n        self._validate_batch_request(batch_request=batch_request)\n\n        if len(self._data_references_cache) == 0:\n            self._refresh_data_references_cache()\n\n        batch_definition_list: List[BatchDefinition] = []\n        try:\n            sub_cache = self._data_references_cache[batch_request.data_asset_name]\n        except KeyError as e:\n            raise KeyError(\n                f\"data_asset_name {batch_request.data_asset_name} is not recognized.\"\n            )\n\n        for partition_definition in sub_cache:\n            batch_definition: BatchDefinition = BatchDefinition(\n                datasource_name=self.datasource_name,\n                data_connector_name=self.name,\n                data_asset_name=batch_request.data_asset_name,\n                partition_definition=PartitionDefinition(partition_definition),\n            )\n            if batch_definition_matches_batch_request(batch_definition, batch_request):\n                batch_definition_list.append(batch_definition)\n\n        return batch_definition_list\n\n    def _get_data_reference_list_from_cache_by_data_asset_name(\n        self, data_asset_name: str\n    ) -> List[str]:\n        return self._data_references_cache[data_asset_name]\n\n    def _map_data_reference_to_batch_definition_list(\n        self, data_reference, data_asset_name: Optional[str] = None  #: Any,\n    ) -> Optional[List[BatchDefinition]]:\n        # Note: This is a bit hacky, but it works. In sql_data_connectors, data references *are* dictionaries,\n        # allowing us to invoke `PartitionDefinition(data_reference)`\n        return [\n            BatchDefinition(\n                datasource_name=self.datasource_name,\n                data_connector_name=self.name,\n                data_asset_name=data_asset_name,\n                partition_definition=PartitionDefinition(data_reference),\n            )\n        ]\n\n    def _generate_batch_spec_parameters_from_batch_definition(\n        self, batch_definition: BatchDefinition\n    ) -> Dict:\n        \"\"\"\n        Build BatchSpec parameters from batch_definition with the following components:\n            1. data_asset_name from batch_definition\n            2. partition_definition from batch_definition\n            3. data_asset from data_connector\n\n        Args:\n            batch_definition (BatchDefinition): to be used to build batch_spec\n\n        Returns:\n            dict built from batch_definition\n        \"\"\"\n        data_asset_name: str = batch_definition.data_asset_name\n        return {\n            \"data_asset_name\": data_asset_name,\n            \"table_name\": data_asset_name,\n            \"partition_definition\": batch_definition.partition_definition,\n            **self.data_assets[data_asset_name],\n        }\n\n    # Splitter methods for listing partitions\n\n    def _split_on_whole_table(\n        self,\n        table_name: str,\n    ):\n        \"\"\"'Split' by returning the whole table\n\n        Note: the table_name parameter is a required to keep the signature of this method consistent with other methods.\n        \"\"\"\n\n        return sa.select([sa.true()])\n\n    def _split_on_column_value(\n        self,\n        table_name: str,\n        column_name: str,\n    ):\n        \"\"\"Split using the values in the named column\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\") FROM {self.table_name}\"\n\n        return sa.select([sa.func.distinct(sa.column(column_name))]).select_from(\n            sa.text(table_name)\n        )\n\n    def _split_on_converted_datetime(\n        self,\n        table_name: str,\n        column_name: str,\n        date_format_string: str = \"%Y-%m-%d\",\n    ):\n        \"\"\"Convert the values in the named column to the given date_format, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT( strftime(\\\"{date_format_string}\\\", \\\"{self.column_name}\\\")) as my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [\n                sa.func.distinct(\n                    sa.func.strftime(\n                        date_format_string,\n                        sa.column(column_name),\n                    )\n                )\n            ]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_divided_integer(\n        self, table_name: str, column_name: str, divisor: int\n    ):\n        \"\"\"Divide the values in the named column by `divisor`, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\" / {divisor}) AS my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [sa.func.distinct(sa.cast(sa.column(column_name) / divisor, sa.Integer))]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_mod_integer(self, table_name: str, column_name: str, mod: int):\n        \"\"\"Divide the values in the named column by `divisor`, and split on that\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\" / {divisor}) AS my_var FROM {self.table_name}\"\n\n        return sa.select(\n            [sa.func.distinct(sa.cast(sa.column(column_name) % mod, sa.Integer))]\n        ).select_from(sa.text(table_name))\n\n    def _split_on_multi_column_values(\n        self,\n        table_name: str,\n        column_names: List[str],\n    ):\n        \"\"\"Split on the joint values in the named columns\"\"\"\n        # query = f\"SELECT DISTINCT(\\\"{self.column_name}\\\") FROM {self.table_name}\"\n\n        return (\n            sa.select([sa.column(column_name) for column_name in column_names])\n            .distinct()\n            .select_from(sa.text(table_name))\n        )\n\n    def _split_on_hashed_column(\n        self,\n        table_name: str,\n        column_name: str,\n        hash_digits: int,\n    ):\n        \"\"\"Note: this method is experimental. It does not work with all SQL dialects.\"\"\"\n        # query = f\"SELECT MD5(\\\"{self.column_name}\\\") = {matching_hash}) AS hashed_var FROM {self.table_name}\"\n\n        return sa.select([sa.func.md5(sa.column(column_name))]).select_from(\n            sa.text(table_name)\n        )\n", "path": "great_expectations/datasource/data_connector/configured_asset_sql_data_connector.py"}]}