Managing Asynchronous Work with Celery and Redis

Pedro Kauati — Fri, 10 Oct 2025 10:41:39 +0000

This is the sequel of my previous post, where I implemented a bug tracking web service with FastAPI. This time I'm gonna talk about the latest addition to the project: An async background job system and what it's being used for.

If you just want to see the completed project, feel free to give it a try, it's live! The source code can be found here.

Why Async?🧠

So after implementing the basic functionalities of the bug tracker, I wanted to get familiar with another important topic of back end development. What caught my attention was the idea of introducing concurrency. To put it simply, concurrency is the ability of a system to switch focus between different tasks. An operating system running on a single core CPU can achieve this, which is how you're able to run multiple programs even if they're not running in parallel.

And why is this useful for a web service? Trivial tasks may simply run one-per-one, without any hiccups. But add tasks with more processing or that depend on external services alongside several more client requests and your API will start to become slower, even for the trivial operations.

So concurrency allows our service to run non-trivial tasks on the background while it's still serving whatever else.

And what kind of task would that be? In my case, I decided to add a feature that project owners can request for a PDF report of their current projects.

Development💻

Stack update

Message Queue Manager: Celery🥕
MQ Data Store: Redis🟥
PDF Generator Library: FPDF📝
Blob Storage to host PDFs: MinIO️🗃️

Gotchas🌀

Celery was the beating heart of this operation. It took me a few tries to understand the system, specially since it involved running some code outside the FastAPI environment.

For starters, I got too used to FastAPI's hot reload, something that the worker container would not benefit from. So to ensure that I was working with the latest worker codebase I had to start running the containers with the --build flag.

I also had to struggle a bit with properly connecting the task methods to the celery app. What worked for me was calling the method autodiscover_tasks in the celery app object. This method takes a list of modules and allows the app to register tasks from files named tasks.py.

from celery import Celery
app = Celery(
    "celery_app", broker={REDIS_URL}, backend={REDIS_URL}
)
app.autodiscover_tasks(["app.worker"])

#tasks.py
from .celery_app import app
@app.task(bind=True,retry_backoff=True,...)
def generate_report(self, job_id: str, user_id: str):
    ...

Another challenging point was testing. I would say you currently have 3 ways to test a celery worker:

Use the pytest-celery library for asynchronous tests
Use celery.contrib.pytest also for asynchronous tests
Run synchronous tests by setting the flag task_always_eager to true

I found out to be a bit confusing to follow the documentation because I would often be unsure of which solution I was being reading about. In the end I opted for the synchronous tests, so the actual work done by the tasks could be asserted. This post goes into more detail on the trade off involved in each method.

When working with MinIO for data storage, I had to create two client connections, one private connection for communication between worker and storage, and a public one so the pdf files could be retrieved with a public facing url. Without that, the files could only be accessed via the docker network. An extra host had to be added to the minio container in order to make this work in dev.

Async Job System🕒

Here's the workflow for this system:

When a request for a non-trivial task happens, the API creates a new job, saves its state in the database and dispatches a new task to celery, with the proper info. Job state is saved as QUEUED
Celery stores the task in the Redis container.
If a worker is available, it picks up the task from the queue, changes the job state to RUNNING
Worker performs the task, which in this case is to generate a pdf and upload to minIO. The pdf url is saved in the database as an artifact.
If successful, the job is marked as SUCCEEDED and it now points to the artifact.
If it fails due to a recoverable error, such as network connection, the task retries after a few seconds. The exponential backoff adds time to each attempt.
If it fails due to an unrecoverable error, the job is marked as FAILED
The API can check the status of jobs via the endpoint /jobs/result. If a job is complete, its result can be accessed.

With this system, more workers can easily be added and future features can offload resource intensive work to the task queue without much extra setup.

Conclusion💡

Coming from Game Development, I was used to concurrency in another context, involving coroutines and game objects, and applying this concept to a backend application deepened my understanding of it.

As demand grows, web services require smart solutions to remain responsive and reliable. That only made me more curious to see how other systems and architectures shape performance and development efficiency.

As for Flyswatter, I’ll keep polishing the codebase and improving test coverage, but when it comes to new features, I’m ready to move on to another project and dive deeper into system design. This one taught me a lot, and I can’t wait to see what comes next!

From Idea to Deployment with FastAPI and Postgres

Pedro Kauati — Fri, 26 Sep 2025 09:52:41 +0000

Hello! My name is Pedro Kauati👋

In this post I'm going to describe my experience creating a webapp using FastAPI⚡

TL;DR: I created a back-end api that you can see it running here and check out the code here.

Motivation💭

My background is in Game Development. I have worked in several companies in Vancouver, mainly using C# and C++. I have dipped my toes in Web Development before, but I was never in charge of seeing something through from start to finish. So I decided to create my own web app (at least the back-end part). Since I'm familiar with Python from other projects, I decided to go with the FastAPI framework.

Idea💡

So I wanted to do something... doable. But still useful. Something that could have a real application, even if it only lives as a hobby project.

An Issue Tracker seemed like a good fit. And with that... Flyswatter was born (well, not yet)! A bug tracker where users could report issues from different projects, comment on them and assign issues to themselves or others.

Development🛠️

Stack

FastAPI⚡
PostgreSQL🐘
SQLAlchemy✨
Alembic🧪
Docker + docker-compose🐋
OAuth2 + JWT tokens for authentication🔐
pytest running on Github Actions🤖
Sentry for error monitoring📊

Gotchas🔍

During development I ran into some points of confusion for someone who's getting into a new development environment:

The difference between running scripts from docker containers and the docker host. Sometimes it makes no difference, but at other times it does. Something I still need to wrap my head around it
The correct way of managing the database while using Alembic. For example, you should ideally let Alembic generate the tables instead of making them in code. It also creates a table holding the current db revision, which you also shouldn't manually tamper with, unless you know what you're doing. Things like that can create a drift between the database and what Alembic tries to do, risking the migrations to not work at all.
Some common features of Python that I'm still not very well versed with, like the different ways to load env vars and for how long they last in memory; unpacking a dictionary to a set of kwargs and also Python's import system.
Watching out for which parts of the database aren't covered in Alembic's auto generation. One that I've noticed is that it can add enum types on upgrade but it doesn't remove them on downgrade.
Base.metadata.drop_all() nukes all tables except for alembic_versions and other remaining data. It does not go well with alembic inner workings, which may give false positives on migrations or fail them entirely. I lost a few hours trying to find the issue there.
In order to rollback database changes (useful for testing), you have to:
1. Start a db connection that lasts for as long as all the changes you want to rollback
2. Start a nested subtransaction
3. Listen to after_transaction_end, which triggers after session.commit()
4. When a transaction ends, start a new subtransaction
5. To rollback, call the rollback method on the original connection. This will revert all previous nested transactions

🐜Flyswatter🐜

After about 3 weeks of work, the main features of Flyswatter are complete!

You can check out the Github repo and the deployment on Render. I didn't put any work on the front-end, but fortunately FastAPI also makes use of Swagger UI, which generates a docs page with access to all the endpoints.

Conclusion📖

This was a very interesting project which taught me lots in terms of web frameworks and Python in itself. And I'm still not done with it, the next step now is to add background processing with Redis. I'm currently planning on generating a PDF bug report, but I may go with something else in the end.

Thank you for reading until this point and feel free to share your thoughts about the project!✌️

DEV Community: Pedro Kauati