How to repeatedly run a function in FastAPI, fastapi-utils repeat_every or asyncio.create_task?

Question

I want to repeatedly run a function in the background forever with FastAPI. I found two solutions on Internet but I don't know which one to choose.

The follow two code are both to update _STATUS value and show the value on the web-page.

In my opinion:

1. the decorator repeat_every from fastapi-utils is more readable (at least for someone, like me, who does not know much about asyncio. However fastapi-utils has no recent updates. Its latest version on PyPI was released on 2020-03-07.
2. On the other hand, asyncio is a built-in library in Python. But I don't quite understand the code below. If my aim is to update _STATUS every 3 seconds, is await asyncio.sleep(3) necessary or just a placeholder? If the _STATUS += 1 is replaced by some function that needs 2 seconds, shall I change it to await asyncio.sleep(1)? And what is the difference of the app_startup function between the 2 solutions? Thanks.

Overall, could you please tell me what the pros and cons of these 2 solutions? Which would one would you prefer? Thanks.

Based on fastapi-utils

from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every

app = FastAPI()

_STATUS: int = 0


@app.on_event('startup')
@repeat_every(seconds=3)
async def app_startup():
    global _STATUS
    _STATUS += 1


@app.get("/")
def root():
    return _STATUS

Based on asyncio.create_task (see insomnes comments)

import asyncio

from fastapi import FastAPI

app = FastAPI()

_STATUS: int = 0


async def run_main():
    global _STATUS
    while True:
        await asyncio.sleep(3)
        _STATUS += 1


@app.on_event('startup')
async def app_startup():
    asyncio.create_task(run_main())


@app.get("/")
def root():
    return _STATUS

And excuse for using _STATUS as a global variable which is a bad habit, here is just to make the code runnable.

Answer 1

If my aim is to update _STATUS every 3 seconds, is await asyncio.sleep(3) necessary or just a placeholder?

Without that sleep statement, your loop would look like this:

while True:
    _STATUS += 1

That will increment _STATUS as fast as possible. You need the sleep statement to delay the loop.

If the _STATUS += 1 is replaced by some function that needs 2 seconds, shall I change it to await asyncio.sleep(1)?

Sure, that's one way of handling it. Alternatively, you can rewrite your code so that the update happens asynchronously from the delay, like this:

async def do_update():
    global _STATUS
    _STATUS += 1

async def run_main():
    global _STATUS
    while True:
        asyncio.create_task(do_update())
        await asyncio.sleep(3)

@app.on_event('startup')
async def app_startup():
    asyncio.create_task(run_main())

This will call do_update every 3 seconds, regardless of how long it takes to execute. Of course, if do_update takes longer than three seconds to execute, you'll end up with two instances of this method running concurrently; there are several ways to mitigate that situation (e.g., a Lock or other synchronization primitives).

Overall, could you please tell me what the pros and cons of these 2 solutions? Which would one would you prefer? Thanks.

If you look at the code for the repeat_every decorator, you'll see that the two solutions are effectively identical.

Using a decorator is probably cleaner if you have to do the same thing at multiple points in your code, but you could just as easily implement that yourself.