Как передать watchdog событие в asyncio?

Добрый человек подскажи, немного другая проблема (сам виноват как сказано так и понято), мне необходимо немного другое, есть карутина которая ожидает сигнала от watchdog (например чтобы начать парсить лог), как мне задать в этой карутине ожидание сигнала, что-то типа

if await some_async_handler(event):
    do_some_code

мне не обязательно сам код. а просто пнуть в нужное направление
а то топчусь на месте второй день, а все должно быть элементарно

MadInc,
pip install janus

import asyncio
import janus
import sys
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


async def some_async_handler(queue):
    while True:
        try:
            event = await queue.get()
            print(event.event_type, event.src_path)
        except asyncio.CancelledError:
            break


class SomeEventHandler(FileSystemEventHandler):
    def __init__(self, queue, *args, **kwargs):
        self._queue = queue
        super().__init__(*args, **kwargs)

    def on_any_event(self, event):
        self._queue.put(event)


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else '.'

    loop = asyncio.get_event_loop()
    queue = janus.Queue(loop=loop)

    event_handler = SomeEventHandler(queue.sync_q)
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    task = loop.create_task(some_async_handler(queue.async_q))
    try:
        loop.run_until_complete(task)
    except KeyboardInterrupt:
        task.cancel()
        loop.run_until_complete(task)
        observer.stop()
    finally:
        loop.close()
        observer.join()

Я бы просто создал Future, ждал бы в asyncio-части его "прибытия". А в обработчике watchdog делал бы future.set_result(). Это если однократно, а если в цикле - надо future каждый раз новый создавать.

Сергей Тихонов, и как бы вы передавали новый future?

Сергей Горностаев,

class Tracker:
    def on_event(self):
        self.future.set_result(True)
        self.future = Future(get_event_loop())

tracker = Tracker()

async def handle():
    while True:
        await tracker.future
        do_something_useful()

Сергей Тихонов, Future не потокобезопасны. Попытка их так использовать - это верный путь к трудноуловимым ошибкам.

Тогда да, вариант с Future не прокатит. А "янус" потокобезопасен?

Сергей Тихонов, да, это очередь специально разрабатываемая для случаев эксплуатации, когда один её конец в цикле событий, а другой в потоке.