Can't use Postgresql Connection ID to connect DB and create DAG

[donglilang]

Hello All,

I deployed the Openmetadata environment in the Docker. The airflow version is 2.9.1
I can create the airflow connection with UI, but can't list with CLI.
So I create one with the CLI:
airflow@68713c565cc0:/opt/airflow$ airflow connections add 'AF_POSTGRES_CONN' \

--conn-json '{

    "conn_type": "postgres",

    "login": "openmetadata_user",

    "password": "openmetadata_password",

    "host": "172.16.240.3",

    "port": 5432,

    "schema": "openmetadata_db"

}'

[2024-12-02T00:57:35.581+0000] {crypto.py:82} WARNING - empty cryptography key - values will not be stored encrypted.

Successfully added conn_id=AF_POSTGRES_CONN : postgres://openmetadata_user:******@172.16.240.3:5432/openmetadata_db

airflow@68713c565cc0:/opt/airflow$ airflow connections list

id | conn_id | conn_type | description | host | schema | login | password | port | is_encrypted | is_extra_encrypted | extra_dejson | get_uri

===+==================+===========+=============+==============+=================+===================+=======================+======+==============+====================+==============+====================================

1 | AF_POSTGRES_CONN | postgres | None | 172.16.240.3 | openmetadata_db | openmetadata_user | openmetadata_password | 5432 | False | False | {} | postgres://openmetadata_user:openme

| | | | | | | | | | | | [email protected]:5432/o

| | | | | | | | | | | | penmetadata_db

When I want to create DAG with Python, It raised this error message:

airflow@68713c565cc0:/opt/airflow/dags$ python ./demo_02.py

Traceback (most recent call last):

File "/opt/airflow/dags/./demo_02.py", line 35, in

create_table = PostgresOperator(

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 484, in apply_defaults

result = func(self, **kwargs, default_args=default_args)

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/common/sql/operators/sql.py", line 232, in init

super().__init__(conn_id=conn_id, database=database, **kwargs)

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 484, in apply_defaults

result = func(self, **kwargs, default_args=default_args)

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/common/sql/operators/sql.py", line 142, in init

super().__init__(**kwargs)

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 484, in apply_defaults

result = func(self, **kwargs, default_args=default_args)

File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 881, in init

raise AirflowException(

airflow.exceptions.AirflowException: Invalid arguments were passed to SQLExecuteQueryOperator (task_id: create_table). Invalid arguments were:

**kwargs: {'postgres_conn_id': 'AF_POSTGRES_CONN'}

airflow@68713c565cc0:/opt/airflow/dags$ cat demo_02.py

from datetime import datetime, timedelta

from airflow import DAG

#from airflow.providers.postgres.operators.postgres import PostgresOperator

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator as PostgresOperator

from airflow.providers.postgres.hooks.postgres import PostgresHook

from airflow.operators.python import PythonOperator

Define the default arguments for the DAG

default_args = {

'owner': 'airflow',

'depends_on_past': False,

'start_date': datetime(2022, 1, 1),

'retries': 1,

'retry_delay': timedelta(minutes=5),

}

Define the DAG

dag = DAG(

'postgresql_pipeline',

default_args=default_args,

description='A simple DAG to interact with PostgreSQL',

schedule=timedelta(days=1),

)

Define a Python function to insert records using PostgresHook

def insert_records_with_hook():

hook = PostgresHook(postgres_conn_id='AF_POSTGRES_CONN',schema='public')

conn = hook.get_conn()

cursor = conn.cursor()

cursor.execute("INSERT INTO your_table (name) VALUES ('record1'), ('record2');")

conn.commit()

conn.close()

Define the tasks

create_table = PostgresOperator(

task_id='create_table',

postgres_conn_id='AF_POSTGRES_CONN',

sql="CREATE TABLE your_table (id serial PRIMARY KEY, name VARCHAR);",

dag=dag,

)

insert_records_task = PythonOperator(

task_id='insert_records',

python_callable=insert_records_with_hook,

dag=dag,

)

Set up the task dependencies

create_table >> insert_records_task

Is there any issue for this? or who can give me one sample code to finish my taks?

Thanks a lot

Tony