Simple task repeated every N minutes - is there a solution for litestar?

[bunny-therapist]

We are using litestar apps in production, each being run by NGINX Unit and packaged inside a docker file, then run inside of a Kubernetes cluster.

As a new requirement, we now want some of the apps to be able to fetch configuration data from an external source (url) at regular intervals, for example, every 5 minutes, or 10 minutes, or 60 minutes or so (not every few seconds - it will never be that frequent), and of course also at start up, to actually have some config to work with.

I found this discussion: https://github.com/orgs/litestar-org/discussions/2311
It is 2 years old and I don't know if that is still the state of things. It referred me to this repo: https://github.com/cofin/litestar-saq
but the README there appears broken (and it is still on zero version). It also appears that I would need to have Redis (or Valkey? Maybe? Not sure.) running as well to get it to work (this is not documented, I just inferred this from the description and comparisons to RQ). So now I am not sure how to do this. Run a Redis instance inside each pod just to be able to have this job running? I ideally do not even want this to run in a separate process - I would have been fine with just running it async in the main thread, or spawn a thread - it just needs to fetch some data from an endpoint once in a while, validate that, then update the app config with it. All this workers and databases etc etc seems so incredibly heavyweight.

What is the recommended solution here? I am incredibly happy with litestar so far, but I feel like I am encountering the first real hurdle.

I see asyncio has some call_later method (https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.lo) - can that work?

[bunny-therapist]

This seems to work?

from litestar import Litestar
import asyncio


def on_clock() -> None:
    loop = asyncio.get_running_loop()
    print("Tick")
    loop.call_later(5, on_clock)


app = Litestar(on_startup=[on_clock])

It prints "Tick" every 5 seconds while the app is running. I am sure there are some problems with it. For example, I don't know how to make it async. loop.call_later crashes if I pass it something async.

[bunny-therapist]

Ok this way it works with a coroutine:

from litestar import Litestar
import asyncio


async def on_clock() -> None:
    loop = asyncio.get_running_loop()
    print("Tick")
    loop.call_later(1, asyncio.create_task, on_clock())


app = Litestar(on_startup=[on_clock])

I guess can also directly schedule coroutines and add asyncio.sleep in the beginning of them.

Is there any reason not to do this?

[cclinet]

Think about using APScheduler

https://apscheduler.readthedocs.io/en/3.x/userguide.html

[bunny-therapist]

Thanks. It seems like it is doing exactly the same thing I was, but it shows how to cancel the pending task properly. I will either use this or get inspired by it (depending on whether I think the extra dependency is worth it).