How to pass variable in Python operator function to another one

Airflow
1

I want to get some records in the database, preprocess them to a new SQL script, and pass them to another operator
Here is my code example:

data = []

def getSomeData(**kwargs):
    global data
    conn = psycopg2.connect()

    sql = "SELECT * from table"
    
    cur = conn.cursor()
    cur.execute(sql)
    result = cur.fetchall()
    data = result
    print(data)  #success 
    conn.commit()
    cur.close()


def preprocessSQL(data: list) -> str:
    sql = """
    INSERT INTO (table2) VALUES
    """

    for i in data:
      #format here
      ....
    return sql

with DAG(dag_id="get_and_push_data",
         start_date=pendulum.datetime(2023, 1, 13),
        ) as dag:

    getDataFromDB = PythonOperator(
                task_id="get_data",
                provide_context=True,
                python_callable=getSomeData,
                op_kwargs={},
            )

sql = preprocessSQL(data)

# After that will push it into another operator

The problem is I can not get the variable in this way, I can print it in the get data but in the preprocess it is empty. How could I do?

2

Hi @christ, thanks for reaching out!

What you’re trying to do can be simplified with Astro SDK, see some helpful operators:
A. run_raw_sql operator allows you to declare any SQL statement,
B. dataframe operator allows you to run Python transformations if needed.

The results of your transformations can be then saved in your destination table.

Please take a look at the links provided above as they contain great examples of how to use the operators.

Hope this helps!