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

Question

[Игорь Статкевич]

Использую библиотеку watchdog для python следит за изменениями в каталоге, при изменения порождает событие, как мне при появлении события поймать его в asyncio и дальше уже, что то сделать с ним.

Answer 1

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

Постановка задачи класса "пойди туда не знаю куда", но если в общих чертах, то:

import asyncio
from contextlib import suppress
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


async def some_async_handler(event):
    # дальше уже, что то сделать с ним


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

    def on_any_event(self, event):
        asyncio.run_coroutine_threadsafe(some_async_handler(event), self._loop)


loop = asyncio.get_event_loop()

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

try:
    loop.run_forever()
except KeyboardInterrupt:
    observer.stop()
    for task in asyncio.Task.all_tasks():
        task.cancel()
        with suppress(asyncio.CancelledError):
            loop.run_until_complete(task)
finally:
    loop.close()
    observer.join()

Comments

[Игорь Статкевич]

Добрый человек подскажи, немного другая проблема (сам виноват как сказано так и понято), мне необходимо немного другое, есть карутина которая ожидает сигнала от 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 не прокатит. А "янус" потокобезопасен?

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

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