FastAPI Mircoservice Patterns: What’s missing to make FastAPI shine in advanced Domain Driven Design?

Ideas for the implementation of a generic service bus in Python.

Last update: 2th Sept 2023

Get 30 days of access to the example code of the overall blog post series on GitHub.

Introduction

FastAPI and all other Python backend (micro service, MVT) frameworks have not been designed with Domain Driven Design in mind. The traditional way how we use FastAPI to implement micro services is implementing the domain model either one of the following two business logic implementation patterns:

Transaction script pattern: The typical usage scenario of FastAPI when it comes to Artificial Intelligence and Machine Learning pipelines. E.g. some micro service providing one or several endpoints to allow people or other systems to trigger AI/ML model inference based on passed input data and to get a prediction in return.
Active record pattern: The typical usage scenario of FastAPI when it comes to fullstack applications. E.g. Using a SQL database, using SQLModel to map the database tables to Python objects/domain models, implementing business logic in the domain models and providing RESTful API CRUD endpoints for backend/frontend integration (like Sebastián Ramírezs full-stack-fastapi-postgresql template).

Over time I observed that advanced full-stack applications require to abstract business models and even more important business processes in a more complex manner.

Often we need to “model the real-world as a collection of entities” (aggregates, entities, value objects, etc.) whose data has stringent data consistency requirements. Typical examples are hierarchical data structures where the “most parent level” entity data needs to change in case the data of one of its “child level” entities changes. In such use case scenarios we should implement the micro service using the Domain model pattern.

Characteristics of the most advanced full-stack applications are that, in addition, they require to model data over time (e.g. in finance industry, in a lot of other regulated industries, whenever you need some log of data changes) instead of modelling data in a static state (like we do with SQL table records in CRUD-like applications). In that case we need to implement the Event-sourced domain model pattern.

Excerpt from “Figure E-1. Decision tree summarizing heuristics for tactical decision-taking”, Learning Domain Driven Design (Khononov, p. 298)

Side note: A typical backend/frontend integration interface of advanced DDD FastAPI micro services

In DDD a FastAPI micro service implements a bounded context (aka “business context”). Depending on the complexity of the micro service it may have several “communication interfaces”:

“Communication interface” to client side code (web frontend or mobile app).
“Communication interface” to external systems or services outside the own company.
“Communication interface” to other micro services implementing other bounded contexts.

When using an HTTP interface between backend micro service and frontend side client code usually you do something like:

For DDD queries implement one or several HTTP GET endpoint/s.
For “data stream DDD queries” or “real-time notifications” implement HTTP WebSockets endpoint/s.
For DDD commands implement one or several HTTP endpoint/s using POST endpoint/s.

Using an HTTP interface between backend and frontend will be your preferred choice how to implement the backend/frontend integration interface. However with FastAPI you can also use GraphQL instead:

For DDD queries use one or several GraphQL Queries.
For DDD “data stream DDD queries” or “real-time notifications” implement one or several GraphQL Subscriptions.
For DDD commands implement one or several GraphQL Mutations.

To learn more about the options available for implementing GraphQL backend/frontend interfaces with FastAPI head over to my blog post FastAPI Micro Service Patterns: GraphQL API.

Why are we not there yet when it comes to advanced DDD?

There is not only communication between the micro service and the client side code. If we implement the Domain model pattern or Event-sourced domain model pattern we need to communicate

Domain commands inside of the micro service like e.g. create an entity.
Domain events inside of a micro service like e.g. entity has been created/updated/deleted.

This Domain Commands and Domain Events are used by Domain services to trigger the execution of business logic based on the type of Domain command or Domain event and the data included in those.

E.g. the Domain command “create entity X” could trigger a Domain service to persist the entity in the database, create a Domain event “entity X created”, which would trigger a Domain service to publish “the new entity created + it’s data” on a WebSocket endpoint to display it in the frontend.

In addition we need to consider that an overall application (e.g. enterprise system) usually consists not only of a single micro service but a lot of micro services and external systems and services. This means we need to communicate Domain events and Domain commands not only inside a single DDD micro service but between (at least our own) DDD micro services in addition.

In most micro service usage scenarios we want to enable scaling the micro service horizontally.

As a consequence the most suitable communication mechanism for Domain events and Domain commands is asynchronous messaging.

FastAPI, Flask, Django, Sanic, … does not matter

It does not make much of a difference when it comes to development methodology, design and communication inside a bounded context w.r.t. framework in use… Django or some other Python library instead of FastAPI as application framework.

The main difference between micro service and MVT framework is related to scalability. From a DDD high level perspective the business logic related code is pretty much the same. The differences are more-less implementation details: E.g. using Django-Ninja instead of FastAPI HTTP endpoints, using the Django ORM instead of SQLModel for ORM, etc.

The need for asynchronous communication of Domain events and Domain commands are independent of the Python framework in use.

If you need a Python micro service framework… you stick to FastAPI. No one of us creates a new micro service framework from scratch if we start a new project. Do not reinvent the wheel… this is what Open Source is all about.

Let’s get some inspiration!

In contrast to my previous blog posts where I usually started with explaining first I’ll do it the other way around this time. This way it could be easier to get into the topic.

The “missing peace in the puzzle” might have different names. Some people call what we are looking for event bus (“event bus” GitHub projects), some call it message bus (“message bus” GitHub projects), some call it service bus (“service bus” GitHub projects). We might find something interesting about distributed messaging (“Distributed messaging” GitHub projects) for inspiration in addition.

In the DDD context the terms event bus and message bus are misleading. We need to integrate (micro) services. Means the term service bus is the most suitable for us now.

Let’s continue and check out some of those projects on GitHub…

There are several projects implementing a service bus. However there is not a single widely used/liked one implemented in Python.

Rebus

The first reasonable hit is the Rebus GitHub project and organization. It is not the most widely used service bus out there for sure but nevertheless a pretty good example from the C# / .NET world for a possible overall component structure of a service bus. If you check out the GitHub project structure you’ll figure out that…

Rebus supports several serializations implementations:

Protobuf via Rebus.Protobuf
MsgPack via Rebus.MsgPack
Hyperion via Rebus.Hyperion
Ceras via Rebus.Ceras

Rebus supports several transport layer implementations:

RabbitMQ via Rebus.RabbitMq
Amazon SQS via Rebus.AmazonSQS
Azure Service Bus via Rebus.AzureServivceBus
Azure Queues via Rebus.AzureQueues
SignalR via Rebus.SignalR
MSMQ via Rebus.Msmq
Google Cloud PubSub via Rebus.GoogleCloudPubSub

Rebus supports several database persistence implementations:

MongoDB via Rebus.MongoDb
PostgreSQL via Rebus.PostgreSql
Microsoft SQL Server via Rebus.SqlServer
RavenDb via Rebus.RavenDb
MySQL via Rebus.MySQL

Rebus supports several integrations for logging:

NLog via Rebus.NLog
Serilog via Rebus.Serilog
Microsoft Extensions Logging via Rebus.Microsoft.Extensions.Logging
Log4Net via Rebus.Log4net

Rebus supports several integrations for tracing:

OpenTelemetry via Rebus.OpenTelemetry

The first thing we should notice is… service busses should be implemented in a composable manner.

We should be able to select a serialization implementation for the serialization of Domain events and Domain commands from a variety of several options.
We should be able to select a transport layer implementation for the management and transmission of Domain event and Domain command messages. (Spoiler: This selection is critical when it comes to reliability of message delivery.)
We should be able to select a database persistence implementation for the storage of Sagas, for supporting Outbox pattern / transactional outbox pattern, message topics, message timeouts, transport implementations of specific subscriptions, etc.
We should be able to select a logging integration for integrating logging with a logging solution which fits our needs best.
We should be able to select a tracing integration for integrating tracing of message transmissions with a tracing solution which fits our needs best.

There are several scenarios w.r.t. to a service bus in the context of integrating several micro services:

There is only one bounded context/micro service. We do not need to integrate with other in-house services and out-of-house services (e.g. web services). This case is super rare. → We can use a single transport layer messaging technology for micro service specific Domain events and Domain commands. (An in-memory implementation might be ok.)
There are several in-house bounded contexts (several micro services) and no need to integrate with other in-house services and out-of-house services (e.g. web services). → We have no full control about transport layer messaging technology. We can use our own choice for in-house micro service integration but might be forced to use another transport layer messaging technology for integrating with out-of-house services. (An in-memory implementation will not work.)
There are several in-house bounded contexts (micro services) and we need to integrate with other in-house services and/or out-of-house services. To most people it’s not obvious in the first place but that’s the most common case actually. → We have no full control about transport layer messaging technology.

Especially in bigger enterprise scenarios where a lot of internal services will be existing in Java, C# or Go already we might be forced to integrate with a transport layer message technology commonly used in those languages and/or environments (e.g. NATS in case of Go, Azure Service Bus in case of C#).

The second thing we should take away is… a service bus should be able to support not only one but several transport layer implementations at the same time. This way we can e.g. use a transport layer with reliable message delivery for integrating our own micro services and we can live with a worse (less reliable) transport layer for integrating with out-of-house services or internal services we’ve no control about the transport layer potentially. Of course one should aim at using a single transport layer only: The more different transport layer implementations are in the system the worse the overall system integrate-ability (Sagas across several micro services integrated with different transport layer messaging technology?) and overall system quality characteristics (debug-ability, maintain-ability, etc.).

MassTransit

Another service bus framework in the C# world is MassTransit. It’s well-known in the community (according to GitHub stars) and has pretty similar high-level components like Rebus: serializers, monitoring, persistence, transports, testing.

In addition MassTransit introduces two, as far as I know, framework specific concepts:

Scheduling: Unlike other frameworks MassTransit supports two different methods of message scheduling: Scheduler-based message scheduling (with either Hangfire or Quartz) and transport layer (builtin) message scheduling.
Riders: “Kafka topics/Event hubs can be consumed using MassTransit consumers and sagas, including saga state machines, and messages can be produced to Kafka topics/Event hubs. Means MassTransit distinguish between message technologies and data streaming technologies.

MassTransit NuGet Package/component overview.

NServiceBus

NServiceBus is the least transparent framework. Like Rebus it’s developed and maintained by a company. The overall component architecture is pretty much the same:

transports: NServiceBus.Transport.AzureServiceBus and others…
persistence: NServiceBus.Persistence.Sql and others…
serializations: NServiceBus.Newtonsoft.Json and others…
debugging: ServiceInsight (“marketing” website)
monitoring: ServicePulse (“marketing” website)
testing: Testing framework

NServiceBus is specifically “marketing” finance, healthcare and retail use cases.

A first comparison

Conceptually all service bus frameworks in the C# world are pretty similar. Rebus is kind of unique due to Rebus Pro which is a cloud-hosted or self-hosted Fleet Manager (for debugging and monitoring). In the Python world this would not be required in the first place cause we can self-host on our own as well.

W.r.t. community adoption MassTransit has by far the most traction in recent years.

C# service bus framework GitHub star history.

Let’s have a more detailed look into the MassTransit concepts and API now…

Let’s have a look into MassTransits concepts

Concept wise MassTransit has:

Messages
(Message) Consumers
(Message) Producers
Exceptions
Requests
Mediator
Riders
Testing (not really a concept but worth having a look at nevertheless)

Messages

In MassTransit, a message contract is defined code first by creating a .NET type. A message can be defined using a record, class, or interface. Messages should only consist of properties, methods and other behavior should not be included.

MassTransit docs — Messages

Messages are used to implement Domain events as well as Domain commands.

Messages can be created from different types.

Message attributes contains configuration data.

The message header contains meta-data and implements the Envelop Wrapper pattern. “How can existing systems participate in a messaging exchange that places specific requirements on the message format, such as message header fields or encryption? … Use a Envelope Wrapper to wrap application data inside an envelope that is compliant with the messaging infrastructure. Unwrap the message when it arrives at the destination.”

Message correlation is a very important concept: “Messages are usually part of a conversation and identifiers are used to connect messages to that conversation.” Messages are identified by Guids which are unique, sequential, clustered-index friendly and ordered so that SQL Servers can efficiently insert them into a database with UI as primary key.

Consumers

Consumer is a widely used noun for something that consumes something. In MassTransit, a consumer consumes one or more message types when configured on or connected to a receive endpoint. MassTransit includes many consumer types, including consumers, sagas, saga state machines, routing slip activities, handlers, and job consumers.

MassTransit docs — Consumers

Consumers handle the message types. MassTransit distinguishes two types of consumers:

Message Consumers
Batch Consumers

Message consumers

In the easiest case the message is not part of a message sequence involving state which is managed by a Saga. In this case e.g. for a SubmitOrder message a SubmitOrderConsumer implements an IConsumer<SubmitOrder> interface.

class SubmitOrderConsumer :
    IConsumer<SubmitOrder>
{
    public async Task Consume(ConsumeContext<SubmitOrder> context)
    {
        await context.Publish<OrderSubmitted>(new
        {
            context.Message.OrderId
        });
    }
}

MassTransit integrates with ASP.NET Core or .NET Generic Host applications. services.AddMassTransit adds MassTransit to the overall application using dependency injection as a service. In .NET a service is a component usable in the overall application context.

The consumer is registered using one of the x.AddConsumer methods, the consume topology is configured (here in-memory transport using x.UsingInMemory) and the endpoint configured with defaults (cfg.ConfigureEndpoints). Instead configuring the consumer and the receiver implicitly they can be configured explicitly as well. Consumers can be configured explicitly with consumer definitions. Receiver endpoints can be configured explicitly using ConfigureConsumer/s.

services.AddMassTransit(x =>
{
    x.AddConsumer<SubmitOrderConsumer>();

    x.UsingInMemory((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });
});

What happens when a message is received:

On a receive endpoint a message is delivered.
Message type is consumed by consumer.
MassTransit creates a container scope.
MassTransit resolves a consumer instance.
Masstransit executes the consume method and passes the message as part of the ConsumeContext.
The consume method returns a Task.
MassTransit awaits for the Task.
If the Task completed successfully: MassTransit acknowledges the message and removes the message from the queue.
If the Task does not complete successfully (either due to exception, implicit or explicit cancellation): MassTransit releases consumer instance, exception is propagated back up the pipeline, depending on pipeline configuration the message transmission is retried or the message put onto a error queue.

If a message is part of a business process consisting of a sequence of messages and state a Saga state machine must be registered and a consumer saga added to it instead of a simple, stateless consumer.

MassTransit distinguishes Consumer Sagas and State Machine Sagas. State Machine Sagas are the recommended Saga types to use in MassTransit. This Saga implementation uses MassTransits own state machine library Automatonymous (maintained and under full control of MassTransit). ConsumerSagas have the purpose of helping to move applications from other Saga implementations to MassTransit.

A message part of a state-full message sequence must be linked to the Saga implementation used during the overall registration of a Saga. Here’s an API example… e.g. the OrderSaga as part of the registration of a Saga state machine using the AddSaga methods with in-memory repository (InMemoryRepository).

services.AddMassTransit(r =>
{
    // add a state machine saga, with the in-memory repository
    r.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .InMemoryRepository();

    // add a consumer saga with the in-memory repository
    r.AddSaga<OrderSaga>()
        .InMemoryRepository();
});

Batch consumers

When a lot of messages need to be handled (rough estimate: thousands of messages per second) writing every single message to storage can be a resource bottleneck. To workaround this MessageTransit supports receiving multiple messages and delivering messages to consumers as part of a batch.

A batch consumer needs to implement the Batch<T> interface. The example below consumes a batch of OrderAudit events, up to several at a time, and up to several concurrent batches (default PrefetchCount and ConcurrentMessageLimit) and is configured implicitly. Instead configuring the consumer and the receiver implicitly they can be configured explicitly as well. Consumers can be configured explicitly with consumer definitions. Receiver endpoints can be configured explicitly using ConfigureConsumer/s.

class BatchMessageConsumer :
    IConsumer<Batch<Message>>
{
    public async Task Consume(ConsumeContext<Batch<Message>> context)
    {
        for(int i = 0; i < context.Message.Length; i++)
        {
            ConsumeContext<Message> message = context.Message[i];
        }
    }
}