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

[ollz272]

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.

[NanezX]

Same question here

[adaschevici]

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.

[yuval9313]

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)

[hitokiri]

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.