PostgresSqlHook needs to override DbApiHook.get_uri to pull in extra for client_encoding=utf-8 during create_engine

[vumdao]

Description

I got following error

[2021-11-09, 08:25:30 UTC] {base.py:70} INFO - Using connection to: id: rdb_conn_id. Host: rdb, Port: 5432, Schema: configuration, Login: user, Password: ***, extra: {'sslmode': 'allow', 'client_encoding': 'utf8'}
[2021-11-09, 08:25:30 UTC] {taskinstance.py:1703} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task
    result = execute_callable(context=context)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 151, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 162, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/repo/dags/run_configuration.py", line 34, in run_job
    dagsUtils.run_step_insert_to_temp_table(tenant, job_name, table_name, job_type)
  File "/opt/airflow/dags/rev-e3db01f68e7979d71d12ae24008a97065db2144f/dags/utils/dag_util.py", line 106, in run_step_insert_to_temp_table
    for df in df_result:
  File "/home/airflow/.local/lib/python3.9/site-packages/pandas/io/sql.py", line 1499, in _query_iterator
    data = result.fetchmany(chunksize)
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1316, in fetchmany
    self.connection._handle_dbapi_exception(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1514, in _handle_dbapi_exception
    util.raise_(exc_info[1], with_traceback=exc_info[2])
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1311, in fetchmany
    l = self.process_rows(self._fetchmany_impl(size))
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1224, in _fetchmany_impl
    return self.cursor.fetchmany(size)
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe2 in position 12: ordinal not in range(128)

Try to set extra in airflow connection but it does not work

Use case/motivation

See that airflow/providers/mysql/hooks support getting extra configs from airflow connection #6816 but not yet for postgresql hook

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[subkanthi]

Just copying the same get_uri to postgres.py seems to like duplicate logic, should we have a common base class for both postgres and mysql instead of deriving from DbApiHook?

[uranusjr]

Honestly just copying the code does not feel that big an issue to me. They are not really logically duplicates but more like “we need to do it for two different things and it just happens to be the same solution”.

[subkanthi]

I can add a quick fix for this.

[potiuk]

Honestly just copying the code does not feel that big an issue to me. They are not really logically duplicates but more like “we need to do it for two different things and it just happens to be the same solution”.

Quite agree - also "common" code between Postgres and MySQL would have to leave in "core" which will add extra dependency between both providers and airflow version (unnecessarily)