title: Airflow - Trying to execute a set of Python functions
body: <p>I am trying to execute a Airflow script that consists of a couple of Python functions. These functions basically query a database and perform few tasks. I am trying to execute this is Airflow so that I would be able to monitor each of these functions seperately. Given below is the code I am trying to execute and get the below error </p>
  
  <p><strong>Subtask: NameError: name 'task_instance' is not defined</strong></p>
  
  <pre><code>## Third party Library Imports
  
  import psycopg2
  import airflow
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  from datetime import datetime, timedelta
  from sqlalchemy import create_engine
  import io
  
  
  # Following are defaults which can be overridden later on
  default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime(2018, 1, 23, 12),
  'email': ['airflow@airflow.com'],
  'email_on_failure': False,
  'email_on_retry': False,
  'retries': 1,
  'retry_delay': timedelta(minutes=5),
  }
  
  dag = DAG('sample_dag', default_args=default_args, catchup=False, schedule_interval="@once")
  
  
  #######################
  ## Login to DB
  
  
  def db_log(**kwargs):
      global db_con
      try:
      db_con = psycopg2.connect(
  " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = 'port' sslmode = 'require' ")
      except:
          print("Connection Failed.")
          print('Connected successfully')
          task_instance = kwargs['task_instance']
          task_instance.xcom_push(value="db_con", key="db_log")
          return (db_con)
  
  def insert_data(**kwargs):
      v1 = task_instance.xcom_pull(key="db_con", task_ids='db_log')
      return (v1)
      cur = db_con.cursor()
      cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")
  
  #def job_run():
  #    db_log()
  #    insert_data()
  
  
  ##########################################
  
  t1 = PythonOperator(
  task_id='Connect',
  python_callable=db_log,provide_context=True,
  dag=dag)
  
  t2 = PythonOperator(
  task_id='Query',
  python_callable=insert_data,provide_context=True,
  dag=dag)
  
  
  t1 &gt;&gt; t2
  </code></pre>
  
  <p>Could anyone assist on this. Thanks..</p>
  
  <p><strong>Update 1 :</strong></p>
  
  <p><em>Encountered an error</em> </p>
  
  <pre><code>AttributeError: 'NoneType' object has no attribute 'execute'
  </code></pre>
  
  <p>pointing to the last line on the above piece of code </p>
  
  <pre><code>cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")
  </code></pre>
  
  <p><strong>Complete code</strong></p>
  
  <p>Complete code:</p>
  
  <pre><code>## Third party Library Imports
  import pandas as pd
  import psycopg2
  import airflow
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  from airflow.operators.bash_operator import BashOperator
  from datetime import datetime, timedelta
  from sqlalchemy import create_engine
  import io
  
  # Following are defaults which can be overridden later on
  default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime(2018, 5, 29, 12),
  'email': ['airflow@airflow.com']
  }
  
  dag = DAG('sample1', default_args=default_args)
  
  ## Login to DB
  
  def db_log(**kwargs):
    global db_con
    try:
      db_con = psycopg2.connect(
         " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
    except:
      print("I am unable to connect")
      print('Connection Task Complete')
      task_instance = kwargs['task_instance']
      task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
      return (dwh_connection)
  
  
  
  t1 = PythonOperator(
    task_id='DWH_Connect',
    python_callable=data_warehouse_login,provide_context=True,
    dag=dag)
  
  #######################
  
  def insert_data(**kwargs):
    task_instance = kwargs['task_instance']
    db_con_xcom = task_instance.xcom_pull(key="dwh_connection", task_ids='DWH_Connect')
    cur = db_con_xcom
    cur.execute("""insert into tbl_1 select limit 2 """)
  
  
  ##########################################
  
  t2 = PythonOperator(
    task_id='DWH_Connect1',
    python_callable=insert_data,provide_context=True,dag=dag)
  
  t1 &gt;&gt; t2
  </code></pre>