Permanently running background tasks

[tweakimp]

How can I add permanently running background tasks?

I want to have the following behaviour:

• Start the api and main starts to run as well.
• Open the browser and call the endpoint /.
• It returns the current value of value.

Example

import time

import uvicorn
from fastapi import FastAPI

app = FastAPI()


def main():
    # where and how should i start this function
    value = 0
    while True:
        time.sleep(0.1)
        value += 1


@app.get("/")
def root():
    # how can i read value from main?
    return value


if __name__ == "__main__":
    uvicorn.run("api:app", host="0.0.0.0", port=8001, reload=True)

[insomnes]

@tweakimp hello!
Maybe you can use on startup event functionality:
https://fastapi.tiangolo.com/advanced/events/

But don't add your main function directly, cause it will run forever.
Maybe you can do it with starting daemon thread of main() in some startup function.

If you consider using of asyncio (and understand async as general concept) you can try something like this:

class BackgroundRunner:
    def __init__(self):
        self.value = 0

    async def run_main(self):
        while True:
            await asyncio.sleep(0.1)
            self.value += 1

runner = BackgroundRunner()

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


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

[Raphencoder]

Is there a way to add a test to this code ?

[Trolldemorted]

will this run on one worker or on all workers if you start it with uvicorn?

[ahmedhr]

how do you run this using the new lifespan events method? Since that doesn't gather the tasks

[tweakimp]

Thank you for your fast answer, this works really well. I can even set the value from another end point:

@app.get("/set_value")
def set_value(x: int):
    runner.value = x
    return f"set runner.value to {x}"

[tweakimp]

This issue can be set to answered.

[Kludex]

The flag is attached when @tiangolo answers 😄

[tiangolo]

Thanks for the help here @insomnes and @Kludex ! 👏 🙇

Thanks for reporting back and closing the issue @tweakimp 👍

Hehe, I add the answered label manually, and then that is used 10 days after to automatically close the issue if there's not reply. 🤓 https://github.com/tiangolo/issue-manager

Sorry for the long delay! 🙈 I wanted to personally address each issue/PR and they piled up through time, but now I'm checking each one in order.

[OctaM]

I followed this example but instead I setup a Kafka consumer that was supposed to read messages from a topic. The problem is that I'm getting worker timeout. And the server just restarts.

[2022-12-23 15:11:37 +0000] [1] [CRITICAL] WORKER TIMEOUT (pid:13)
[2022-12-23 15:11:37 +0000] [1] [CRITICAL] WORKER TIMEOUT (pid:14)
[2022-12-23 15:11:37 +0000] [1] [WARNING] Worker with pid 14 was terminated due to signal 6
[2022-12-23 15:11:37 +0000] [157] [INFO] Booting worker with pid: 157
[2022-12-23 15:11:37 +0000] [1] [WARNING] Worker with pid 13 was terminated due to signal 9
[2022-12-23 15:11:37 +0000] [158] [INFO] Booting worker with pid: 158
[2022-12-23 15:11:37 +0000] [157] [INFO] Started server process [157]

class BackgroundKafkaConsumer:
    def __init__(self, bootstrap_server: str, topic_name: str, group_id=None):
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_server,
            'group.id': 'test4',
            'auto.offset.reset': 'earliest'
        })
        self.consumer.subscribe([topic_name])
    async def consume(self):
        while True:
            try:
                msgs = self.consumer.consume(num_messages=30, timeout=30)

                if len(msgs) == 0:
                    print("❌ No message(s) consumed (maybe we timed out waiting?)\n")
                else:
                    for msg in msgs:
                        print('✅  💌  Message received:  "{}" from topic {}\n'.format(msg.value().decode('utf-8'),
                                                                                     msg.topic()))
            except Exception as e:
                print("❌ Consumer error: {}\n".format(e))
                self.consumer.close()

And this is my start-up event

background_kafka = BackgroundKafkaConsumer(settings.KAFKA_BOOTSTRAP_SERVER, settings.KAFKA_TOPIC)
async def on_startup() -> None:
    logger.info('On startup')
    asyncio.create_task(background_kafka.consume())

At first it works and reads the messages from my topic, but then it just crashes the workers

[insomnes]

You run sync code inside async loop.
Your loop is blocked by created task. And worker is not started normally because of that

[kurkurzz]

Can you provide a workaround to solve this? I'm trying to run live opencv in the background but it seems to block the starting app entirely.

class BackgroundRunner:
	def __init__(self):
		self.frame = None

	async def run_video_inference(self):
		stream_url = 'sample.mp4'
		cap = cv2.VideoCapture(stream_url)
		while True:
			success, frame = cap.read()
			# run object detection here
			# ...
			# update the latest frame in the shared variable
			self.frame = frame

[AdamGold]

@OctaM Have you figured out a way to solve this, and run your Kafka consumer with a background task?

[bingblackbean]

thanks for the answers. it works well in development env.
if we try to switch to production env, I mean, if we use multiple uvicorn workers (e.g 4), then this Runner will initialize 4 times. It will waste the computer resources as these 4 runner are doing same things.
is it possible to make Runner run only once or make these 4 Runners running in a sequence?
thanks

[afaheem88]

thanks for the answers. it works well in development env. if we try to switch to production env, I mean, if we use multiple uvicorn workers (e.g 4), then this Runner will initialize 4 times. It will waste the computer resources as these 4 runner are doing same things. is it possible to make Runner run only once or make these 4 Runners running in a sequence? thanks

@bingblackbean Try using some non-blocking lock such as linux lock(fcntl.flock) if your app is running on a single server or use redis SETNX(to have a single worker across all servers) to tell other background workers to exit. something like-

async def consume(self):
    if lock.acquire():
        while True:
            try:
                msgs = self.consumer.consume(num_messages=30, timeout=30)

                if len(msgs) == 0:
                    print("❌ No message(s) consumed (maybe we timed out waiting?)\n")
                else:
                    for msg in msgs:
                        print('✅  💌  Message received:  "{}" from topic {}\n'.format(msg.value().decode('utf-8'),
                                                                                     msg.topic()))
            except Exception as e:
                print("❌ Consumer error: {}\n".format(e))
                self.consumer.close()

[bobogogo1990]

you can use pm2 to run fastapi as a background service.
pm2 start serve.py --interpreter python. where serve.py run the fastapi app: uvicorn.run(app, host="localhost", port=8000)

[amacfie]

You might find #10743 helpful