title: Trigger dag via file watcher in airflow
body: <p>I came accross this article using the watchdog API and it seemed like exactly what I need:
  <a href="https://medium.com/@phanikumaryadavilli/hacking-apache-airflow-to-trigger-dags-based-on-filesystem-events-25f822fd08c3" rel="nofollow noreferrer">https://medium.com/@phanikumaryadavilli/hacking-apache-airflow-to-trigger-dags-based-on-filesystem-events-25f822fd08c3</a></p>
  <p>(code was not written by me)</p>
  <pre><code>import os
  import time
  from airflow import DAG
  from airflow.operators.dagrun_operator import TriggerDagRunOperator
  from watchdog.observers import Observer
  from watchdog.events import FileSystemEventHandler
  from datetime import datetime, timedelta
  
  
  dag = DAG(dag_id=&quot;test_trigger_dag_operator&quot;,default_args={&quot;owner&quot;:&quot;Airflow&quot;, &quot;start_date&quot;:datetime(2020,3,9)})
  
  trigger = TriggerDagRunOperator(
      task_id=&quot;test_trigger_dag_run_operator&quot;,
      trigger_dag_id=&quot;dummy_operator&quot;,
      conf={&quot;message&quot;: &quot;Hello World&quot;},
      dag=dag,
      )
  
  class Handler(FileSystemEventHandler):
      def on_created(self, event):
          if event.event_type == 'created':
              print(&quot;file created&quot;)
              print('Executing the dag')
              trigger
  
  def main():
      observer = Observer()
      event_handler = FileSystemEventHandler()
      observer_path = os.getcwd()
      observer.schedule(Handler(), observer_path, recursive=False)
      observer.start()
  
      try:
          while True:
              time.sleep(1)
      except KeyboardInterrupt:
          observer.stop()
      
      observer.join()
  
  if __name__ == '__main__':
      main()
  </code></pre>
  <p>Unfortunately using the authors code the only thing the dag does is to unconditionally trigger the target dag and <code>main()</code> never gets called, i.e. there is also no file-watching.</p>
  <p>I made some slight modifications of the code adding the <code>python_callable</code> attribute in the <code>TriggerDagRunOperator</code> as well as adding the necessary args to main (<code>context, dag_run_obj</code>)</p>
  <pre><code>trigger = TriggerDagRunOperator(
      task_id=&quot;test_trigger_dag_run_operator&quot;,
      trigger_dag_id=&quot;dummy_operator&quot;,
      conf={&quot;message&quot;: &quot;Hello World&quot;},
      python_callable: main,
      dag=dag,
      )
  </code></pre>
  <p>and removing the</p>
  <pre><code>if __name__ == '__main__':
      main()
  </code></pre>
  <p>part.</p>
  <p>Now the file-watchdog is working, still the target-dag is triggered once in any case and the scheduler hangs as soon as starting the dag. (Which is kinda what one would expect with <code>while (true)</code>) How can I use the provided code in a working manner?</p>