[Python] decorator that will keep the main loop alive

In the following implementation, this creates a decorator that wraps a function with an async call. This calls the function using a periodic function, so the user can decorate a function and have it be called as part of a loop without needing to explicitly handle it. If I do not add the injector.run_forever() call at the end though, the main loop ends and the httpx.AsyncClient fails with the error:

RuntimeError: cannot schedule new futures after shutdown


Is there a way to keep the main process alive without having an explicit blocking call? I'd like to make it so I don't need to call injector.run_forever(), but can put something inside of the decorator that will keep the process alive.

import asyncio
import signal
import threading
import time
from functools import wraps

import httpx


async def long_printer(func, stop_event):
while not stop_event.is_set():
try:
async with httpx.AsyncClient() as client:
response = await client.get(
"https://random-word-api.herokuapp.com/word"
)
word = response.json()[0]
except Exception:
import traceback

traceback.print_exc()
await asyncio.sleep(3)
continue
func(word)
await asyncio.sleep(3)


class Injector:
stop_event: threading.Event = threading.Event()
loop: asyncio.AbstractEventLoop | None = None
_thread: threading.Thread | None = None

def __init__(self):
signal.signal(signal.SIGINT, self.stop)
signal.signal(signal.SIGTERM, self.stop)

def add_thread(self, func):
if self.loop is None:
self.loop = asyncio.new_event_loop()
# Pre-warm the thread pool so that atexit registration happens now
self.loop.run_in_executor(None, lambda: None)
self._thread = threading.Thread(target=self._run_event_loop, daemon=False)
self._thread.start()
asyncio.set_event_loop(self.loop)

# Create task in the loop
@wraps(func)
async def create_and_start_task():
await long_printer(func, self.stop_event)

asyncio.run_coroutine_threadsafe(create_and_start_task(), self.loop)

def run_forever(self):
while not self.stop_event.is_set():
time.sleep(3)

def _run_event_loop(self):
asyncio.set_event_loop(self.loop)
try:
self.loop.run_forever()
finally:
self.loop.close()

def stop(self, *args, timeout: int = 5):
self.stop_event.set()

if self.loop is not None:
self.loop.stop()
if self._thread:
self._thread.join(timeout)

self.loop = None
self._thread = None

def decorator(self):
def wrapper(func):
self.add_thread(func)
return func

return wrapper


injector = Injector()


@injector.decorator()
def print_text(text: str):
print("Received:", text)


@injector.decorator()
def print_text2(text: str):
print("Received2:", text)


injector.run_forever()

Continue reading...