title: Python asyncio: how to have a producer/consumer pattern linked to a watchdog process?
body: <p>I have a infinite python-watchdog process (daemonized with <code>supervisor</code>) that trigger new events when files are created within a folder.</p>
  <p>On the other side is a asyncio queue, with 3 consumers only, waiting for new messages of created files (consumers then upload the files to a server).</p>
  <p>The code below works, but is sequential, and I just don't understand why (despite the many documentation I read about asyncio).</p>
  <p>main.py (daemonized):</p>
  <pre class="lang-py prettyprint-override"><code>paths_observer = PathsObserver()
  
  if __name__ == &quot;__main__&quot;:
      paths_observer.start()
  
      try:
          while paths_observer.is_alive():
              paths_observer.join(1)
      except KeyboardInterrupt:
          paths_observer.stop()
  
      paths_observer.join()
  </code></pre>
  <p>pathobserver.py (simplified, used to record which folders to watch):</p>
  <pre><code>from watchdog.observers import Observer
  
  class PathsObserver(Observer):
      def observe_folder(self, folder_path: str) -&gt; None:
          event_handler = DataFileHandler(path=folder_path)
          watch = self.schedule(event_handler, folder_path, recursive=True)
  </code></pre>
  <p>filehandler.py: <strong>note the import of the scheduler</strong></p>
  <pre><code>from watchdog.events import FileCreatedEvent, FileSystemEventHandler
  
  from .scheduler import scheduler
  
  class DataFileHandler(FileSystemEventHandler):
      def on_created(self, event):        
          scheduler.prepare_and_upload(event.src_path)
  
  </code></pre>
  <p>scheduler.py:</p>
  <pre><code>MAX_TASKS = 3
  
  class UploadScheduler(object):
      def __init__(self):
          self._loop = asyncio.get_event_loop()
          self._queue = asyncio.Queue(0, loop=self._loop)
          self._consumers = [asyncio.ensure_future(self._consumer(self._queue)) for _ in range(MAX_TASKS)]
  
      async def _consumer(self, queue):
          while True:
              file_path = await queue.get()
              await &lt;do some stuff with file_path&gt;
              queue.task_done()
  
      async def _producer(self, file_path):
          await self._queue.put(file_path)
  
      def prepare_and_upload(self, file_path):
          self._loop.run_until_complete(self._producer(file_path))
  
  
  scheduler = UploadScheduler()
  </code></pre>
  <p>If I remove <code>self._loop.run_until_complete</code>, tasks aren't fetch by consumers, and a message says the <code>producer</code> hasn't been waited. If I put <code>self._queue.put_no_wait(file_path)</code> in a non-async func, consumer jobs don't get started either.</p>
  <p>I am doing something wrong ?</p>
  <p>What I expected: when multiple files are created, a bunch a new producers are being called. And since consumers take a bit of time to upload everything, they slowly catch up. And the whole thing stays, and wait for new files to upload.</p>