We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Use cleaner solution for non-truncating division in python 2
Prefer `from __future__ import division` to `1.*x/y`

--- 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/sqlalchemy_datasource.py`
Content:
```
1 import datetime
2 import logging
3 from string import Template
4 from urllib.parse import urlparse
5 
6 from great_expectations.core.batch import Batch
7 from great_expectations.core.util import nested_update
8 from great_expectations.dataset.sqlalchemy_dataset import SqlAlchemyBatchReference
9 from great_expectations.datasource import Datasource
10 from great_expectations.datasource.types import BatchMarkers
11 from great_expectations.exceptions import DatasourceInitializationError
12 from great_expectations.types import ClassConfig
13 from great_expectations.types.configurations import classConfigSchema
14 
15 logger = logging.getLogger(__name__)
16 
17 try:
18     import sqlalchemy
19     from sqlalchemy import create_engine
20 except ImportError:
21     sqlalchemy = None
22     create_engine = None
23     logger.debug("Unable to import sqlalchemy.")
24 
25 
26 try:
27     import google.auth
28 
29     datasource_initialization_exceptions = (
30         sqlalchemy.exc.OperationalError,
31         sqlalchemy.exc.DatabaseError,
32         sqlalchemy.exc.ArgumentError,
33         google.auth.exceptions.GoogleAuthError,
34     )
35 except ImportError:
36     datasource_initialization_exceptions = (
37         sqlalchemy.exc.OperationalError,
38         sqlalchemy.exc.DatabaseError,
39         sqlalchemy.exc.ArgumentError,
40     )
41 
42 
43 class SqlAlchemyDatasource(Datasource):
44     """
45     A SqlAlchemyDatasource will provide data_assets converting batch_kwargs using the following rules:
46       - if the batch_kwargs include a table key, the datasource will provide a dataset object connected
47         to that table
48       - if the batch_kwargs include a query key, the datasource will create a temporary table using that
49         that query. The query can be parameterized according to the standard python Template engine, which
50         uses $parameter, with additional kwargs passed to the get_batch method.
51     """
52 
53     recognized_batch_parameters = {"query_parameters", "limit", "dataset_options"}
54 
55     @classmethod
56     def build_configuration(
57         cls, data_asset_type=None, batch_kwargs_generators=None, **kwargs
58     ):
59         """
60         Build a full configuration object for a datasource, potentially including generators with defaults.
61 
62         Args:
63             data_asset_type: A ClassConfig dictionary
64             batch_kwargs_generators: Generator configuration dictionary
65             **kwargs: Additional kwargs to be part of the datasource constructor's initialization
66 
67         Returns:
68             A complete datasource configuration.
69 
70         """
71 
72         if data_asset_type is None:
73             data_asset_type = {
74                 "class_name": "SqlAlchemyDataset",
75                 "module_name": "great_expectations.dataset",
76             }
77         else:
78             data_asset_type = classConfigSchema.dump(ClassConfig(**data_asset_type))
79 
80         configuration = kwargs
81         configuration["data_asset_type"] = data_asset_type
82         if batch_kwargs_generators is not None:
83             configuration["batch_kwargs_generators"] = batch_kwargs_generators
84 
85         return configuration
86 
87     def __init__(
88         self,
89         name="default",
90         data_context=None,
91         data_asset_type=None,
92         credentials=None,
93         batch_kwargs_generators=None,
94         **kwargs
95     ):
96         if not sqlalchemy:
97             raise DatasourceInitializationError(
98                 name, "ModuleNotFoundError: No module named 'sqlalchemy'"
99             )
100 
101         configuration_with_defaults = SqlAlchemyDatasource.build_configuration(
102             data_asset_type, batch_kwargs_generators, **kwargs
103         )
104         data_asset_type = configuration_with_defaults.pop("data_asset_type")
105         batch_kwargs_generators = configuration_with_defaults.pop(
106             "batch_kwargs_generators", None
107         )
108         super(SqlAlchemyDatasource, self).__init__(
109             name,
110             data_context=data_context,
111             data_asset_type=data_asset_type,
112             batch_kwargs_generators=batch_kwargs_generators,
113             **configuration_with_defaults
114         )
115 
116         if credentials is not None:
117             self._datasource_config.update({"credentials": credentials})
118         else:
119             credentials = {}
120 
121         try:
122             # if an engine was provided, use that
123             if "engine" in kwargs:
124                 self.engine = kwargs.pop("engine")
125 
126             # if a connection string or url was provided, use that
127             elif "connection_string" in kwargs:
128                 connection_string = kwargs.pop("connection_string")
129                 self.engine = create_engine(connection_string, **kwargs)
130                 self.engine.connect()
131             elif "url" in credentials:
132                 url = credentials.pop("url")
133                 self.drivername = urlparse(url).scheme
134                 self.engine = create_engine(url, **kwargs)
135                 self.engine.connect()
136 
137             # Otherwise, connect using remaining kwargs
138             else:
139                 options, drivername = self._get_sqlalchemy_connection_options(**kwargs)
140                 self.drivername = drivername
141                 self.engine = create_engine(options)
142                 self.engine.connect()
143 
144         except datasource_initialization_exceptions as sqlalchemy_error:
145             raise DatasourceInitializationError(self._name, str(sqlalchemy_error))
146 
147         self._build_generators()
148 
149     def _get_sqlalchemy_connection_options(self, **kwargs):
150         drivername = None
151         if "credentials" in self._datasource_config:
152             credentials = self._datasource_config["credentials"]
153         else:
154             credentials = {}
155 
156         # if a connection string or url was provided in the profile, use that
157         if "connection_string" in credentials:
158             options = credentials["connection_string"]
159         elif "url" in credentials:
160             options = credentials["url"]
161         else:
162             # Update credentials with anything passed during connection time
163             drivername = credentials.pop("drivername")
164             schema_name = credentials.pop("schema_name", None)
165             if schema_name is not None:
166                 logger.warning(
167                     "schema_name specified creating a URL with schema is not supported. Set a default "
168                     "schema on the user connecting to your database."
169                 )
170             options = sqlalchemy.engine.url.URL(drivername, **credentials)
171         return options, drivername
172 
173     def get_batch(self, batch_kwargs, batch_parameters=None):
174         # We need to build a batch_id to be used in the dataframe
175         batch_markers = BatchMarkers(
176             {
177                 "ge_load_time": datetime.datetime.now(datetime.timezone.utc).strftime(
178                     "%Y%m%dT%H%M%S.%fZ"
179                 )
180             }
181         )
182 
183         if "bigquery_temp_table" in batch_kwargs:
184             query_support_table_name = batch_kwargs.get("bigquery_temp_table")
185         elif "snowflake_transient_table" in batch_kwargs:
186             # Snowflake uses a transient table, so we expect a table_name to be provided
187             query_support_table_name = batch_kwargs.get("snowflake_transient_table")
188         else:
189             query_support_table_name = None
190 
191         if "query" in batch_kwargs:
192             if "limit" in batch_kwargs or "offset" in batch_kwargs:
193                 logger.warning(
194                     "Limit and offset parameters are ignored when using query-based batch_kwargs; consider "
195                     "adding limit and offset directly to the generated query."
196                 )
197             if "query_parameters" in batch_kwargs:
198                 query = Template(batch_kwargs["query"]).safe_substitute(
199                     batch_kwargs["query_parameters"]
200                 )
201             else:
202                 query = batch_kwargs["query"]
203             batch_reference = SqlAlchemyBatchReference(
204                 engine=self.engine,
205                 query=query,
206                 table_name=query_support_table_name,
207                 schema=batch_kwargs.get("schema"),
208             )
209         elif "table" in batch_kwargs:
210             table = batch_kwargs["table"]
211             limit = batch_kwargs.get("limit")
212             offset = batch_kwargs.get("offset")
213             if limit is not None or offset is not None:
214                 logger.info(
215                     "Generating query from table batch_kwargs based on limit and offset"
216                 )
217                 # In BigQuery the table name is already qualified with its schema name
218                 if self.engine.dialect.name.lower() == "bigquery":
219                     schema = None
220                 else:
221                     schema = batch_kwargs.get("schema")
222                 raw_query = (
223                     sqlalchemy.select([sqlalchemy.text("*")])
224                     .select_from(
225                         sqlalchemy.schema.Table(
226                             table, sqlalchemy.MetaData(), schema=schema
227                         )
228                     )
229                     .offset(offset)
230                     .limit(limit)
231                 )
232                 query = str(
233                     raw_query.compile(
234                         self.engine, compile_kwargs={"literal_binds": True}
235                     )
236                 )
237                 batch_reference = SqlAlchemyBatchReference(
238                     engine=self.engine,
239                     query=query,
240                     table_name=query_support_table_name,
241                     schema=batch_kwargs.get("schema"),
242                 )
243             else:
244                 batch_reference = SqlAlchemyBatchReference(
245                     engine=self.engine,
246                     table_name=table,
247                     schema=batch_kwargs.get("schema"),
248                 )
249         else:
250             raise ValueError(
251                 "Invalid batch_kwargs: exactly one of 'table' or 'query' must be specified"
252             )
253 
254         return Batch(
255             datasource_name=self.name,
256             batch_kwargs=batch_kwargs,
257             data=batch_reference,
258             batch_parameters=batch_parameters,
259             batch_markers=batch_markers,
260             data_context=self._data_context,
261         )
262 
263     def process_batch_parameters(
264         self, query_parameters=None, limit=None, dataset_options=None
265     ):
266         batch_kwargs = super(SqlAlchemyDatasource, self).process_batch_parameters(
267             limit=limit, dataset_options=dataset_options,
268         )
269         nested_update(batch_kwargs, {"query_parameters": query_parameters})
270         return batch_kwargs
271
```
--- 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/sqlalchemy_datasource.py b/great_expectations/datasource/sqlalchemy_datasource.py
--- a/great_expectations/datasource/sqlalchemy_datasource.py
+++ b/great_expectations/datasource/sqlalchemy_datasource.py
@@ -23,21 +23,22 @@
     logger.debug("Unable to import sqlalchemy.")
 
 
-try:
-    import google.auth
+if sqlalchemy != None:
+    try:
+        import google.auth
 
-    datasource_initialization_exceptions = (
-        sqlalchemy.exc.OperationalError,
-        sqlalchemy.exc.DatabaseError,
-        sqlalchemy.exc.ArgumentError,
-        google.auth.exceptions.GoogleAuthError,
-    )
-except ImportError:
-    datasource_initialization_exceptions = (
-        sqlalchemy.exc.OperationalError,
-        sqlalchemy.exc.DatabaseError,
-        sqlalchemy.exc.ArgumentError,
-    )
+        datasource_initialization_exceptions = (
+            sqlalchemy.exc.OperationalError,
+            sqlalchemy.exc.DatabaseError,
+            sqlalchemy.exc.ArgumentError,
+            google.auth.exceptions.GoogleAuthError,
+        )
+    except ImportError:
+        datasource_initialization_exceptions = (
+            sqlalchemy.exc.OperationalError,
+            sqlalchemy.exc.DatabaseError,
+            sqlalchemy.exc.ArgumentError,
+        )
 
 
 class SqlAlchemyDatasource(Datasource):

{"golden_diff": "diff --git a/great_expectations/datasource/sqlalchemy_datasource.py b/great_expectations/datasource/sqlalchemy_datasource.py\n--- a/great_expectations/datasource/sqlalchemy_datasource.py\n+++ b/great_expectations/datasource/sqlalchemy_datasource.py\n@@ -23,21 +23,22 @@\n     logger.debug(\"Unable to import sqlalchemy.\")\n \n \n-try:\n-    import google.auth\n+if sqlalchemy != None:\n+    try:\n+        import google.auth\n \n-    datasource_initialization_exceptions = (\n-        sqlalchemy.exc.OperationalError,\n-        sqlalchemy.exc.DatabaseError,\n-        sqlalchemy.exc.ArgumentError,\n-        google.auth.exceptions.GoogleAuthError,\n-    )\n-except ImportError:\n-    datasource_initialization_exceptions = (\n-        sqlalchemy.exc.OperationalError,\n-        sqlalchemy.exc.DatabaseError,\n-        sqlalchemy.exc.ArgumentError,\n-    )\n+        datasource_initialization_exceptions = (\n+            sqlalchemy.exc.OperationalError,\n+            sqlalchemy.exc.DatabaseError,\n+            sqlalchemy.exc.ArgumentError,\n+            google.auth.exceptions.GoogleAuthError,\n+        )\n+    except ImportError:\n+        datasource_initialization_exceptions = (\n+            sqlalchemy.exc.OperationalError,\n+            sqlalchemy.exc.DatabaseError,\n+            sqlalchemy.exc.ArgumentError,\n+        )\n \n \n class SqlAlchemyDatasource(Datasource):\n", "issue": "Use cleaner solution for non-truncating division in python 2\nPrefer `from __future__ import division` to `1.*x/y`\n", "before_files": [{"content": "import datetime\nimport logging\nfrom string import Template\nfrom urllib.parse import urlparse\n\nfrom great_expectations.core.batch import Batch\nfrom great_expectations.core.util import nested_update\nfrom great_expectations.dataset.sqlalchemy_dataset import SqlAlchemyBatchReference\nfrom great_expectations.datasource import Datasource\nfrom great_expectations.datasource.types import BatchMarkers\nfrom great_expectations.exceptions import DatasourceInitializationError\nfrom great_expectations.types import ClassConfig\nfrom great_expectations.types.configurations import classConfigSchema\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import sqlalchemy\n    from sqlalchemy import create_engine\nexcept ImportError:\n    sqlalchemy = None\n    create_engine = None\n    logger.debug(\"Unable to import sqlalchemy.\")\n\n\ntry:\n    import google.auth\n\n    datasource_initialization_exceptions = (\n        sqlalchemy.exc.OperationalError,\n        sqlalchemy.exc.DatabaseError,\n        sqlalchemy.exc.ArgumentError,\n        google.auth.exceptions.GoogleAuthError,\n    )\nexcept ImportError:\n    datasource_initialization_exceptions = (\n        sqlalchemy.exc.OperationalError,\n        sqlalchemy.exc.DatabaseError,\n        sqlalchemy.exc.ArgumentError,\n    )\n\n\nclass SqlAlchemyDatasource(Datasource):\n    \"\"\"\n    A SqlAlchemyDatasource will provide data_assets converting batch_kwargs using the following rules:\n      - if the batch_kwargs include a table key, the datasource will provide a dataset object connected\n        to that table\n      - if the batch_kwargs include a query key, the datasource will create a temporary table using that\n        that query. The query can be parameterized according to the standard python Template engine, which\n        uses $parameter, with additional kwargs passed to the get_batch method.\n    \"\"\"\n\n    recognized_batch_parameters = {\"query_parameters\", \"limit\", \"dataset_options\"}\n\n    @classmethod\n    def build_configuration(\n        cls, data_asset_type=None, batch_kwargs_generators=None, **kwargs\n    ):\n        \"\"\"\n        Build a full configuration object for a datasource, potentially including generators with defaults.\n\n        Args:\n            data_asset_type: A ClassConfig dictionary\n            batch_kwargs_generators: Generator configuration dictionary\n            **kwargs: Additional kwargs to be part of the datasource constructor's initialization\n\n        Returns:\n            A complete datasource configuration.\n\n        \"\"\"\n\n        if data_asset_type is None:\n            data_asset_type = {\n                \"class_name\": \"SqlAlchemyDataset\",\n                \"module_name\": \"great_expectations.dataset\",\n            }\n        else:\n            data_asset_type = classConfigSchema.dump(ClassConfig(**data_asset_type))\n\n        configuration = kwargs\n        configuration[\"data_asset_type\"] = data_asset_type\n        if batch_kwargs_generators is not None:\n            configuration[\"batch_kwargs_generators\"] = batch_kwargs_generators\n\n        return configuration\n\n    def __init__(\n        self,\n        name=\"default\",\n        data_context=None,\n        data_asset_type=None,\n        credentials=None,\n        batch_kwargs_generators=None,\n        **kwargs\n    ):\n        if not sqlalchemy:\n            raise DatasourceInitializationError(\n                name, \"ModuleNotFoundError: No module named 'sqlalchemy'\"\n            )\n\n        configuration_with_defaults = SqlAlchemyDatasource.build_configuration(\n            data_asset_type, batch_kwargs_generators, **kwargs\n        )\n        data_asset_type = configuration_with_defaults.pop(\"data_asset_type\")\n        batch_kwargs_generators = configuration_with_defaults.pop(\n            \"batch_kwargs_generators\", None\n        )\n        super(SqlAlchemyDatasource, self).__init__(\n            name,\n            data_context=data_context,\n            data_asset_type=data_asset_type,\n            batch_kwargs_generators=batch_kwargs_generators,\n            **configuration_with_defaults\n        )\n\n        if credentials is not None:\n            self._datasource_config.update({\"credentials\": credentials})\n        else:\n            credentials = {}\n\n        try:\n            # if an engine was provided, use that\n            if \"engine\" in kwargs:\n                self.engine = kwargs.pop(\"engine\")\n\n            # if a connection string or url was provided, use that\n            elif \"connection_string\" in kwargs:\n                connection_string = kwargs.pop(\"connection_string\")\n                self.engine = create_engine(connection_string, **kwargs)\n                self.engine.connect()\n            elif \"url\" in credentials:\n                url = credentials.pop(\"url\")\n                self.drivername = urlparse(url).scheme\n                self.engine = create_engine(url, **kwargs)\n                self.engine.connect()\n\n            # Otherwise, connect using remaining kwargs\n            else:\n                options, drivername = self._get_sqlalchemy_connection_options(**kwargs)\n                self.drivername = drivername\n                self.engine = create_engine(options)\n                self.engine.connect()\n\n        except datasource_initialization_exceptions as sqlalchemy_error:\n            raise DatasourceInitializationError(self._name, str(sqlalchemy_error))\n\n        self._build_generators()\n\n    def _get_sqlalchemy_connection_options(self, **kwargs):\n        drivername = None\n        if \"credentials\" in self._datasource_config:\n            credentials = self._datasource_config[\"credentials\"]\n        else:\n            credentials = {}\n\n        # if a connection string or url was provided in the profile, use that\n        if \"connection_string\" in credentials:\n            options = credentials[\"connection_string\"]\n        elif \"url\" in credentials:\n            options = credentials[\"url\"]\n        else:\n            # Update credentials with anything passed during connection time\n            drivername = credentials.pop(\"drivername\")\n            schema_name = credentials.pop(\"schema_name\", None)\n            if schema_name is not None:\n                logger.warning(\n                    \"schema_name specified creating a URL with schema is not supported. Set a default \"\n                    \"schema on the user connecting to your database.\"\n                )\n            options = sqlalchemy.engine.url.URL(drivername, **credentials)\n        return options, drivername\n\n    def get_batch(self, batch_kwargs, batch_parameters=None):\n        # We need to build a batch_id to be used in the dataframe\n        batch_markers = BatchMarkers(\n            {\n                \"ge_load_time\": datetime.datetime.now(datetime.timezone.utc).strftime(\n                    \"%Y%m%dT%H%M%S.%fZ\"\n                )\n            }\n        )\n\n        if \"bigquery_temp_table\" in batch_kwargs:\n            query_support_table_name = batch_kwargs.get(\"bigquery_temp_table\")\n        elif \"snowflake_transient_table\" in batch_kwargs:\n            # Snowflake uses a transient table, so we expect a table_name to be provided\n            query_support_table_name = batch_kwargs.get(\"snowflake_transient_table\")\n        else:\n            query_support_table_name = None\n\n        if \"query\" in batch_kwargs:\n            if \"limit\" in batch_kwargs or \"offset\" in batch_kwargs:\n                logger.warning(\n                    \"Limit and offset parameters are ignored when using query-based batch_kwargs; consider \"\n                    \"adding limit and offset directly to the generated query.\"\n                )\n            if \"query_parameters\" in batch_kwargs:\n                query = Template(batch_kwargs[\"query\"]).safe_substitute(\n                    batch_kwargs[\"query_parameters\"]\n                )\n            else:\n                query = batch_kwargs[\"query\"]\n            batch_reference = SqlAlchemyBatchReference(\n                engine=self.engine,\n                query=query,\n                table_name=query_support_table_name,\n                schema=batch_kwargs.get(\"schema\"),\n            )\n        elif \"table\" in batch_kwargs:\n            table = batch_kwargs[\"table\"]\n            limit = batch_kwargs.get(\"limit\")\n            offset = batch_kwargs.get(\"offset\")\n            if limit is not None or offset is not None:\n                logger.info(\n                    \"Generating query from table batch_kwargs based on limit and offset\"\n                )\n                # In BigQuery the table name is already qualified with its schema name\n                if self.engine.dialect.name.lower() == \"bigquery\":\n                    schema = None\n                else:\n                    schema = batch_kwargs.get(\"schema\")\n                raw_query = (\n                    sqlalchemy.select([sqlalchemy.text(\"*\")])\n                    .select_from(\n                        sqlalchemy.schema.Table(\n                            table, sqlalchemy.MetaData(), schema=schema\n                        )\n                    )\n                    .offset(offset)\n                    .limit(limit)\n                )\n                query = str(\n                    raw_query.compile(\n                        self.engine, compile_kwargs={\"literal_binds\": True}\n                    )\n                )\n                batch_reference = SqlAlchemyBatchReference(\n                    engine=self.engine,\n                    query=query,\n                    table_name=query_support_table_name,\n                    schema=batch_kwargs.get(\"schema\"),\n                )\n            else:\n                batch_reference = SqlAlchemyBatchReference(\n                    engine=self.engine,\n                    table_name=table,\n                    schema=batch_kwargs.get(\"schema\"),\n                )\n        else:\n            raise ValueError(\n                \"Invalid batch_kwargs: exactly one of 'table' or 'query' must be specified\"\n            )\n\n        return Batch(\n            datasource_name=self.name,\n            batch_kwargs=batch_kwargs,\n            data=batch_reference,\n            batch_parameters=batch_parameters,\n            batch_markers=batch_markers,\n            data_context=self._data_context,\n        )\n\n    def process_batch_parameters(\n        self, query_parameters=None, limit=None, dataset_options=None\n    ):\n        batch_kwargs = super(SqlAlchemyDatasource, self).process_batch_parameters(\n            limit=limit, dataset_options=dataset_options,\n        )\n        nested_update(batch_kwargs, {\"query_parameters\": query_parameters})\n        return batch_kwargs\n", "path": "great_expectations/datasource/sqlalchemy_datasource.py"}], "after_files": [{"content": "import datetime\nimport logging\nfrom string import Template\nfrom urllib.parse import urlparse\n\nfrom great_expectations.core.batch import Batch\nfrom great_expectations.core.util import nested_update\nfrom great_expectations.dataset.sqlalchemy_dataset import SqlAlchemyBatchReference\nfrom great_expectations.datasource import Datasource\nfrom great_expectations.datasource.types import BatchMarkers\nfrom great_expectations.exceptions import DatasourceInitializationError\nfrom great_expectations.types import ClassConfig\nfrom great_expectations.types.configurations import classConfigSchema\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import sqlalchemy\n    from sqlalchemy import create_engine\nexcept ImportError:\n    sqlalchemy = None\n    create_engine = None\n    logger.debug(\"Unable to import sqlalchemy.\")\n\n\nif sqlalchemy != None:\n    try:\n        import google.auth\n\n        datasource_initialization_exceptions = (\n            sqlalchemy.exc.OperationalError,\n            sqlalchemy.exc.DatabaseError,\n            sqlalchemy.exc.ArgumentError,\n            google.auth.exceptions.GoogleAuthError,\n        )\n    except ImportError:\n        datasource_initialization_exceptions = (\n            sqlalchemy.exc.OperationalError,\n            sqlalchemy.exc.DatabaseError,\n            sqlalchemy.exc.ArgumentError,\n        )\n\n\nclass SqlAlchemyDatasource(Datasource):\n    \"\"\"\n    A SqlAlchemyDatasource will provide data_assets converting batch_kwargs using the following rules:\n      - if the batch_kwargs include a table key, the datasource will provide a dataset object connected\n        to that table\n      - if the batch_kwargs include a query key, the datasource will create a temporary table using that\n        that query. The query can be parameterized according to the standard python Template engine, which\n        uses $parameter, with additional kwargs passed to the get_batch method.\n    \"\"\"\n\n    recognized_batch_parameters = {\"query_parameters\", \"limit\", \"dataset_options\"}\n\n    @classmethod\n    def build_configuration(\n        cls, data_asset_type=None, batch_kwargs_generators=None, **kwargs\n    ):\n        \"\"\"\n        Build a full configuration object for a datasource, potentially including generators with defaults.\n\n        Args:\n            data_asset_type: A ClassConfig dictionary\n            batch_kwargs_generators: Generator configuration dictionary\n            **kwargs: Additional kwargs to be part of the datasource constructor's initialization\n\n        Returns:\n            A complete datasource configuration.\n\n        \"\"\"\n\n        if data_asset_type is None:\n            data_asset_type = {\n                \"class_name\": \"SqlAlchemyDataset\",\n                \"module_name\": \"great_expectations.dataset\",\n            }\n        else:\n            data_asset_type = classConfigSchema.dump(ClassConfig(**data_asset_type))\n\n        configuration = kwargs\n        configuration[\"data_asset_type\"] = data_asset_type\n        if batch_kwargs_generators is not None:\n            configuration[\"batch_kwargs_generators\"] = batch_kwargs_generators\n\n        return configuration\n\n    def __init__(\n        self,\n        name=\"default\",\n        data_context=None,\n        data_asset_type=None,\n        credentials=None,\n        batch_kwargs_generators=None,\n        **kwargs\n    ):\n        if not sqlalchemy:\n            raise DatasourceInitializationError(\n                name, \"ModuleNotFoundError: No module named 'sqlalchemy'\"\n            )\n\n        configuration_with_defaults = SqlAlchemyDatasource.build_configuration(\n            data_asset_type, batch_kwargs_generators, **kwargs\n        )\n        data_asset_type = configuration_with_defaults.pop(\"data_asset_type\")\n        batch_kwargs_generators = configuration_with_defaults.pop(\n            \"batch_kwargs_generators\", None\n        )\n        super(SqlAlchemyDatasource, self).__init__(\n            name,\n            data_context=data_context,\n            data_asset_type=data_asset_type,\n            batch_kwargs_generators=batch_kwargs_generators,\n            **configuration_with_defaults\n        )\n\n        if credentials is not None:\n            self._datasource_config.update({\"credentials\": credentials})\n        else:\n            credentials = {}\n\n        try:\n            # if an engine was provided, use that\n            if \"engine\" in kwargs:\n                self.engine = kwargs.pop(\"engine\")\n\n            # if a connection string or url was provided, use that\n            elif \"connection_string\" in kwargs:\n                connection_string = kwargs.pop(\"connection_string\")\n                self.engine = create_engine(connection_string, **kwargs)\n                self.engine.connect()\n            elif \"url\" in credentials:\n                url = credentials.pop(\"url\")\n                self.drivername = urlparse(url).scheme\n                self.engine = create_engine(url, **kwargs)\n                self.engine.connect()\n\n            # Otherwise, connect using remaining kwargs\n            else:\n                options, drivername = self._get_sqlalchemy_connection_options(**kwargs)\n                self.drivername = drivername\n                self.engine = create_engine(options)\n                self.engine.connect()\n\n        except datasource_initialization_exceptions as sqlalchemy_error:\n            raise DatasourceInitializationError(self._name, str(sqlalchemy_error))\n\n        self._build_generators()\n\n    def _get_sqlalchemy_connection_options(self, **kwargs):\n        drivername = None\n        if \"credentials\" in self._datasource_config:\n            credentials = self._datasource_config[\"credentials\"]\n        else:\n            credentials = {}\n\n        # if a connection string or url was provided in the profile, use that\n        if \"connection_string\" in credentials:\n            options = credentials[\"connection_string\"]\n        elif \"url\" in credentials:\n            options = credentials[\"url\"]\n        else:\n            # Update credentials with anything passed during connection time\n            drivername = credentials.pop(\"drivername\")\n            schema_name = credentials.pop(\"schema_name\", None)\n            if schema_name is not None:\n                logger.warning(\n                    \"schema_name specified creating a URL with schema is not supported. Set a default \"\n                    \"schema on the user connecting to your database.\"\n                )\n            options = sqlalchemy.engine.url.URL(drivername, **credentials)\n        return options, drivername\n\n    def get_batch(self, batch_kwargs, batch_parameters=None):\n        # We need to build a batch_id to be used in the dataframe\n        batch_markers = BatchMarkers(\n            {\n                \"ge_load_time\": datetime.datetime.now(datetime.timezone.utc).strftime(\n                    \"%Y%m%dT%H%M%S.%fZ\"\n                )\n            }\n        )\n\n        if \"bigquery_temp_table\" in batch_kwargs:\n            query_support_table_name = batch_kwargs.get(\"bigquery_temp_table\")\n        elif \"snowflake_transient_table\" in batch_kwargs:\n            # Snowflake uses a transient table, so we expect a table_name to be provided\n            query_support_table_name = batch_kwargs.get(\"snowflake_transient_table\")\n        else:\n            query_support_table_name = None\n\n        if \"query\" in batch_kwargs:\n            if \"limit\" in batch_kwargs or \"offset\" in batch_kwargs:\n                logger.warning(\n                    \"Limit and offset parameters are ignored when using query-based batch_kwargs; consider \"\n                    \"adding limit and offset directly to the generated query.\"\n                )\n            if \"query_parameters\" in batch_kwargs:\n                query = Template(batch_kwargs[\"query\"]).safe_substitute(\n                    batch_kwargs[\"query_parameters\"]\n                )\n            else:\n                query = batch_kwargs[\"query\"]\n            batch_reference = SqlAlchemyBatchReference(\n                engine=self.engine,\n                query=query,\n                table_name=query_support_table_name,\n                schema=batch_kwargs.get(\"schema\"),\n            )\n        elif \"table\" in batch_kwargs:\n            table = batch_kwargs[\"table\"]\n            limit = batch_kwargs.get(\"limit\")\n            offset = batch_kwargs.get(\"offset\")\n            if limit is not None or offset is not None:\n                logger.info(\n                    \"Generating query from table batch_kwargs based on limit and offset\"\n                )\n                # In BigQuery the table name is already qualified with its schema name\n                if self.engine.dialect.name.lower() == \"bigquery\":\n                    schema = None\n                else:\n                    schema = batch_kwargs.get(\"schema\")\n                raw_query = (\n                    sqlalchemy.select([sqlalchemy.text(\"*\")])\n                    .select_from(\n                        sqlalchemy.schema.Table(\n                            table, sqlalchemy.MetaData(), schema=schema\n                        )\n                    )\n                    .offset(offset)\n                    .limit(limit)\n                )\n                query = str(\n                    raw_query.compile(\n                        self.engine, compile_kwargs={\"literal_binds\": True}\n                    )\n                )\n                batch_reference = SqlAlchemyBatchReference(\n                    engine=self.engine,\n                    query=query,\n                    table_name=query_support_table_name,\n                    schema=batch_kwargs.get(\"schema\"),\n                )\n            else:\n                batch_reference = SqlAlchemyBatchReference(\n                    engine=self.engine,\n                    table_name=table,\n                    schema=batch_kwargs.get(\"schema\"),\n                )\n        else:\n            raise ValueError(\n                \"Invalid batch_kwargs: exactly one of 'table' or 'query' must be specified\"\n            )\n\n        return Batch(\n            datasource_name=self.name,\n            batch_kwargs=batch_kwargs,\n            data=batch_reference,\n            batch_parameters=batch_parameters,\n            batch_markers=batch_markers,\n            data_context=self._data_context,\n        )\n\n    def process_batch_parameters(\n        self, query_parameters=None, limit=None, dataset_options=None\n    ):\n        batch_kwargs = super(SqlAlchemyDatasource, self).process_batch_parameters(\n            limit=limit, dataset_options=dataset_options,\n        )\n        nested_update(batch_kwargs, {\"query_parameters\": query_parameters})\n        return batch_kwargs\n", "path": "great_expectations/datasource/sqlalchemy_datasource.py"}]}