title: Airflow xcom pull only returns string
body: <p>I have an airflow pipeline where I need to get a filename from a pubsub subscription and then import that file into a cloud sql instance. I use the CloudSqlInstanceImportOperator to import the CSV file. This operator needs a body, which contains the filename and other parameters. Since I read that filename during runtime, I also have to define the body during runtime. This all works. But when I pull the body from xcom, it returns a string instead of a python dictionary. So the CloudSqlInstanceImportOperator gives me the following error (my guess is, because the body is a string and not a dictionary):</p>
  <pre><code>Traceback (most recent call last)
    File &quot;/usr/local/lib/airflow/airflow/models/taskinstance.py&quot;, line 984, in _run_raw_tas
      result = task_copy.execute(context=context
    File &quot;/usr/local/lib/airflow/airflow/contrib/operators/gcp_sql_operator.py&quot;, line 715, in execut
      self._validate_body_fields(
    File &quot;/usr/local/lib/airflow/airflow/contrib/operators/gcp_sql_operator.py&quot;, line 712, in _validate_body_field
      api_version=self.api_version).validate(self.body
    File &quot;/usr/local/lib/airflow/airflow/contrib/utils/gcp_field_validator.py&quot;, line 420, in validat
      dictionary_to_validate=body_to_validate
    File &quot;/usr/local/lib/airflow/airflow/contrib/utils/gcp_field_validator.py&quot;, line 341, in _validate_fiel
      value = dictionary_to_validate.get(field_name
  AttributeError: 'str' object has no attribute 'get
  </code></pre>
  <p>This is the code I use:</p>
  <pre><code>import json 
  import os
  from datetime import datetime, timedelta
  import ast
  from airflow import DAG
  from airflow.contrib.operators.gcs_to_gcs import GoogleCloudStorageToGoogleCloudStorageOperator
  from airflow.operators.python_operator import PythonOperator
  from airflow.operators.bash_operator import BashOperator
  from airflow.contrib.sensors.pubsub_sensor import PubSubPullSensor
  from airflow.contrib.sensors.gcs_sensor import GoogleCloudStoragePrefixSensor
  from airflow.operators.dagrun_operator import TriggerDagRunOperator
  from airflow.contrib.operators.gcp_sql_operator import CloudSqlInstanceImportOperator
  
  
  def create_dag(dag_id,default_args):
      BUCKET = &quot;{{ var.value.gp2pg_bucket }}&quot;
      GCP_PROJECT_ID = &quot;{{ var.value.gp2pg_project_id }}&quot;
      INSTANCE_NAME = &quot;{{ var.value.gp2pg_instance_name }}&quot;
  
      def define_import_body(file,**kwargs):
          import_body = {
              &quot;importContext&quot;: {
                  &quot;importUser&quot;: &quot;databasename&quot;,
                  &quot;database&quot;: &quot;databaseuser&quot;,
                  &quot;fileType&quot;: &quot;csv&quot;,
                  &quot;uri&quot;: &quot;bucketname&quot; + file,
                  &quot;csvImportOptions&quot;: {
                      &quot;table&quot;: &quot;schema.tablename&quot;,
                      &quot;columns&quot;: [&quot;columns1&quot;,
                                  &quot;column2&quot;]}
              }
          }
          task_instance = kwargs['task_instance']
          task_instance.xcom_push(key='import_body', value=import_body)
          print(import_body)
  
      def get_filename(var,**kwargs):
          message = ast.literal_eval(var)
          file = message[0].get('message').get('attributes').get('objectId')
          task_instance = kwargs['task_instance']
          task_instance.xcom_push(key='filename', value=file)
          print(file)
  
      dag = DAG(dag_id=dag_id, schedule_interval=None, default_args=default_args)
  
      with dag:
          t1 = PubSubPullSensor(task_id='pull-messages',
                                project=&quot;projectname&quot;,
                                ack_messages=True,
                                max_messages=1,
                                subscription=&quot;subscribtionname&quot;)
  
  
          message = &quot;{{ task_instance.xcom_pull() }}&quot;
  
          t2 = PythonOperator(
              task_id='get_filename',
              python_callable=get_filename,
              op_kwargs={'var': message},
              provide_context=True,
          )
  
          file = &quot;{{ task_instance.xcom_pull(task_ids='get_filename', key='filename') }}&quot;
  
          t3 = PythonOperator(
              task_id='define_import_body',
              python_callable=define_import_body,
              op_kwargs={'file': file},
              provide_context=True,
          )
  
          import_body = &quot;{{ task_instance.xcom_pull(task_ids='define_import_body', key='import_body') }}&quot;
  
          t4 = CloudSqlInstanceImportOperator(
              project_id=GCP_PROJECT_ID,
              body= import_body,
              instance=INSTANCE_NAME,
              gcp_conn_id='postgres_default',
              task_id='sql_import_task',
              validate_body=True,
          )
  
          t5 = GoogleCloudStorageToGoogleCloudStorageOperator(
              task_id='copy_files',
              source_bucket=BUCKET,
              source_object=file,
              destination_bucket=BUCKET,
              destination_object='processed/import/'+file, )
  
          t1 &gt;&gt; t2 &gt;&gt; t3 &gt;&gt; t4 &gt;&gt; t5
  
      return dag
  
  
  dags_folder = os.getenv('DAGS_FOLDER', &quot;./dags&quot;)
  flow_config = open(f'{dags_folder}/gp2pg/flow_config.json', 'r').read()
  for key, values in json.loads(flow_config).items():
      default_args = {
          &quot;owner&quot;: &quot;owner&quot;,
          &quot;start_date&quot;: datetime(2020, 1, 1),
          &quot;email&quot;: [],
          &quot;email_on_failure&quot;: False,
          &quot;email_on_retry&quot;: False,
          &quot;retries&quot;: 0,
          &quot;retry_delay&quot;: timedelta(minutes=5),
      }
  
      dag_id = f&quot;gp2pg_{key}_data_to_pg&quot;
  
      globals()[dag_id] = create_dag(dag_id, default_args)
  </code></pre>
  <p>Any idea how I could solve that problem?</p>