Как соединить redis и телеграм бота?

Question

[Elvis]

К примеру telethon, но в принципе можно и другую библиотеку, просто эта привычнее.
Суть в чем: есть отдельный скрипт, который по вебсокету получает данные и заносит в базу. Когда приходят определенные данные нужно их положить в redis. эта часть не проблема.
Далее бот должен постоянно слушать ключ и при появлении данных присылать в телеграм информацию. Но он так же должен и выполнять команды.
Как я понял в redis нужно подписаться на ключ и в цикле проверять get_message()
Однако я не могу запустить вечный цикл, так как бот работать не будет. Отсюда, вроде как, вывод - нужно запустить их в разных потоках, но, как я понял, в telethon это сделать не просто, чтобы стабильно работало(или невозможно).
Посоветуйте каким способом можно подружить их.

Comments

[shurshur]

telethon асинхронная библиотека, соответственно, если решить проблему с асинхронным взаимодействием с redis, то никакие потоки не потребуются...

Пишут, что redis-py с версии 4.2.0 поддерживает asyncio...

[Totem]

shurshur, aioredis все нормально, не надо париться с версиями, причем не проверенными

[shurshur]

InternetMaster, aioredis сам и пишет, что он теперь есть в составе redis-py.

[Elvis]

shurshur, redis-py выглядит тем что нужно. Теперь осталось понять как одновременно запустить и бота и бесконечный цикл на SUBSCRIBE redis.

[shurshur]

Elvis, смотреть в сторону create_task. Ну или как-то так:

async def func1():
    do_something_awesome

async def func2():
    do_something_awesome

async def main()
    result1 = func1()
    result2 = func2()
    await result1
    await result2

asyncio.run(main())

[Elvis]

shurshur, что-то всё равно не могу запустить

import redis
import asyncio
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils.executor import start_webhook
from aiogram.contrib.middlewares.logging import LoggingMiddleware
import config as cfg

bot = Bot(token=cfg.bot_token)
dp = Dispatcher(bot)
dp.middleware.setup(LoggingMiddleware())
WEBHOOK_HOST = cfg.WEBHOOK_HOST
WEBHOOK_PATH = cfg.WEBHOOK_PATH
WEBHOOK_URL = f'{WEBHOOK_HOST}{WEBHOOK_PATH}'
WEBAPP_HOST = cfg.WEBAPP_HOST
WEBAPP_PORT = cfg.WEBAPP_PORT

async def on_startup(dispatcher: Dispatcher):
    await bot.set_webhook(WEBHOOK_URL, drop_pending_updates=True)

async def on_shutdown(dispatcher: Dispatcher):
    await bot.delete_webhook()


@dp.message_handler()
async def echo(message: types.Message):
    await message.answer(message.text)


async def botrun():
    start_webhook(
        dispatcher=dp,
        webhook_path=WEBHOOK_PATH,
        skip_updates=True,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        host=WEBAPP_HOST,
        port=WEBAPP_PORT,
    )


async def redisrun():
    r = redis.Redis(host='localhost', port=6379, db=0)
    p = r.pubsub()
    p.subscribe('test')
    while True:
        mes = p.get_message()
        if mes:
            print(mes['data'])
        await asyncio.sleep(0.5)


async def main():
    f1 = asyncio.create_task(botrun())
    f2 = asyncio.create_task(redisrun())
    await f1
    await f2

if __name__ == '__main__':
    asyncio.run(main())

(bot) xxx@xxx:/home/bot/ldbot# python ldbotaio.py
1
Traceback (most recent call last):
  File "ldbotaio.py", line 60, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "ldbotaio.py", line 56, in main
    await f1
  File "ldbotaio.py", line 38, in botrun
    port=WEBAPP_PORT,
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 107, in start_webhook
    route_name=route_name)
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 78, in set_webhook
    executor.set_webhook(webhook_path, route_name=route_name, web_app=web_app)
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 289, in set_webhook
    self.loop.run_until_complete(self._startup_webhook())
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 571, in run_until_complete
    self.run_forever()
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 526, in run_forever
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x2aaf3f3c3a58>

[shurshur]

Elvis, start_webhook - это синхронная функция, которая сама запускает event loop. Глянул в код, кажется можно ей передать параметр loop=asyncio.get_event_loop(), но не проверял.

[Elvis]

shurshur, попробовал, всё равно не хочет. Не думал что так сложно будет.

Traceback (most recent call last):
  File "ldbotaio.py", line 126, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "ldbotaio.py", line 110, in main
    await f1
  File "ldbotaio.py", line 86, in botrun
    port=WEBAPP_PORT,
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 107, in start_webhook
    route_name=route_name)
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 78, in set_webhook
    executor.set_webhook(webhook_path, route_name=route_name, web_app=web_app)
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiogram/utils/executor.py", line 289, in set_webhook
    self.loop.run_until_complete(self._startup_webhook())
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 571, in run_until_complete
    self.run_forever()
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 526, in run_forever
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running
--- Logging error ---
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/logging/handlers.py", line 70, in emit
    logging.FileHandler.emit(self, record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1131, in emit
    self.stream = self._open()
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1121, in _open
    return open(self.baseFilename, self.mode, encoding=self.encoding)
NameError: name 'open' is not defined
Call stack:
  File "/home/bot/ldbot/lib/python3.7/site-packages/aiohttp/client.py", line 347, in __del__
    self._loop.call_exception_handler(context)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 1634, in call_exception_handler
    self.default_exception_handler(context)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 1608, in default_exception_handler
    logger.error('\n'.join(log_lines), exc_info=exc_info)
Message: 'Unclosed client session\nclient_session: <aiohttp.client.ClientSession object at 0x2b66aee70b00>'
Arguments: ()

[Elvis]

shurshur, таки запустил. результат в ответе.

Answer 1

[Elvis]

С aiogram у меня так и не сраслось, но получилось с telethon

import redis
import asyncio
from telethon import TelegramClient, events
import config as cfg

bot = TelegramClient('xxx', cfg.api_id, cfg.api_hash)


@bot.on(events.NewMessage())
async def echo(event):
    await bot.send_message(event.chat.id, event.text)

async def botrun():
    await bot.start(bot_token=cfg.bot_token)
    await bot.run_until_disconnected()


async def redisrun():
    r = redis.Redis(host=cfg.REDIS_HOST, port=6379, db=0)
    p = r.pubsub()
    p.subscribe('test')
    while True:
        mes = p.get_message()
        if mes:
            print(str(mes['data']))
        await asyncio.sleep(1)


async def main():
    f1 = loop.create_task(redisrun())
    f2 = loop.create_task(botrun())
    await asyncio.wait([f1, f2])

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())