Can I get a review on whether I am getting things done correctly? Or there’s a newer way to achieve the same thing, but I somehow missed?
I don’t see any code. Where is the code?
It appears to be behind the link. Perfectly understandable IMO if you’d prefer it to be reproduced here, though.
The final iteration is this
from aiohttp import web
import discord
from telegram.ext import ApplicationBuilder
client = discord.Client(..)
async def discord_run(exit_event: asyncio.Event):
global client
logger.info("DISCORD: Starting")
asyncio.create_task(client.start(os.environ["DISCORD_TOKEN"]))
await exit_event.wait()
logger.info("DISCORD: Stopping")
await client.close()
async def telegram_run(exit_event: asyncio.Event) -> None:
application = ApplicationBuilder().token(os.environ["TELEGRAM_TOKEN"]).build()
# -- snipped handlers registration --
asyncio.create_task(
application.bot.set_webhook(
f'{os.environ["WEBHOOK_URL"]}/telegram',
allowed_updates=Update.ALL_TYPES,
secret_token=settings.SECRET_TOKEN,
)
)
async with application:
logger.info("TELEGRAM: Starting")
await application.start()
await exit_event.wait()
logger.info("TELEGRAM: Stopping")
await application.stop()
async def web_run(exit_event: asyncio.Event) -> None:
application = web.Application()
logger.info("WEBHOOK: Starting", url=os.environ["WEBHOOK_URL"])
web_runner = web.AppRunner(application)
await web_runner.setup()
web_site = web.TCPSite(web_runner, port=8080)
await web_site.start()
await exit_event.wait()
logger.info("WEBHOOK: Stopping")
await web_site.stop()
await web_runner.cleanup()
def exception_handler(
loop: asyncio.AbstractEventLoop, context: dict[str, Any], exit_event: asyncio.Event
) -> None:
message = context.get("exception", context["message"])
logger.error("Caught exception", message=message)
logger.error("MAIN: Shutting down")
asyncio.create_task(shutdown_handler(loop, exit_event))
async def shutdown_handler(
loop: asyncio.AbstractEventLoop, exit_event: asyncio.Event
) -> None:
logger.info("MAIN: Sending exit event")
exit_event.set()
await asyncio.sleep(5)
tasks = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()]
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
loop.stop()
def main() -> None:
loop, exit_event = asyncio.get_event_loop(), asyncio.Event()
loop.set_exception_handler(partial(exception_handler, exit_event=exit_event))
for s in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(
s,
lambda: asyncio.create_task(shutdown_handler(loop, exit_event)),
)
try:
loop.create_task(telegram_run(exit_event))
loop.create_task(discord_run(exit_event))
loop.create_task(web_run(exit_event, web_init(telegram.application)))
loop.run_forever()
finally:
loop.close()
if __name__ == "__main__":
main()
Just curious if this is the right way to run multiple long-running coroutines together. The blog post details how I reached this.
I would run each client in separate threads.
Here is a similar example to: discord.py · PyPI - Quickstart
import asyncio
import threading
import time
class DummyClient:
def __init__(self):
self.loop = asyncio.new_event_loop()
self.user = 'DummyUser'
async def on_ready(self):
print(f'We have logged in as {self.user}')
async def on_message(self, message):
if message.author == self.user:
return
if message.content.startswith('$hello'):
await self.send_message('Hello!')
async def send_message(self, content):
print(f'Message sent: {content}')
async def simulate_message(self, message):
await self.on_message(message)
async def print_periodically(self):
while True:
print('Print from DummyClient')
await asyncio.sleep(2.0) # Print every 2 seconds
def run(self):
asyncio.set_event_loop(self.loop)
self.loop.run_until_complete(self.on_ready())
# Simulate receiving messages
self.loop.create_task(self.simulate_message(DummyMessage('User1', '$hello')))
self.loop.create_task(self.simulate_message(DummyMessage('DummyUser', '$hello')))
# Start printing periodically
self.loop.create_task(self.print_periodically())
# Keep the loop running
self.loop.run_forever()
class DummyMessage:
def __init__(self, author, content):
self.author = author
self.content = content
def run_dummy_client():
client = DummyClient()
client.run()
# Create a thread to run the dummy client
client_thread = threading.Thread(target=run_dummy_client)
# Start the thread
client_thread.start()
# Now you can continue with other tasks in the main thread
print("Dummy async client is running in a separate thread.")
# This loop runs in the main thread
while True:
time.sleep(1.0)
print('Hi!')
Result:
Dummy async client is running in a separate thread.
We have logged in as DummyUser
Message sent: Hello!
Print from DummyClient
Hi!
Hi!
Print from DummyClient
Hi!
Hi!
...
so instead of keeping everything in the same thread in the same event-loop, better idea is to spread them into 3 different threads in production projects (where each thread has their own event loop)?
That would be straightforward and maintainable.