Creating resolved Future without an event loop

Python Help
1

I have some old code in a __ init __ file that does:

LOGGING_DISABLED: LOGGING_STATUS = asyncio.Future()
LOGGING_DISABLED.set_result("DISABLED")
LOGGING_IGNORED: LOGGING_STATUS = asyncio.Future()
LOGGING_IGNORED.set_result("IGNORED")
LOGGING_RUNNING: LOGGING_STATUS = asyncio.Future()
LOGGING_RUNNING.set_result("RUNNING")

never python complains
DeprecationWarning: There is no current event loop

How do I make a resolved Future without an event loop.
I really do not want to make an event loop just to close it again 6 lines later

2

That doesn’t look like valid Python code to me, and it doesn’t mention Futures at all. Maybe it’s logging output? Could you quote the actual code instead?

I assume “never” → “newer”.

3

It is valid code. I assume that LOGGING_STATUS is some static type:

In [39]: import asyncio

In [40]: LOGGING_STATUS = int

In [41]: LOGGING_DISABLED: LOGGING_STATUS = asyncio.Future()
    ...: LOGGING_DISABLED.set_result("DISABLED")
    ...: LOGGING_IGNORED: LOGGING_STATUS = asyncio.Future()
    ...: LOGGING_IGNORED.set_result("IGNORED")
    ...: LOGGING_RUNNING: LOGGING_STATUS = asyncio.Future()
    ...: LOGGING_RUNNING.set_result("RUNNING")
<ipython-input-41-b566686b297e>:1: DeprecationWarning: There is no current event loop
  LOGGING_DISABLED: LOGGING_STATUS = asyncio.Future()
4

Ah, I didn’t realize the : LOGGING_STATUS parts were type annotations. :frowning:

Anyways, @pebl: Why do you want variables that are resolved futures, disconnected from any loop? How are these used? That DeprecationWarning isn’t going to be removed and will only get worse in the future, so the best thing to do is figure out what that code was trying to do and find a different way to express it.

5

instead of a Future (which you should basically never use directly - it’s only for low level code to translate between callbacks/protocols and tasks) use an Event wrapped with an awaitable class:

class MyAwaitable(Generic[T]):
    def __init__(self):
        self._result: T | None = None
        self._event = asyncio.Event()

    async def wait(self) -> T:
        await self._event.wait()
        return cast(T, self._result)

    def set_result(self, result: T) -> None:
         if self._event.is_set():
             raise RuntimeError("Invalid state")

         self._result = result
         self._event.set()

    def __await__(self) -> Generator[T]:
        return self.wait().__await__()