title: trying to create dynamic subdags from parent dag based on array of filenames
body: <p>I am trying to move s3 files from a "non-deleting" bucket (meaning I can't delete the files) to GCS using airflow.  I cannot be guaranteed that new files will be there everyday, but I must check for new files everyday.</p>
  
  <p>my problem is the dynamic creation of subdags.  If there ARE files, I need subdags.  If there are NOT files, I don't need subdags.  My problem is the upstream/downstream settings.  In my code, it does detect files, but does not kick off the subdags as they are supposed to.  I'm missing something.</p>
  
  <p>here's my code:</p>
  
  <pre><code>from airflow import models
  from  airflow.utils.helpers import chain
  from airflow.providers.amazon.aws.hooks.s3 import S3Hook
  from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators.subdag_operator import SubDagOperator
  from airflow.contrib.operators.s3_to_gcs_operator import S3ToGoogleCloudStorageOperator
  from airflow.utils import dates
  from airflow.models import Variable
  import logging
  
  args = {
      'owner': 'Airflow',
      'start_date': dates.days_ago(1),
      'email': ['sinistersparrow1701@gmail.com'],
      'email_on_failure': True,
      'email_on_success': True,
  }
  
  bucket = 'mybucket'
  prefix = 'myprefix/'
  LastBDEXDate = int(Variable.get("last_publish_date"))
  maxdate = LastBDEXDate
  files = []
  
  parent_dag = models.DAG(
      dag_id='My_Ingestion',
      default_args=args,
      schedule_interval='@daily',
      catchup=False
  )
  
  def Check_For_Files(**kwargs):
      s3 = S3Hook(aws_conn_id='S3_BOX')
      s3.get_conn()
      bucket = bucket
      LastBDEXDate = int(Variable.get("last_publish_date"))
      maxdate = LastBDEXDate
      files = s3.list_keys(bucket_name=bucket, prefix='myprefix/file')
      for file in files:
          print(file)
          print(file.split("_")[-2])
          print(file.split("_")[-2][-8:])  ##proves I can see a date in the file name is ok.
          maxdate = maxdate if maxdate &gt; int(file.split("_")[-2][-8:]) else int(file.split("_")[-2][-8:])
      if maxdate &gt; LastBDEXDate:
          return 'Start_Process'
      return 'finished'
  
  def create_subdag(dag_parent, dag_id_child_prefix, file_name):
      # dag params
      dag_id_child = '%s.%s' % (dag_parent.dag_id, dag_id_child_prefix)
  
      # dag
      subdag = models.DAG(dag_id=dag_id_child,
                default_args=args,
                schedule_interval=None)
  
      # operators
      s3_to_gcs_op = S3ToGoogleCloudStorageOperator(
          task_id=dag_id_child,
          bucket=bucket,
          prefix=file_name,
          dest_gcs_conn_id='GCP_Account',
          dest_gcs='gs://my_files/To_Process/',
          replace=False,
          gzip=True,
          dag=subdag)
  
  
      return subdag
  
  def create_subdag_operator(dag_parent, filename, index):
      tid_subdag = 'file_{}'.format(index)
      subdag = create_subdag(dag_parent, tid_subdag, filename)
      sd_op = SubDagOperator(task_id=tid_subdag, dag=dag_parent, subdag=subdag)
      return sd_op
  
  def create_subdag_operators(dag_parent, file_list):
      subdags = [create_subdag_operator(dag_parent, file, file_list.index(file)) for file in file_list]
      # chain subdag-operators together
      chain(*subdags)
      return subdags
  
  check_for_files = BranchPythonOperator(
      task_id='Check_for_s3_Files',
      provide_context=True,
      python_callable=Check_For_Files,
      dag=parent_dag
  )
  
  finished = DummyOperator(
      task_id='finished',
      dag=parent_dag
  )
  
  decision_to_continue = DummyOperator(
      task_id='Start_Process',
      dag=parent_dag
  )
  
  if len(files) &gt; 0:
      subdag_ops = create_subdag_operators(parent_dag, files)
      check_for_files &gt;&gt; decision_to_continue &gt;&gt; subdag_ops[0] &gt;&gt; subdag_ops[-1] &gt;&gt; finished
  
  
  check_for_files &gt;&gt; finished
  </code></pre>