atul-astronomer opened a new issue, #49761:
URL: https://github.com/apache/airflow/issues/49761
### Apache Airflow version
main (development)
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
ERROR - Task failed with exception: source="task"
ProgrammingError: invalid dsn: invalid connection option "a"
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 825 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1088 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 408 in
wrapper
File
"/opt/airflow/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py",
line 307 in execute
File
"/opt/airflow/providers/common/sql/src/airflow/providers/common/sql/hooks/sql.py",
line 725 in run
File "/usr/local/lib/python3.9/contextlib.py", line 119 in __enter__
File
"/opt/airflow/providers/common/sql/src/airflow/providers/common/sql/hooks/sql.py",
line 832 in _create_autocommit_connection
File
"/opt/airflow/providers/postgres/src/airflow/providers/postgres/hooks/postgres.py",
line 173 in get_conn
File "/usr/local/lib/python3.9/site-packages/psycopg2/__init__.py", line 121
in connect
File "/usr/local/lib/python3.9/site-packages/psycopg2/extensions.py", line
167 in make_dsn
### What you think should happen instead?
_No response_
### How to reproduce
Example DAG is creating a postgres connection and then a table using
SQLExecuteQueryOperator.
```python
import json
import requests
from airflow.models import DAG, Connection
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.common.sql.operators.sql import
SQLExecuteQueryOperator
from airflow.hooks.base import BaseHook
from pendulum import today
from airflow.exceptions import AirflowNotFoundException
from airflow import settings
import os
from random import choices
from string import ascii_lowercase, digits
docs = """
####Purpose
This dag tests the template '{ conn.conn_id.connection_param }' to ensure
you can grab various items with the template
####Expected Behavior
This dag has 2 tasks both of which are expected to succeed. If one or both
of the tasks fail then something is wrong with the jinja 'conn' template.\n
The 1st task defines a table with Postgres with the random connection ID.\n
The 2nd task checks to make sure that the connection values passed in stay
the same.
"""
dag_name = "test_jinja_connection_id"
POSTGRES_HOST = os.getenv("POSTGRES_HOST",
"database-1.cxmxicvi57az.us-east-2.rds.amazonaws.com")
POSTGRES_PASS = os.getenv("POSTGRES_PASS", "TestAirflow123")
def add_conn():
try:
found = BaseHook().get_connection(f"{dag_name}_connection")
except AirflowNotFoundException:
found = None
print("The connection is not defined, please add a connection in the
dags first task")
if found:
print("The connection has been made previously.")
else:
with open('dags/plugins/api_endpoints.json', 'r') as file:
data = json.load(file)
url = data['create_connection']['url']
headers = data['create_connection']['headers']
request_body = data['create_connection']['request_body']
request_body['connection_id'] = f"{dag_name}_connection"
request_body['conn_type'] = "postgres"
request_body['host'] = POSTGRES_HOST
request_body['login'] = "postgres"
request_body['schema'] = "postgres"
request_body['port'] = 5432
request_body['password'] = POSTGRES_PASS
request_body['extra'] = "{\"a\":\"10\"}"
response = requests.request(method='POST', url=url, headers=headers,
json=request_body)
assert response.json()['connection_id'] == f"{dag_name}_connection"
def conn_id_test(**context):
print(f"The connection type is: {context['get_conn_type']}")
print(f"The host is: {context['check_host']}")
print(f"The schema is: {context['get_schema']}")
print(f"The login is: {context['get_login']}")
print(f"The password is: {context['get_pass']}") # returns '***' hides
the password
print(f"The port is: {context['get_port']}")
print(f"The extras are: {context['get_extras']}")
print("asserting the connection type")
assert context["get_conn_type"] == "postgres"
print("asserting the host")
assert context["check_host"] == POSTGRES_HOST
print("asserting the schema")
assert context["get_schema"] == "postgres"
print("asserting the login")
assert context["get_login"] == "postgres"
print("asserting the port")
assert context["get_port"] == "5432"
with DAG(
dag_id=dag_name,
schedule=None,
start_date=today('UTC').add(days=-2),
doc_md=docs,
tags=["core"],
) as dag:
py0 = PythonOperator(
task_id="add_conn",
python_callable=add_conn,
)
P0 = SQLExecuteQueryOperator(
task_id="create_table_define_cols",
conn_id=f"{dag_name}_connection",
sql="""
CREATE TABLE IF NOT EXISTS jinja_connection_template_test(
random_str varchar,
herbs varchar,
primary key(herbs));
""",
)
py1 = PythonOperator(
task_id="check_jinja_conn_id",
python_callable=conn_id_test,
op_kwargs={
"get_conn_type": f"{{{{ conn.{dag_name}_connection.conn_type
}}}}",
"check_host": f"{{{{ conn.{dag_name}_connection.host }}}}",
"get_schema": f"{{{{ conn.{dag_name}_connection.schema }}}}",
"get_login": f"{{{{ conn.{dag_name}_connection.login }}}}",
"get_pass": f"{{{{ conn.{dag_name}_connection.password }}}}",
"get_port": f"{{{{ conn.{dag_name}_connection.port }}}}"
},
)
py0 >> P0 >> py1
```
### Operating System
Linux
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
