reflex
reflex copied to clipboard
add support for lifespan tasks
Add support for running task during the whole lifespan of the app, rather than linked to any specific state.
import asyncio
import reflex as rx
@rx.page()
def index():
return rx.box(), rx.color_mode.button(position="top-right")
async def long_running_task(key: str = "A"):
i = 1
while True:
await asyncio.sleep(5)
print(f"{key}: Oh I woke up {i} times.... Back to sleep again")
i += 1
async def longer_running_task():
while True:
await asyncio.sleep(10)
print("I sleep more than the others.")
@contextlib.asynccontextmanager
async def my_custom_context():
print("Startup process....")
yield
print("Shutdown process....")
app = rx.App()
app.register_lifespan_task(long_running_task)
app.register_lifespan_task(long_running_task, "B")
app.register_lifespan_task(longer_running_task)
app.register_lifespan_task(my_custom_context)
The tasks should have a chance to clean up nicely without being cancelled right?
I think it's up to the task itself to handle being cancelled cleanly, by using try/finally or catching the asyncio.CancelledError