FastAPI has quickly become one of the most popular frameworks in the Python ecosystem because of its modern design, async-first approach, and developer-friendly features such as automatic validation and interactive API documentation. But building real-world applications involves more than just creating endpoints—it requires designing software that is scalable, maintainable, and testable.
Two architectural ideas that support this goal are dependency injection and event-driven design. Dependency injection makes it possible to separate concerns by removing hardcoded dependencies and allowing them to be provided from the outside. This approach improves flexibility and makes testing significantly easier. Event-driven architecture, on the other hand, encourages components to communicate through events instead of direct calls, leading to systems that are loosely coupled and easier to scale.
When combined, these two patterns offer a powerful foundation for building robust applications with FastAPI. In this post, we will explore how to take advantage of FastAPI’s built-in dependency injection system and how to integrate it with a lightweight event bus to create APIs that are clean, testable, and ready for production.
Vlog
What is Event Driven Design?
Have you ever wondered how Starbucks manages its queues so efficiently? You walk in, and the first barista takes your name and order, then writes it on a cup. That’s their role—quick and focused. The cup is then passed to another barista, whose job is to prepare your coffee. Meanwhile, someone else may be restocking coffee beans to ensure supplies never run out.
Writing your name on a paper cup is quick and effortless, much like the producing phase of event-driven architecture. Brewing the coffee, on the other hand, takes more time, similar to the consuming and processing phase. And just like the resumable infrastructure of event-driven systems, even if the baristas change shifts, the next barista can simply pick up the cups waiting on the shelf and continue brewing without any disruption.
What is Dependency Injection?
Imagine you walk into Starbucks again. When you order, you don’t bring your own coffee machine, beans, milk, and cups. The barista already has everything they need provided for them. Their job is just to prepare your coffee — they don’t need to know where the beans came from, who supplied the milk, or how the machine works.
That’s exactly what dependency injection is in software. Instead of having your code create and manage all its dependencies by itself (like the barista growing their own coffee beans before making your latte), those dependencies are provided to it from the outside. The function, class, or module just focuses on its own responsibility.
In FastAPI, for example, if an endpoint needs access to a database, you don’t make it connect directly inside the function. Instead, you “inject” the database connection as a dependency. This keeps your code clean, testable, and easy to maintain — just like a barista can easily switch from one brand of beans to another without having to change how they make coffee.
Why Event Driven Architecture is So Important?
In production, we generally try to avoid long-running for loops. Python behaves differently from JavaScript in this regard: JavaScript can process forEach loops in parallel, while Python runs them serially by default. You can use multiprocessing to parallelize Python tasks, but that approach is limited by the number of cores on a single machine.
Event-driven architecture (EDA) allows us to scale beyond a single machine. Instead of processing everything sequentially, we can distribute work across multiple servers and take full advantage of many cores on each machine. A common guideline is to run 2–4 worker processes per core. For example, the machine I’m writing this on has 6 physical cores (12 virtual cores), which means I could run 24–48 workers for experimentation.
If you need even more processing power, you can simply add more servers to your architecture. With modern cloud infrastructure, scaling up is often just a matter of changing configuration. In short, adopting EDA removes the bottleneck of single-threaded execution and makes scalability a much easier problem to solve.
Use Case
Similar to the experiment we did with Flask and Kafka, we will implement the same use case: analyzing facial attributes, such as age and gender, for multiple faces in a single photo. The source code for this experiment has already been pushed to GitHub.
Variables and Modules
The purpose of dependency injection in this experiment serves two main goals: loading environment variables just once and loading modules according to the principle of separation of concerns, also only once.
I prefer to store environment variables in a .env file. Since there’s no sensitive information in this project, I pushed the .env file to Git. However, if your environment variables contain sensitive data, make sure to add .env to your .gitignore file and add only keys into .env.example.
# .env EDA_ACTIVATED=1 DETECTOR_BACKEND=mtcnn KAFKA_URI=localhost:9093 TF_CPP_MIN_LOG_LEVEL=2
Then, I load them during application initialization using the load_dotenv function from the python-dotenv package.
# src/app.py from dotenv import load_dotenv load_dotenv()
I have a Variables class that loads environment variables and assigns them to attributes, setting default values when a variable does not exist. In this class, I also define some constants, such as topic names, for easier reference throughout the application.
# src/dependencies/variables.py
# built-in dependencies
import os
# pylint: disable=too-few-public-methods
class Variables:
def __init__(self):
self.kafka_uri = os.getenv("KAFKA_URI", "CHANGEME")
self.detector_backend = os.getenv("DETECTOR_BACKEND", "opencv")
self.is_eda_activated = os.getenv("EDA_ACTIVATED", "0") == "1"
self.topics = ["faces.extracted"]
Next, I’ll create a singleton Container class. This class will initialize application modules according to the principle of separation of concerns and provide them with the necessary variables.
# src/dependencies/container.py
from dependencies.variables import Variables
class Container:
_instance = None
_initialized = False
def __init__(self, variables: Variables):
self.logger = Logger()
self.deepface_service = DeepFaceService(
logger=self.logger,
detector_backend=variables.detector_backend,
)
self.event_service = KafkaService(
logger=self.logger,
server_uri=variables.kafka_uri,
)
for topic_name in variables.topics:
self.event_service.create_topic_if_not_exists(
topic_name=topic_name
)
self.core_service = CoreService(
logger=self.logger,
deepface_service=self.deepface_service,
event_service=self.event_service,
is_eda_activated=variables.is_eda_activated,
)
self._initialized = True
self.logger.info("Container initialized")
def __new__(cls, variables=None)
if cls._instance = None:
cls._instance = super().__new__(cls)
return cls._instance
Dependency Injection – Traditional Way
We could create the container during application startup and attach it to the router.
# src/app.py
def create_app() -> FastAPI:
app = FastAPI()
# enable CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
variables = Variables()
container = Container(variables=variables)
# dependency injection to router directly
router.container = container
app.include_router(router)
container.logger.info("router registered")
# startup event for FastStream
@app.on_event("startup")
async def start_faststream():
# FastStream is blocking; we are staring with background task
asyncio.create_task(faststream_app.run())
container.logger.info("FastStream broker started")
return app
Then, we will access the container from the router within the routes, whether we are exposing web service endpoints or consuming Kafka topics.
# src/modules/core/routes.py
# 3rd party dependencies
from fastapi import APIRouter
from pydantic import BaseModel
# projet dependencies
from modules.core.bus import broker
router = APIRouter()
class AnalyzeRequest(BaseModel):
image: str
@router.post("/analyze")
async def analyze(
payload: AnalyzeRequest,
background_tasks: BackgroundTasks,
):
container: Container = router.container # directly attachted to router
class AnalyzeExtractedFaceRequest(BaseModel):
face_id: str
request_id: str
face_index: int
encoded_face: str
shape: tuple
@broker.subscriber("faces.extracted")
async def analyze_extracted_face_kafka(
input_value: AnalyzeExtractedFaceRequest,
):
container: Container = router.container # directly attachted to router
This approach allows the router to access all required services and objects via the container. In other words, any service your endpoints need is immediately available through router.container. But this approach comes with some advantages and disadvantages.
Pros
- Simplicity: It’s straightforward and works well in small projects.
- Centralized access: All services are stored in one container, so you don’t have to pass them individually.
Cons / Limitations
- Not the standard FastAPI DI: FastAPI’s recommended dependency injection system is based on Depends(), which integrates with the request lifecycle. Directly attaching a container bypasses this system.
- No request-scoped management: Services are not automatically tied to a request lifecycle. You have to manage instantiation, cleanup, and thread-safety manually.
- Testing and mocking are harder: Since dependencies are not injected per-request, replacing services with mocks requires extra boilerplate.
- Reduced type safety: You lose FastAPI’s type-based dependency resolution, which can help catch errors at design time.
- Less scalable: This pattern can work for small projects but becomes harder to maintain in large, complex applications.
In short: Assigning a container directly to a router is convenient, but it trades off the safety, flexibility, and scalability offered by FastAPI’s built-in Depends system. For production or larger applications, it’s generally recommended to use Depends for dependency injection.
Dependency Injection – FastAPI’s Way
FastAPI provides a built-in dependency injection system using Depends(), which allows you to define dependencies at the request level.
Pros
- Request-scoped dependencies: Each request can get its own instance of a dependency, which is important for things like database sessions or user-specific services.
- Type safety: FastAPI uses Python type hints to validate and resolve dependencies, reducing runtime errors.
- Automatic lifecycle management: FastAPI can handle initialization and cleanup of dependencies, including async context management.
- Easier testing and mocking: Dependencies can be overridden for testing using dependency_overrides, making it easy to inject mocks or stubs.
- Better scalability: This approach works well in large, complex applications because dependencies are explicit and managed consistently.
Cons / Limitations
- Slightly more boilerplate: You need to define dependency functions or classes and use Depends in each endpoint.
- Indirect access: Compared to attaching a container directly, you cannot access all services globally; everything must go through dependency injection.
- Request-scoped by default: Dependencies are resolved per request. This means if you use a container or a heavy service as a dependency, it will be recreated for each request. If you want it to behave like a singleton, you have to manage that explicitly.
In short: Depends provides a structured, scalable, and testable way to manage dependencies in FastAPI. While it requires a bit more setup than directly attaching a container, it improves safety, maintainability, and alignment with FastAPI best practices.
Dependency Injection
Previously, we were creating the container in src/app.py. In this approach, however, we won’t initialize the container there; instead, we’ll handle everything in src/modules/core/routes.py.
When defining the function for the HTTP POST endpoint /analyze, we will include the container as an input argument using FastAPI’s Depends. The container initialization logic is encapsulated in a get_container function, which is then passed to Depends to provide the container to the endpoint.
Similarly, we will stream the faces.extracted Kafka topic, and whenever a message arrives, the analyze_extracted_face_kafka function will be triggered. The container will be passed as an input argument using FastStream’s Depends. The container initialization logic is encapsulated in the get_container function, which is then provided to Depends to make the container available to the streaming process.
# src/modules/core/routes.py
# 3rd party dependencies
from fastapi import APIRouter, Depends
from faststream import Depends as FastStreamDepends
# project dependencies
from dependencies.variables import Variables
from dependencies.container import Container
from modules.core.bus import broker
router = APIRouter()
def get_container() -> Container:
variables = Variables()
return Container(variables=variables)
class AnalyzeRequest(BaseModel):
image: str
@router.post("/analyze")
async def analyze(
payload: AnalyzeRequest,
background_tasks: BackgroundTasks,
container: Container = Depends(get_container), # via FastAPI Depends
):
container.logger.info("POST /analyze endpoint is called")
class AnalyzeExtractedFaceRequest(BaseModel):
face_id: str
request_id: str
face_index: int
encoded_face: str
shape: tuple
@broker.subscriber("faces.extracted")
async def analyze_extracted_face_kafka(
input_value: AnalyzeExtractedFaceRequest,
container: Container = FastStreamDepends(get_container), # via FastAPI Depends
):
container.logger.info("STREAM faces.extracted triggered")
Event Driven Architecture with FastAPI
While setting up the infrastructure for dependency injection, we also built the structure to listen to the Kafka topic. In other words, the consumer part of the event-driven architecture was partially designed as well.
Thanks to FastStream, the analyze_extracted_face_kafka function is triggered automatically whenever a message is published to the faces.extracted Kafka topic, just like a web service endpoint. Without FastStream, using the standard Kafka Python package would require running a long-lived application, which is much harder to debug and test.
However, in its current state, it won’t actually listen to the topic because the broker we imported at the routes level is not yet fully initialized.
# src/modules/core/bus.py # 3rd party dependencies from faststream import FastStream from faststream.kafka import KafkaBroker # project dependencies from dependencies.container import Variables variables = Variables() broker = KafkaBroker(variables.kafka_uri) faststream_app = FastStream(broker)
Later, when we start our FastAPI application, we will also need to start the FastStream application.
# src/app.py
def create_app() -> FastAPI:
app = FastAPI()
# enable CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# dependency injection to router directly
# variables = Variables()
# container = Container(variables=variables)
# router.container = container
app.include_router(router)
print("router registered")
# startup event for FastStream
@app.on_event("startup")
async def start_faststream():
# FastStream is blocking; we are staring with background task
asyncio.create_task(faststream_app.run())
print("FastStream broker started")
return app
service = create_app()
Now, our application is ready to stream incoming messages to the Kafka topics.
Finally, we can start our application using Uvicorn. Here, app comes from src/app.py, and service is obtained from the corresponding variable defined there.
uvicorn app:service --host 0.0.0.0 --port 5000 --workers 2
Producing Messages
In our core service, the event service is already referenced. We simply need to call its produce method to send messages to a Kafka topic.
# src/modules/core/service.py
def analyze(self, image: str, request_id: str):
faces = self.deepface_service.extract_faces(image)
self.logger.info(f"extracted {len(faces)} faces")
for idx, face in enumerate(faces):
encoded_face = base64.b64encode(face.tobytes()).decode("utf-8")
self.event_service.produce(
topic_name="faces.extracted",
key="extracted_face",
value={
"face_id": uuid.uuid4().hex,
"face_index": idx,
"encoded_face": encoded_face,
"request_id": request_id or "N/A",
"shape": face.shape,
},
)
self.logger.info(
f"{idx+1}-th face sent to kafka topic faces.extracted"
)
FastAPI vs Flask
There’s a long-standing debate in the Python community about which is better: Flask or FastAPI. To be honest, both are excellent frameworks, but FastAPI has become the more popular choice in recent years. That said, if you’re a Flask fan, you can check out this post to see the Flask equivalent: Dependency Injection and Event-Driven Architecture with Flask and Kafka.
Conclusion
Designing applications with FastAPI goes beyond defining endpoints—it’s about building systems that are resilient, testable, and ready to grow. By embracing dependency injection, we gain flexibility and the ability to swap or mock components without rewriting business logic. By applying event-driven principles, we decouple responsibilities and make it easier for services and features to evolve independently.
When used together, these patterns create a foundation where FastAPI applications remain clean and maintainable while also being prepared for the demands of production environments. Whether you start with a simple in-memory event bus or integrate with a distributed message broker like Kafka or RabbitMQ, the combination of dependency injection and event-driven architecture can significantly improve the quality and scalability of your projects.
The key takeaway is that good architecture doesn’t just solve today’s problems—it ensures the codebase is ready for tomorrow’s challenges. With FastAPI, dependency injection, and event-driven design working hand in hand, you’ll be well equipped to build applications that stand the test of time.
I pushed the source code of this study into GitHub. You can support this work by starring the repo.
Support this blog financially if you do like!
