Background Tasks: From FastAPI to Hatchet

Since you’re here, you might be interested in checking out Hatchet — the platform for running background tasks, data pipelines and AI agents at scale.

FastAPI, Python’s new favorite web framework, offers a lightweight, easy-to-use background tasks feature for triggering async tasks that can run later without blocking an incoming request from completing. Of course, this is a killer feature: Eventually every app needs a way to run background tasks.

FastAPI Background Tasks 101

Under the hood, FastAPI background tasks just wrap the Starlette implementation, which simply awaits the background task after sending the response to the client.

This is a handy trick. First, we send the response, and then we run the background task in a non-blocking way afterwards. As FastAPI advertises, this lets you do things like sending emails, processing data, etc. that the client does not need to wait for.

Getting to Production

As you start getting ready to move your FastAPI application to a production setting, you might notice some issues with how FastAPI handles background tasks. In no particular order, a handful of the most immediate issues are:

1. There’s very little observability here. Since the task is just awaited when the response is sent, it’s difficult to know if the background tasks are actually completing or if they’re being dropped.
2. If the server shuts down, you might have background tasks still waiting to be run. If they don’t complete before the application is terminated, those tasks will be dropped, leading to data loss.
3. There’s no way to handle common task queueing issues like concurrency, retries, and so on.
4. All of this work is being run on your web server, even though it’s done in a non-blocking way. This means that background tasks still will eat up CPU and memory on your server, which can be hard to debug, and can lead to performance issues for clients.

These issues, in addition to others that are likely to come up, are the reason you might migrate your FastAPI background tasks over to a more robust tool like Hatchet when you’re getting ready to ship your app to production.

Migrating to Hatchet

Hatchet’s functionality is built to solve exactly these problems, and many more that you’ll face as you continue to scale and overcome new obstacles! For the issues above, Hatchet solves them by:

1. Having a fully featured dashboard showing task statuses, runtimes, etc. and by providing additional tools like an OpenTelemetry integration to help you monitor your application.
2. Hatchet will “reassign” tasks that do not run to completion to a new (running) worker, so you don’t need to worry about your worker shutting down and a task being dropped. Workers can safely shut down without data loss.
3. Hatchet offers lots of concurrency, rate limiting, retrying, and many more features to help you build background tasks that adhere to your business logic.
4. You can scale your Hatchet workers independently of your web server, and can even utilize Hatchet’s managed compute for autoscaling as your queue grows.

Porting your tasks from FastAPI background tasks to Hatchet is simple - all you need to do is create Hatchet tasks out of the functions you’re passing to add_task. For instance:

async def send_welcome_email_task(user_id: int) -> None:
    async with Session() as db:
        user = await get_user(db, user_id)
 
        await send_welcome_email(user.email)
 
@app.post("/user")
async def create_user(background_tasks: BackgroundTasks) -> User:
    async with Session() as db:
        user = await create_user(db)
 
        background_tasks.add_task(send_welcome_email_task, user.id)
 
        return user

Would become:

class WelcomeEmailInput(BaseModel):
    user_id: int
 
@hatchet.task(input_validator=WelcomeEmailInput)
async def send_welcome_email_task(input: WelcomeEmailInput, _ctx: Context) -> None:
    async with Session() as db:
        user = await get_user(db, input.user_id)
 
        await send_welcome_email(user.email)
 
@app.post("/user")
async def create_user() -> User:
    async with Session() as db:
        user = await create_user(db)
 
        await send_welcome_email_task.aio_run_no_wait(
            WelcomeEmailInput(user_id=user.id)
        )
 
        return user

And that’s it! When you trigger the Hatchet task (in this case, in “fire and forget” style), your task will be sent through the Hatchet Engine to your worker, where it will execute, and report its result in the dashboard for you to see. Or if something goes wrong, you can be notified.

Feature Comparison

Ready to Migrate?

Check out our blog post on Hatchet using modern Python for a thorough introduction to Hatchet.

You can get up and running in just five minutes on Hatchet Cloud. And if you’d like to learn more, you can find us:

• On GitHub
• On Discord

Or check out our documentation.