[QUESTION] Is it possible to use repeated tasks with lifetimes? if so how? #357

Activity

1. ollz272

   added

   questionFurther information is requestedFurther information is requested

   on Jan 16, 2025
2. NanezX commented on Jan 23, 2025

   

   NanezX

   on Jan 23, 2025

   More actions

   Same question here
3. adaschevici commented on Jan 24, 2025

   

   adaschevici

   on Jan 24, 2025

   More actions

   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.

   ❤️React with ❤️1Reacted by Gabriel
4. yuval9313 commented on Feb 19, 2025

   

   yuval9313

   on Feb 19, 2025

   Collaborator

   More actions

   Oh 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)
5. hitokiri commented on Jun 17, 2025

   

   hitokiri

   on Jun 17, 2025

   More actions

   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.