FastApi background tasks — but better.

Say you’re running a FastApi service and you want to trigger a long running task that you don’t want to wait for, a common use case for it is analytics, you just want to fire and forget it.

One way to solve it is to use FastApi background tasks, but those are very limiting they force you to start a background task at the router level, which means your whole flow is going to run as a background task, that won’t work.

Now depends on what you’re trying to run and your use case there a couple more options

1. For synchronous IO blocking — schedule it on a thread with asyncio to_thread or use a library like greenletio which can do it for you but on a thread.
2. For async calls — you can either send the coroutine to your existing FastApi loop or start another thread with another loop and submit your coroutine there.

1. Synchronous blocking calls

The use case here is when running an async app and you have a IO blocking calls or a CPU heavy call and you want to run it in a thread/greenlet so it won’t block the main thread from accepting calls

Using to_thread

A simple use case of that will be:

import time
import asyncio 

def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.to_thread(blocking_io)
    print(f"finished main at {time.strftime('%X')}")

asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

Using greenletio, this library will use greenlets to bridge async and sync calls.

import asyncio
from greenletio import async_
import time

@async_
def sync_function():
    print(f"start blocking_io at {time.strftime('%X')}")
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")
    await sync_function()
    print(f"finished main at {time.strftime('%X')}")

asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

2. Async calls

This is a slightly easier situation to manage because there is no need for additional threads (necessarily) which could add complexity.

Submit to existing loop

Good for simple use cases where we want to run a light task on the current loop, will not be good for CPU heavy tasks as they will consume a lot of time of the loop and might hurt performance, but very simple to manage. good for sending api calls for analytics.

import time
import asyncio


async def async_func():
    print(f"start async_func at {time.strftime('%X')}")
    await asyncio.sleep(1)
    print(f"async_func complete at {time.strftime('%X')}")


async def main():
    print(f"started main at {time.strftime('%X')}")
    asyncio.create_task(async_func())
    print(f"finished main at {time.strftime('%X')}")
    await asyncio.sleep(2) # just adds time for the loop to wait for the async response

asyncio.run(main())
# results: 
# started main at 09:30:17
# finished main at 09:30:17
# start async_func at 09:30:17
# async_func complete at 09:30:18
# you can see the the main started and finished before async

This is a simple solution, but be careful with creating Tasks if you are running multiple threads since tasks in python are not thread safe.

Create a new thread with a loop

A Good use cases for this one is where you are seeing a negative performance impact from the previous option where the task is running on the loop in the main thread.

import time
import asyncio
from threading import Thread


async def async_func():
    print(f"start async_func at {time.strftime('%X')}")
    await asyncio.sleep(1)
    print(f"async_func complete at {time.strftime('%X')}")


async def main():
    print(f"started main at {time.strftime('%X')}")
    # starts a new loop in a thread
    loop = asyncio.new_event_loop()  # assuming a loop is running if using fastapi
    thread = Thread(target=loop.run_forever)
    thread.start()
    asyncio.run_coroutine_threadsafe(async_func(), loop) # scheudle coro in loop
    print(f"finished main at {time.strftime('%X')}")
    await asyncio.sleep(2)

    # stop loop and thread
    for task in asyncio.all_tasks(loop):
        if not task.done():
            task.cancel()
    loop.call_soon_threadsafe(loop.stop)

asyncio.run(main())
# results: 
# started main at 09:30:17
# finished main at 09:30:17
# start async_func at 09:30:17
# async_func complete at 09:30:18
# you can see the the main started and finished before async

By running tasks in another thread you step in the world of parallelism that presentness a new set of challenges.

Challenges of running async tasks in another thread (and loop):

• Race conditions — A race condition occurs when two threads access a shared variable at the same time. The first thread reads the variable, and the second thread reads the same value from the variable. Can be tricky with static objects and singletons (like api/db clients).
• Asyncio loop sync — You cannot run a tasks that starts in one loop and ends in another.

Hope you enjoyed!