Query the Airflow metadata db from within a python operator

airflow_metadata_query.py

def set_task_status(**kwargs):
    dag_id = kwargs["dag_id"]
    task_id = kwargs["task_id"]
    start_date = kwargs["start_date"]
    end_date = kwargs["end_date"]
    
    session = settings.Session()
    print("session: ", str(session))


    # stmt = f"UPDATE task_instance SET state = 'success', try_number = 0 WHERE task_id = '{task_id}' AND dag_id = '{dag_id}' AND execution_date = '{start_date}';"
    stmt = "Select * from task_instance;"
    result = session.execute(stmt)
    print(f"Attributes - {result._metadata.keys}")
    for row in result:
        print(f"Result from task instance - {row}")


PythonOperator(
    task_id="backfill",
    python_callable=set_task_status,
    dag=dag,
    op_kwargs={
        "start_date": '{{ dag_run.conf["start_date"] if dag_run else "" }}',
        "end_date": '{{ dag_run.conf["end_date"] if dag_run else "" }}',
        "dag_id": '{{ dag_run.conf["dag_id"] if dag_run else "" }}',
        "task_id": '{{ dag_run.conf["task_id"] if dag_run and "task_id" in dag_run.conf else None }}',
    },
)