Repository navigation
[QUESTION] Is it possible to use repeated tasks with lifetimes? if so how? #357
Description
Activity
Same question here
def fake_answer_to_everything_ml_model(x: float): return x * 42 @repeat_every(seconds=5) def logstuff(): logger.warning("Periodic execution") ml_models = {} @asynccontextmanager async def lifespan(app: FastAPI): # Load the ML model ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model await logstuff() # -> this run periodically yield # Clean up the ML models and release the resources ml_models.clear()
This seems to work. Args won't get propagated since it doesn't really make sense in a periodic task i think.
Reacted by GabrielOh yeah, it's just the docs fault, ill fix it soon, essentially periodic task is a very simple (maybe unnecessary due to other more robust libs) decorator for function to call themselves using
asyncio.wait_for(if I remember correctly) with the introduction of lifetime you could schedule it yourself in more simple manner with apscheduler (for example)I think I can keep the original method for simplicity but there are more robust solutions that support more cases and have more docs online (apscheduler, again)
Hello, my solution is using
@repeat_every(seconds=5)` def logstuff(): logger.warning("Periodic execution") app.add_event_handler("startup", logstuff)
so do not use an async or lifespan.
Description
How can I use repeated tasks with lifetimes?
The docs only give examples how how to use these utils with the app.on_event(...) api, but fastapi now recommends using lifetimes. Its unclear to me whether the repeated tasks are supported here, and if they are how they would be used.