Как объединить 2 потока в Asyncio?

Question

[evamaliko]

Есть 2 скрипта:
1- Observer - получает событие об изменении файла и читает сам файл.
Пример: В директории 'C:\ForTest' лежит файл text.txt
Если мы что-то напишем в в файл text.txt и сохраним его, то в консоль должно выводиться: путь к файлу, который изменен, и текст из файла.
2- Стандартный бот на AIOGram
Как объединить 2 потока, чтобы при работающем боте, работал скрипт Observer?

Пробовала вот так, через добавление задачи, до полной реализации, но при этом бот работает, а файлы не читаются:

import asyncio
import logging
from datetime import datetime

from aiogram import Bot, Dispatcher, types, html, F
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.filters import Command, CommandObject
from aiogram.types import FSInputFile, URLInputFile, BufferedInputFile
from aiogram.utils.markdown import hide_link

from config import config
import Handlers
from sqlFile import db, sql, create_sql

##########
import time

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

##########

bot = Bot(token=config['token'], parse_mode="HTML")
storage: MemoryStorage = MemoryStorage()
dp = Dispatcher(storage=storage)
logging.basicConfig(level=logging.INFO)
create_sql()
loop = asyncio.get_event_loop()


##########
class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.txt", "*.jpg"]

    def process(self, event):
        """
        event.event_type
            'modified' | 'created' | 'moved' | 'deleted'
        event.is_directory
            True | False
        event.src_path
            path/to/observed/file
        """
        # print(event.src_path, event.event_type)
        out = str(event.src_path)
        with open(out, "r") as t1:
            text = t1.read()
        print(out)
        print(text)

    def on_modified(self, event):
        self.process(event)

    def on_created(self, event):
        self.process(event)


##########
async def observer_run():
    args = 'C:\\ForTest'
    observer = Observer()
    observer.schedule(MyHandler(), path=args if args else '.')
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


async def main():
    dp.include_router(Handlers.user_router)
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())
    loop.run_until_complete(observer_run())

Answer 1

[Taulan Khatuaev]

import asyncio
import logging
import time

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

from aiogram import Bot, Dispatcher, MemoryStorage

from config import config
import Handlers
from sqlFile import db, sql, create_sql

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.txt", "*.jpg"]

    def process(self, event):
        out = str(event.src_path)
        with open(out, "r") as t1:
            text = t1.read()
        print(out)
        print(text)

    def on_modified(self, event):
        self.process(event)

    def on_created(self, event):
        self.process(event)

async def observer_run():
    args = 'C:\\ForTest'
    observer = Observer()
    observer.schedule(MyHandler(), path=args if args else '.')
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

async def main():
    bot = Bot(token=config['token'], parse_mode="HTML")
    storage = MemoryStorage()
    dp = Dispatcher(storage=storage)
    logging.basicConfig(level=logging.INFO)
    create_sql()

    dp.include_router(Handlers.user_router)
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(asyncio.gather(main(), observer_run()))

Comments

[evamaliko]

Выдает ошибку на эту строку: asyncio.run(asyncio.gather(main(), observer_run()))

File "C:\Users\Eva\AppData\Local\Programs\Python\Python38\lib\asyncio\runners.py", line 37, in run
raise ValueError("a coroutine was expected, got {!r}".format(main))
ValueError: a coroutine was expected, got <_GatheringFuture pending>