Step 8 - Extend the Loop¶
As a next step we want to extend the loop so that we can get the current loop.
Loop v2¶
from typing import Any, Generator
class Loop:
"""Loop v2"""
_instance: "Loop" = None
def __init__(self):
self._running = False
@classmethod
def get_current_loop(cls) -> "Loop":
if not cls._instance:
cls._instance = Loop()
return cls._instance
def run(self, coroutine: Generator[Any, None, Any]) -> Any:
"""Run a coroutine"""
self._running = True
step = 1
while self._running:
print("Loop step", step)
try:
next(coroutine)
step += 1
except StopIteration as e:
self._running = False
return e.value
def stop(self) -> None:
"""Stop running the loop"""
self._running = False
The loop has been converted into a class that uses a singleton pattern to return
the current loop. The current loop is now accessible via Loop.get_current_loop()
.
A coroutine can be started by calling the run
method on the loop instance. The
state of the loop (running
/not running
) is tracked in the private _running
property.
from future import Future
from loop import Loop
def some_result(value):
future = Future()
future.set_result(value)
return (yield from future)
def add(coroutine1, coroutine2):
x = yield from coroutine1
y = yield from coroutine2
return x + y
def main():
return (yield from add(some_result(1), some_result(2)))
loop = Loop.get_current_loop()
result = loop.run(main())
print("Loop finished with result", result)
Output:
Loop step 1
Loop step 2
Loop step 3
Loop finished with result 3
Summary
The loop will not be instantiated by the application code directly.
Instead the current loop is requested at a single API point.
The loop could decide whether it returns a new instance or not.