We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
ExasolHook get_pandas_df does not return pandas dataframe but None

When calling the exasol hooks get_pandas_df function (https://github.com/apache/airflow/blob/main/airflow/providers/exasol/hooks/exasol.py) I noticed that it does not return a pandas dataframe. It returns None. In fact the function definition type hint explicitly states that None is returned. But the name of the function suggests otherwise. The name get_pandas_df implies that it should return a dataframe and not None.

I think that it would make more sense if get_pandas_df would indeed return a dataframe as the name is alluring to. So the code should be like this:

`def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> pd.DataFrame:
... some code ...
with closing(self.get_conn()) as conn:
df=conn.export_to_pandas(sql, query_params=parameters, **kwargs)
return df`

INSTEAD OF:

`def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:
... some code ...
with closing(self.get_conn()) as conn:
conn.export_to_pandas(sql, query_params=parameters, **kwargs)`

**Apache Airflow version**: 2.1.0


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): Not using Kubernetes

**Environment**:Official Airflow-Docker Image

- **Cloud provider or hardware configuration**: no cloud - docker host (DELL Server with 48 Cores, 512GB RAM and many TB storage)
- **OS** (e.g. from /etc/os-release):Official Airflow-Docker Image on CentOS 7 Host
- **Kernel** (e.g. `uname -a`): Linux cad18b35be00 3.10.0-1160.21.1.el7.x86_64 #1 SMP Tue Mar 16 18:28:22 UTC 2021 x86_64 GNU/Linux
- **Install tools**: only docker
- **Others**:

**What happened**:
You can replicate the findings with following dag file:

import datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.providers.exasol.operators.exasol import ExasolHook
import pandas as pd


default_args = {"owner": "airflow"}


def call_exasol_hook(**kwargs):
    #Make connection to Exasol
    hook = ExasolHook(exasol_conn_id='Exasol QA')
    sql = 'select 42;'    
    df = hook.get_pandas_df(sql = sql) 
    return df
    
with DAG(
    dag_id="exasol_hook_problem",
    start_date=datetime.datetime(2021, 5, 5),
    schedule_interval="@once",
    default_args=default_args,
    catchup=False,
) as dag:
      
    set_variable = PythonOperator(
        task_id='call_exasol_hook',
        python_callable=call_exasol_hook
    )

Sorry for the strange code formatting. I do not know how to fix this in the github UI form. 
Sorry also in case I missed something.
 
When testing or executing the task via CLI:
` airflow tasks test exasol_hook_problem call_exasol_hook 2021-07-20`

the logs show:
`[2021-07-21 12:53:19,775] {python.py:151} INFO - Done. Returned value was: None`

None was returned - although get_pandas_df was called. A pandas df should have been returned instead.



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/exasol/hooks/exasol.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 
19 from contextlib import closing
20 from typing import Any, Dict, List, Optional, Tuple, Union
21 
22 import pyexasol
23 from pyexasol import ExaConnection
24 
25 from airflow.hooks.dbapi import DbApiHook
26 
27 
28 class ExasolHook(DbApiHook):
29     """
30     Interact with Exasol.
31     You can specify the pyexasol ``compression``, ``encryption``, ``json_lib``
32     and ``client_name``  parameters in the extra field of your connection
33     as ``{"compression": True, "json_lib": "rapidjson", etc}``.
34     See `pyexasol reference
35     <https://github.com/badoo/pyexasol/blob/master/docs/REFERENCE.md#connect>`_
36     for more details.
37     """
38 
39     conn_name_attr = 'exasol_conn_id'
40     default_conn_name = 'exasol_default'
41     conn_type = 'exasol'
42     hook_name = 'Exasol'
43     supports_autocommit = True
44 
45     def __init__(self, *args, **kwargs) -> None:
46         super().__init__(*args, **kwargs)
47         self.schema = kwargs.pop("schema", None)
48 
49     def get_conn(self) -> ExaConnection:
50         conn_id = getattr(self, self.conn_name_attr)
51         conn = self.get_connection(conn_id)
52         conn_args = dict(
53             dsn=f'{conn.host}:{conn.port}',
54             user=conn.login,
55             password=conn.password,
56             schema=self.schema or conn.schema,
57         )
58         # check for parameters in conn.extra
59         for arg_name, arg_val in conn.extra_dejson.items():
60             if arg_name in ['compression', 'encryption', 'json_lib', 'client_name']:
61                 conn_args[arg_name] = arg_val
62 
63         conn = pyexasol.connect(**conn_args)
64         return conn
65 
66     def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:
67         """
68         Executes the sql and returns a pandas dataframe
69 
70         :param sql: the sql statement to be executed (str) or a list of
71             sql statements to execute
72         :type sql: str or list
73         :param parameters: The parameters to render the SQL query with.
74         :type parameters: dict or iterable
75         :param kwargs: (optional) passed into pyexasol.ExaConnection.export_to_pandas method
76         :type kwargs: dict
77         """
78         with closing(self.get_conn()) as conn:
79             conn.export_to_pandas(sql, query_params=parameters, **kwargs)
80 
81     def get_records(
82         self, sql: Union[str, list], parameters: Optional[dict] = None
83     ) -> List[Union[dict, Tuple[Any, ...]]]:
84         """
85         Executes the sql and returns a set of records.
86 
87         :param sql: the sql statement to be executed (str) or a list of
88             sql statements to execute
89         :type sql: str or list
90         :param parameters: The parameters to render the SQL query with.
91         :type parameters: dict or iterable
92         """
93         with closing(self.get_conn()) as conn:
94             with closing(conn.execute(sql, parameters)) as cur:
95                 return cur.fetchall()
96 
97     def get_first(self, sql: Union[str, list], parameters: Optional[dict] = None) -> Optional[Any]:
98         """
99         Executes the sql and returns the first resulting row.
100 
101         :param sql: the sql statement to be executed (str) or a list of
102             sql statements to execute
103         :type sql: str or list
104         :param parameters: The parameters to render the SQL query with.
105         :type parameters: dict or iterable
106         """
107         with closing(self.get_conn()) as conn:
108             with closing(conn.execute(sql, parameters)) as cur:
109                 return cur.fetchone()
110 
111     def export_to_file(
112         self,
113         filename: str,
114         query_or_table: str,
115         query_params: Optional[Dict] = None,
116         export_params: Optional[Dict] = None,
117     ) -> None:
118         """
119         Exports data to a file.
120 
121         :param filename: Path to the file to which the data has to be exported
122         :type filename: str
123         :param query_or_table: the sql statement to be executed or table name to export
124         :type query_or_table: str
125         :param query_params: Query parameters passed to underlying ``export_to_file``
126             method of :class:`~pyexasol.connection.ExaConnection`.
127         :type query_params: dict
128         :param export_params: Extra parameters passed to underlying ``export_to_file``
129             method of :class:`~pyexasol.connection.ExaConnection`.
130         :type export_params: dict
131         """
132         self.log.info("Getting data from exasol")
133         with closing(self.get_conn()) as conn:
134             conn.export_to_file(
135                 dst=filename,
136                 query_or_table=query_or_table,
137                 query_params=query_params,
138                 export_params=export_params,
139             )
140         self.log.info("Data saved to %s", filename)
141 
142     def run(self, sql: Union[str, list], autocommit: bool = False, parameters: Optional[dict] = None) -> None:
143         """
144         Runs a command or a list of commands. Pass a list of sql
145         statements to the sql parameter to get them to execute
146         sequentially
147 
148         :param sql: the sql statement to be executed (str) or a list of
149             sql statements to execute
150         :type sql: str or list
151         :param autocommit: What to set the connection's autocommit setting to
152             before executing the query.
153         :type autocommit: bool
154         :param parameters: The parameters to render the SQL query with.
155         :type parameters: dict or iterable
156         """
157         if isinstance(sql, str):
158             sql = [sql]
159 
160         with closing(self.get_conn()) as conn:
161             if self.supports_autocommit:
162                 self.set_autocommit(conn, autocommit)
163 
164             for query in sql:
165                 self.log.info(query)
166                 with closing(conn.execute(query, parameters)) as cur:
167                     self.log.info(cur.row_count)
168             # If autocommit was set to False for db that supports autocommit,
169             # or if db does not supports autocommit, we do a manual commit.
170             if not self.get_autocommit(conn):
171                 conn.commit()
172 
173     def set_autocommit(self, conn, autocommit: bool) -> None:
174         """
175         Sets the autocommit flag on the connection
176 
177         :param conn: Connection to set autocommit setting to.
178         :type conn: connection object
179         :param autocommit: The autocommit setting to set.
180         :type autocommit: bool
181         """
182         if not self.supports_autocommit and autocommit:
183             self.log.warning(
184                 "%s connection doesn't support autocommit but autocommit activated.",
185                 getattr(self, self.conn_name_attr),
186             )
187         conn.set_autocommit(autocommit)
188 
189     def get_autocommit(self, conn) -> bool:
190         """
191         Get autocommit setting for the provided connection.
192         Return True if autocommit is set.
193         Return False if autocommit is not set or set to False or conn
194         does not support autocommit.
195 
196         :param conn: Connection to get autocommit setting from.
197         :type conn: connection object
198         :return: connection autocommit setting.
199         :rtype: bool
200         """
201         autocommit = conn.attr.get('autocommit')
202         if autocommit is None:
203             autocommit = super().get_autocommit(conn)
204         return autocommit
205 
206     @staticmethod
207     def _serialize_cell(cell, conn=None) -> object:
208         """
209         Exasol will adapt all arguments to the execute() method internally,
210         hence we return cell without any conversion.
211 
212         :param cell: The cell to insert into the table
213         :type cell: object
214         :param conn: The database connection
215         :type conn: connection object
216         :return: The cell
217         :rtype: object
218         """
219         return cell
220
```
--- 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/airflow/providers/exasol/hooks/exasol.py b/airflow/providers/exasol/hooks/exasol.py
--- a/airflow/providers/exasol/hooks/exasol.py
+++ b/airflow/providers/exasol/hooks/exasol.py
@@ -19,6 +19,7 @@
 from contextlib import closing
 from typing import Any, Dict, List, Optional, Tuple, Union
 
+import pandas as pd
 import pyexasol
 from pyexasol import ExaConnection
 
@@ -63,7 +64,9 @@
         conn = pyexasol.connect(**conn_args)
         return conn
 
-    def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:
+    def get_pandas_df(
+        self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs
+    ) -> pd.DataFrame:
         """
         Executes the sql and returns a pandas dataframe
 
@@ -76,7 +79,8 @@
         :type kwargs: dict
         """
         with closing(self.get_conn()) as conn:
-            conn.export_to_pandas(sql, query_params=parameters, **kwargs)
+            df = conn.export_to_pandas(sql, query_params=parameters, **kwargs)
+            return df
 
     def get_records(
         self, sql: Union[str, list], parameters: Optional[dict] = None

{"golden_diff": "diff --git a/airflow/providers/exasol/hooks/exasol.py b/airflow/providers/exasol/hooks/exasol.py\n--- a/airflow/providers/exasol/hooks/exasol.py\n+++ b/airflow/providers/exasol/hooks/exasol.py\n@@ -19,6 +19,7 @@\n from contextlib import closing\n from typing import Any, Dict, List, Optional, Tuple, Union\n \n+import pandas as pd\n import pyexasol\n from pyexasol import ExaConnection\n \n@@ -63,7 +64,9 @@\n         conn = pyexasol.connect(**conn_args)\n         return conn\n \n-    def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:\n+    def get_pandas_df(\n+        self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs\n+    ) -> pd.DataFrame:\n         \"\"\"\n         Executes the sql and returns a pandas dataframe\n \n@@ -76,7 +79,8 @@\n         :type kwargs: dict\n         \"\"\"\n         with closing(self.get_conn()) as conn:\n-            conn.export_to_pandas(sql, query_params=parameters, **kwargs)\n+            df = conn.export_to_pandas(sql, query_params=parameters, **kwargs)\n+            return df\n \n     def get_records(\n         self, sql: Union[str, list], parameters: Optional[dict] = None\n", "issue": "ExasolHook get_pandas_df does not return pandas dataframe but None\n\r\nWhen calling the exasol hooks get_pandas_df function (https://github.com/apache/airflow/blob/main/airflow/providers/exasol/hooks/exasol.py) I noticed that it does not return a pandas dataframe. It returns None. In fact the function definition type hint explicitly states that None is returned. But the name of the function suggests otherwise. The name get_pandas_df implies that it should return a dataframe and not None.\r\n\r\nI think that it would make more sense if get_pandas_df would indeed return a dataframe as the name is alluring to. So the code should be like this:\r\n\r\n`def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> pd.DataFrame:\r\n... some code ...\r\nwith closing(self.get_conn()) as conn:\r\ndf=conn.export_to_pandas(sql, query_params=parameters, **kwargs)\r\nreturn df`\r\n\r\nINSTEAD OF:\r\n\r\n`def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:\r\n... some code ...\r\nwith closing(self.get_conn()) as conn:\r\nconn.export_to_pandas(sql, query_params=parameters, **kwargs)`\r\n\r\n**Apache Airflow version**: 2.1.0\r\n\r\n\r\n**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): Not using Kubernetes\r\n\r\n**Environment**:Official Airflow-Docker Image\r\n\r\n- **Cloud provider or hardware configuration**: no cloud - docker host (DELL Server with 48 Cores, 512GB RAM and many TB storage)\r\n- **OS** (e.g. from /etc/os-release):Official Airflow-Docker Image on CentOS 7 Host\r\n- **Kernel** (e.g. `uname -a`): Linux cad18b35be00 3.10.0-1160.21.1.el7.x86_64 #1 SMP Tue Mar 16 18:28:22 UTC 2021 x86_64 GNU/Linux\r\n- **Install tools**: only docker\r\n- **Others**:\r\n\r\n**What happened**:\r\nYou can replicate the findings with following dag file:\r\n\r\nimport datetime\r\n\r\nfrom airflow import DAG\r\nfrom airflow.operators.python_operator import PythonOperator\r\nfrom airflow.providers.exasol.operators.exasol import ExasolHook\r\nimport pandas as pd\r\n\r\n\r\ndefault_args = {\"owner\": \"airflow\"}\r\n\r\n\r\ndef call_exasol_hook(**kwargs):\r\n    #Make connection to Exasol\r\n    hook = ExasolHook(exasol_conn_id='Exasol QA')\r\n    sql = 'select 42;'    \r\n    df = hook.get_pandas_df(sql = sql) \r\n    return df\r\n    \r\nwith DAG(\r\n    dag_id=\"exasol_hook_problem\",\r\n    start_date=datetime.datetime(2021, 5, 5),\r\n    schedule_interval=\"@once\",\r\n    default_args=default_args,\r\n    catchup=False,\r\n) as dag:\r\n      \r\n    set_variable = PythonOperator(\r\n        task_id='call_exasol_hook',\r\n        python_callable=call_exasol_hook\r\n    )\r\n\r\nSorry for the strange code formatting. I do not know how to fix this in the github UI form. \r\nSorry also in case I missed something.\r\n \r\nWhen testing or executing the task via CLI:\r\n` airflow tasks test exasol_hook_problem call_exasol_hook 2021-07-20`\r\n\r\nthe logs show:\r\n`[2021-07-21 12:53:19,775] {python.py:151} INFO - Done. Returned value was: None`\r\n\r\nNone was returned - although get_pandas_df was called. A pandas df should have been returned instead.\r\n\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom contextlib import closing\nfrom typing import Any, Dict, List, Optional, Tuple, Union\n\nimport pyexasol\nfrom pyexasol import ExaConnection\n\nfrom airflow.hooks.dbapi import DbApiHook\n\n\nclass ExasolHook(DbApiHook):\n    \"\"\"\n    Interact with Exasol.\n    You can specify the pyexasol ``compression``, ``encryption``, ``json_lib``\n    and ``client_name``  parameters in the extra field of your connection\n    as ``{\"compression\": True, \"json_lib\": \"rapidjson\", etc}``.\n    See `pyexasol reference\n    <https://github.com/badoo/pyexasol/blob/master/docs/REFERENCE.md#connect>`_\n    for more details.\n    \"\"\"\n\n    conn_name_attr = 'exasol_conn_id'\n    default_conn_name = 'exasol_default'\n    conn_type = 'exasol'\n    hook_name = 'Exasol'\n    supports_autocommit = True\n\n    def __init__(self, *args, **kwargs) -> None:\n        super().__init__(*args, **kwargs)\n        self.schema = kwargs.pop(\"schema\", None)\n\n    def get_conn(self) -> ExaConnection:\n        conn_id = getattr(self, self.conn_name_attr)\n        conn = self.get_connection(conn_id)\n        conn_args = dict(\n            dsn=f'{conn.host}:{conn.port}',\n            user=conn.login,\n            password=conn.password,\n            schema=self.schema or conn.schema,\n        )\n        # check for parameters in conn.extra\n        for arg_name, arg_val in conn.extra_dejson.items():\n            if arg_name in ['compression', 'encryption', 'json_lib', 'client_name']:\n                conn_args[arg_name] = arg_val\n\n        conn = pyexasol.connect(**conn_args)\n        return conn\n\n    def get_pandas_df(self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs) -> None:\n        \"\"\"\n        Executes the sql and returns a pandas dataframe\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        :param kwargs: (optional) passed into pyexasol.ExaConnection.export_to_pandas method\n        :type kwargs: dict\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            conn.export_to_pandas(sql, query_params=parameters, **kwargs)\n\n    def get_records(\n        self, sql: Union[str, list], parameters: Optional[dict] = None\n    ) -> List[Union[dict, Tuple[Any, ...]]]:\n        \"\"\"\n        Executes the sql and returns a set of records.\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            with closing(conn.execute(sql, parameters)) as cur:\n                return cur.fetchall()\n\n    def get_first(self, sql: Union[str, list], parameters: Optional[dict] = None) -> Optional[Any]:\n        \"\"\"\n        Executes the sql and returns the first resulting row.\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            with closing(conn.execute(sql, parameters)) as cur:\n                return cur.fetchone()\n\n    def export_to_file(\n        self,\n        filename: str,\n        query_or_table: str,\n        query_params: Optional[Dict] = None,\n        export_params: Optional[Dict] = None,\n    ) -> None:\n        \"\"\"\n        Exports data to a file.\n\n        :param filename: Path to the file to which the data has to be exported\n        :type filename: str\n        :param query_or_table: the sql statement to be executed or table name to export\n        :type query_or_table: str\n        :param query_params: Query parameters passed to underlying ``export_to_file``\n            method of :class:`~pyexasol.connection.ExaConnection`.\n        :type query_params: dict\n        :param export_params: Extra parameters passed to underlying ``export_to_file``\n            method of :class:`~pyexasol.connection.ExaConnection`.\n        :type export_params: dict\n        \"\"\"\n        self.log.info(\"Getting data from exasol\")\n        with closing(self.get_conn()) as conn:\n            conn.export_to_file(\n                dst=filename,\n                query_or_table=query_or_table,\n                query_params=query_params,\n                export_params=export_params,\n            )\n        self.log.info(\"Data saved to %s\", filename)\n\n    def run(self, sql: Union[str, list], autocommit: bool = False, parameters: Optional[dict] = None) -> None:\n        \"\"\"\n        Runs a command or a list of commands. Pass a list of sql\n        statements to the sql parameter to get them to execute\n        sequentially\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param autocommit: What to set the connection's autocommit setting to\n            before executing the query.\n        :type autocommit: bool\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        if isinstance(sql, str):\n            sql = [sql]\n\n        with closing(self.get_conn()) as conn:\n            if self.supports_autocommit:\n                self.set_autocommit(conn, autocommit)\n\n            for query in sql:\n                self.log.info(query)\n                with closing(conn.execute(query, parameters)) as cur:\n                    self.log.info(cur.row_count)\n            # If autocommit was set to False for db that supports autocommit,\n            # or if db does not supports autocommit, we do a manual commit.\n            if not self.get_autocommit(conn):\n                conn.commit()\n\n    def set_autocommit(self, conn, autocommit: bool) -> None:\n        \"\"\"\n        Sets the autocommit flag on the connection\n\n        :param conn: Connection to set autocommit setting to.\n        :type conn: connection object\n        :param autocommit: The autocommit setting to set.\n        :type autocommit: bool\n        \"\"\"\n        if not self.supports_autocommit and autocommit:\n            self.log.warning(\n                \"%s connection doesn't support autocommit but autocommit activated.\",\n                getattr(self, self.conn_name_attr),\n            )\n        conn.set_autocommit(autocommit)\n\n    def get_autocommit(self, conn) -> bool:\n        \"\"\"\n        Get autocommit setting for the provided connection.\n        Return True if autocommit is set.\n        Return False if autocommit is not set or set to False or conn\n        does not support autocommit.\n\n        :param conn: Connection to get autocommit setting from.\n        :type conn: connection object\n        :return: connection autocommit setting.\n        :rtype: bool\n        \"\"\"\n        autocommit = conn.attr.get('autocommit')\n        if autocommit is None:\n            autocommit = super().get_autocommit(conn)\n        return autocommit\n\n    @staticmethod\n    def _serialize_cell(cell, conn=None) -> object:\n        \"\"\"\n        Exasol will adapt all arguments to the execute() method internally,\n        hence we return cell without any conversion.\n\n        :param cell: The cell to insert into the table\n        :type cell: object\n        :param conn: The database connection\n        :type conn: connection object\n        :return: The cell\n        :rtype: object\n        \"\"\"\n        return cell\n", "path": "airflow/providers/exasol/hooks/exasol.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom contextlib import closing\nfrom typing import Any, Dict, List, Optional, Tuple, Union\n\nimport pandas as pd\nimport pyexasol\nfrom pyexasol import ExaConnection\n\nfrom airflow.hooks.dbapi import DbApiHook\n\n\nclass ExasolHook(DbApiHook):\n    \"\"\"\n    Interact with Exasol.\n    You can specify the pyexasol ``compression``, ``encryption``, ``json_lib``\n    and ``client_name``  parameters in the extra field of your connection\n    as ``{\"compression\": True, \"json_lib\": \"rapidjson\", etc}``.\n    See `pyexasol reference\n    <https://github.com/badoo/pyexasol/blob/master/docs/REFERENCE.md#connect>`_\n    for more details.\n    \"\"\"\n\n    conn_name_attr = 'exasol_conn_id'\n    default_conn_name = 'exasol_default'\n    conn_type = 'exasol'\n    hook_name = 'Exasol'\n    supports_autocommit = True\n\n    def __init__(self, *args, **kwargs) -> None:\n        super().__init__(*args, **kwargs)\n        self.schema = kwargs.pop(\"schema\", None)\n\n    def get_conn(self) -> ExaConnection:\n        conn_id = getattr(self, self.conn_name_attr)\n        conn = self.get_connection(conn_id)\n        conn_args = dict(\n            dsn=f'{conn.host}:{conn.port}',\n            user=conn.login,\n            password=conn.password,\n            schema=self.schema or conn.schema,\n        )\n        # check for parameters in conn.extra\n        for arg_name, arg_val in conn.extra_dejson.items():\n            if arg_name in ['compression', 'encryption', 'json_lib', 'client_name']:\n                conn_args[arg_name] = arg_val\n\n        conn = pyexasol.connect(**conn_args)\n        return conn\n\n    def get_pandas_df(\n        self, sql: Union[str, list], parameters: Optional[dict] = None, **kwargs\n    ) -> pd.DataFrame:\n        \"\"\"\n        Executes the sql and returns a pandas dataframe\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        :param kwargs: (optional) passed into pyexasol.ExaConnection.export_to_pandas method\n        :type kwargs: dict\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            df = conn.export_to_pandas(sql, query_params=parameters, **kwargs)\n            return df\n\n    def get_records(\n        self, sql: Union[str, list], parameters: Optional[dict] = None\n    ) -> List[Union[dict, Tuple[Any, ...]]]:\n        \"\"\"\n        Executes the sql and returns a set of records.\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            with closing(conn.execute(sql, parameters)) as cur:\n                return cur.fetchall()\n\n    def get_first(self, sql: Union[str, list], parameters: Optional[dict] = None) -> Optional[Any]:\n        \"\"\"\n        Executes the sql and returns the first resulting row.\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        with closing(self.get_conn()) as conn:\n            with closing(conn.execute(sql, parameters)) as cur:\n                return cur.fetchone()\n\n    def export_to_file(\n        self,\n        filename: str,\n        query_or_table: str,\n        query_params: Optional[Dict] = None,\n        export_params: Optional[Dict] = None,\n    ) -> None:\n        \"\"\"\n        Exports data to a file.\n\n        :param filename: Path to the file to which the data has to be exported\n        :type filename: str\n        :param query_or_table: the sql statement to be executed or table name to export\n        :type query_or_table: str\n        :param query_params: Query parameters passed to underlying ``export_to_file``\n            method of :class:`~pyexasol.connection.ExaConnection`.\n        :type query_params: dict\n        :param export_params: Extra parameters passed to underlying ``export_to_file``\n            method of :class:`~pyexasol.connection.ExaConnection`.\n        :type export_params: dict\n        \"\"\"\n        self.log.info(\"Getting data from exasol\")\n        with closing(self.get_conn()) as conn:\n            conn.export_to_file(\n                dst=filename,\n                query_or_table=query_or_table,\n                query_params=query_params,\n                export_params=export_params,\n            )\n        self.log.info(\"Data saved to %s\", filename)\n\n    def run(self, sql: Union[str, list], autocommit: bool = False, parameters: Optional[dict] = None) -> None:\n        \"\"\"\n        Runs a command or a list of commands. Pass a list of sql\n        statements to the sql parameter to get them to execute\n        sequentially\n\n        :param sql: the sql statement to be executed (str) or a list of\n            sql statements to execute\n        :type sql: str or list\n        :param autocommit: What to set the connection's autocommit setting to\n            before executing the query.\n        :type autocommit: bool\n        :param parameters: The parameters to render the SQL query with.\n        :type parameters: dict or iterable\n        \"\"\"\n        if isinstance(sql, str):\n            sql = [sql]\n\n        with closing(self.get_conn()) as conn:\n            if self.supports_autocommit:\n                self.set_autocommit(conn, autocommit)\n\n            for query in sql:\n                self.log.info(query)\n                with closing(conn.execute(query, parameters)) as cur:\n                    self.log.info(cur.row_count)\n            # If autocommit was set to False for db that supports autocommit,\n            # or if db does not supports autocommit, we do a manual commit.\n            if not self.get_autocommit(conn):\n                conn.commit()\n\n    def set_autocommit(self, conn, autocommit: bool) -> None:\n        \"\"\"\n        Sets the autocommit flag on the connection\n\n        :param conn: Connection to set autocommit setting to.\n        :type conn: connection object\n        :param autocommit: The autocommit setting to set.\n        :type autocommit: bool\n        \"\"\"\n        if not self.supports_autocommit and autocommit:\n            self.log.warning(\n                \"%s connection doesn't support autocommit but autocommit activated.\",\n                getattr(self, self.conn_name_attr),\n            )\n        conn.set_autocommit(autocommit)\n\n    def get_autocommit(self, conn) -> bool:\n        \"\"\"\n        Get autocommit setting for the provided connection.\n        Return True if autocommit is set.\n        Return False if autocommit is not set or set to False or conn\n        does not support autocommit.\n\n        :param conn: Connection to get autocommit setting from.\n        :type conn: connection object\n        :return: connection autocommit setting.\n        :rtype: bool\n        \"\"\"\n        autocommit = conn.attr.get('autocommit')\n        if autocommit is None:\n            autocommit = super().get_autocommit(conn)\n        return autocommit\n\n    @staticmethod\n    def _serialize_cell(cell, conn=None) -> object:\n        \"\"\"\n        Exasol will adapt all arguments to the execute() method internally,\n        hence we return cell without any conversion.\n\n        :param cell: The cell to insert into the table\n        :type cell: object\n        :param conn: The database connection\n        :type conn: connection object\n        :return: The cell\n        :rtype: object\n        \"\"\"\n        return cell\n", "path": "airflow/providers/exasol/hooks/exasol.py"}]}