title: Apache Airflow - trigger/schedule DAG rerun on completion (File Sensor)
body: <p>Good Morning.</p>
  
  <p>I'm trying to setup a DAG too</p>
  
  <ol>
  <li>Watch/sense for a file to hit a network folder</li>
  <li>Process the file</li>
  <li>Archive the file</li>
  </ol>
  
  <p>Using the tutorials online and stackoverflow I have been able to come up with the following DAG and Operator that successfully achieves the objectives, however I would like the DAG to be rescheduled or rerun on completion so it starts watching/sensing for another file.</p>
  
  <p>I attempted to set a variable <code>max_active_runs:1</code> and then a <code>schedule_interval: timedelta(seconds=5)</code> this yes reschedules the DAG but starts queuing task and locks the file.</p>
  
  <p>Any ideas welcome on how I could rerun the DAG after the archive_task?</p>
  
  <p>Thanks</p>
  
  <p>DAG CODE</p>
  
  <pre><code>from airflow import DAG
  from airflow.operators import PythonOperator, OmegaFileSensor, ArchiveFileOperator
  from datetime import datetime, timedelta
  from airflow.models import Variable
  
  default_args = {
      'owner': 'glsam',
      'depends_on_past': False,
      'start_date': datetime.now(),
      'provide_context': True,
      'retries': 100,
      'retry_delay': timedelta(seconds=30),
      'max_active_runs': 1,
      'schedule_interval': timedelta(seconds=5),
  }
  
  dag = DAG('test_sensing_for_a_file', default_args=default_args)
  
  filepath = Variable.get("soucePath_Test")
  filepattern = Variable.get("filePattern_Test")
  archivepath = Variable.get("archivePath_Test")
  
  sensor_task = OmegaFileSensor(
      task_id='file_sensor_task',
      filepath=filepath,
      filepattern=filepattern,
      poke_interval=3,
      dag=dag)
  
  
  def process_file(**context):
      file_to_process = context['task_instance'].xcom_pull(
          key='file_name', task_ids='file_sensor_task')
      file = open(filepath + file_to_process, 'w')
      file.write('This is a test\n')
      file.write('of processing the file')
      file.close()
  
  
  proccess_task = PythonOperator(
      task_id='process_the_file', 
      python_callable=process_file,
      provide_context=True,
      dag=dag
  )
  
  archive_task = ArchiveFileOperator(
      task_id='archive_file',
      filepath=filepath,
      archivepath=archivepath,
      dag=dag)
  
  sensor_task &gt;&gt; proccess_task &gt;&gt; archive_task
  </code></pre>
  
  <p>FILE SENSOR OPERATOR</p>
  
  <pre><code>    import os
      import re
  
      from datetime import datetime
      from airflow.models import BaseOperator
      from airflow.plugins_manager import AirflowPlugin
      from airflow.utils.decorators import apply_defaults
      from airflow.operators.sensors import BaseSensorOperator
  
  
      class ArchiveFileOperator(BaseOperator):
          @apply_defaults
          def __init__(self, filepath, archivepath, *args, **kwargs):
              super(ArchiveFileOperator, self).__init__(*args, **kwargs)
              self.filepath = filepath
              self.archivepath = archivepath
  
          def execute(self, context):
              file_name = context['task_instance'].xcom_pull(
                  'file_sensor_task', key='file_name')
              os.rename(self.filepath + file_name, self.archivepath + file_name)
  
  
      class OmegaFileSensor(BaseSensorOperator):
          @apply_defaults
          def __init__(self, filepath, filepattern, *args, **kwargs):
              super(OmegaFileSensor, self).__init__(*args, **kwargs)
              self.filepath = filepath
              self.filepattern = filepattern
  
          def poke(self, context):
              full_path = self.filepath
              file_pattern = re.compile(self.filepattern)
  
              directory = os.listdir(full_path)
  
              for files in directory:
                  if re.match(file_pattern, files):
                      context['task_instance'].xcom_push('file_name', files)
                      return True
              return False
  
  
      class OmegaPlugin(AirflowPlugin):
          name = "omega_plugin"
          operators = [OmegaFileSensor, ArchiveFileOperator]
  </code></pre>