In this article, we will explore how to adapt the Ports and Adapters design pattern for use with popular frameworks and libraries. As modern software development heavily relies on these tools to streamline the development process, it is crucial to understand how to integrate the Ports and Adapters pattern effectively.
What is the main reason of IT projects failure.
US Air Force’s Expeditionary Combat Support System (ECSS) project, which was intended to create a single system for managing logistics across the Air Force was to consolidate and replace 200 independent outdated systems. The project was initiated in 2005 with a budget of $1.1 billion and was expected to be completed by 2013. However, after several delays and cost overruns, the project was ultimately cancelled in 2012, with an estimated cost of $1 billion. USAF concluded that the system, “has not yielded any significant military capability” and estimated that, “it would require an additional $1.1 billion for about a quarter of the original scope to continue and fielding would not be until 2020.”
Often when we hear about IT project failures, we imagine poor management, unmotivated engineers, financial frauds, or underestimated budgets. While these factors can contribute to failure, the main reason startups and projects fail is the high cost of maintenance and development. Typically, a startup is developed quickly with basic functionality, prioritizing speed over code quality and architecture. The initial focus is on proving the concept, which often lacks significant revenue. Software development relies on simple MVC frameworks that enable rapid development through automatically generated database schemas, active record CRUD operations, and APIs handled by the framework’s libraries.
However, when an application’s complexity grows, both technically and in terms of business requirements, problems begin to emerge. Business logic rules become scattered across various code layers, making it difficult to understand the code’s behavior. Developers must navigate multiple layers of code, memorizing the sequence of business rule execution and the impact of one rule on another. Additionally, business rules are often written in technical jargon, distancing them from real-world business scenarios.
As the code becomes harder to read and understand, it also becomes more challenging to modify. Developers may become specialized in specific areas of the code, leading to a lack of comprehensive understanding of the application. This can create significant challenges when onboarding new employees, as the process of understanding the code takes months.
As a result, implementing similar business requirement changes takes increasingly more time and effort as the project progresses. This exponential growth in complexity often leads businesses to allocate larger budgets to hire additional engineers. However, simply increasing the number of engineers does not guarantee improved efficiency or code quality, due to the challenges previously described.
Business logic separation
Our goal is to accurately represent the business concept within the code. This means that minor changes in business behavior should correspond to small adjustments in the code, while substantial coding tasks should be reserved for significant changes in the project domain only and business logic should be easy to read from the code and easy to understand. Additionally, the time required for making changes in the code should not increase throughout the project’s lifetime.
The code should consist of two distinct types of layers:
- Business layer (also referred to as the domain layer), where the domain logic of our application resides. This layer should represent concepts such as User, Product, Shopping Cart, Discount, Delivery, Loyalty Program, and so on.
- Implementation layers that contain all the necessary technical logic to execute the code. These layers should utilize concepts such as database, SQL query, API call, file, HTTP request, HTTP response status, Cloud Service, etc. While it is essential to divide these layers into separate technical components responsible for presentation, application, and infrastructure, a detailed discussion of this division is beyond the scope of this article.
The primary rule and a straightforward test for ensuring proper separation is to ensure that no technical “if” conditions exist within the business layer, and no business conditions are present in the implementation layer.
Ports And Adapters
Ports and Adapters, also known as the Hexagonal Architecture, is a design pattern that aims to improve software maintainability and testability by clearly separating the core business logic from external dependencies and implementation details. The main idea is to create a central, domain-centric core that is surrounded by various adapters that communicate with the external world.
On the other hand, adapters are responsible for facilitating communication between the domain and non-business components of the software, such as databases, APIs, and file systems. Adapters implement interfaces defined by the domain in the ports layer, ensuring a clear distinction between the layers. The domain layer remains unaware of the adapters, as adapters are considered implementation details. Consequently, adapters must conform to the ports, effectively adapting to the requirements of the domain.
Domain with ports
The domain consists of classes that represent business concepts, describing the behavior and data of the system. It encompasses the understanding of business concepts within a given domain. In the context of a store, this may include entities such as “User”, “Sale”, “Product”, “Order”, and “Discount Calculation Strategy”, among others.
Domain classes should not merely mirror the data structure in the persistence layer. While entities may occasionally resemble SQL records and properties might share names with SQL columns, this should not be a guiding principle. When designing the domain, architects or engineers should focus on business purposes and needs. Decisions regarding how to store an entity, whether to use a relational or non-relational database, and how many tables or column names to employ should be made independently. Otherwise, technical details may begin to seep into the domain layer.
A key component of the domain is a set of ports that serve as a bridge between business rules and technical implementations or external systems. Ports can be categorized into two primary types.
Input ports
Input ports, also known as driving or primary ports, define an interface that can be invoked from outside the domain to trigger the execution of domain-specific code. It’s important to note that the concept of a port as an interface is not a strict rule dictating its appearance. These ports are not typically defined as interfaces or abstract classes (although they can still be used) but are usually directly implemented by a Service class. Input ports guide the technology layer in executing technology-agnostic classes. Adapters often utilize input ports to handle API calls, event processing, scheduled tasks, or shell commands.
from decimal import Decimal
class AddProductUseCase:
"""
Domain model
"""
def execute(self, cart_id: int, product_id: int, amount: int) -> bool:
"""
Some business logic required to add the product
"""
class CalculateFinalPriceUseCase:
"""
Domain model
"""
def execute(self, cart_id: int) -> Decimal:
"""
Some logic that calculates final price
"""
class ShoppingCartService:
"""
Input port example
"""
def __init__(self, cart_id: int)
self._cart_id = cart_id
def add_product(self, product_id: int, amount: int) -> bool:
add_product_use_case = AddProductUseCase()
return add_product_use_case.execute(self.cart_id, product_id, amount)
def calculate_final_price(self):
calculate_use_case = CalculateFinalPriceUseCase()
return calculate_use_case.execute(self._cart_id)
In the example above, we defined two input ports as methods, add_product and calculate_final_price, within the ShoppingCartService class. These methods can be called from the adapters layer to execute domain logic without exposing too much of the domain implementation details to the adapters.
Output ports
Output ports define the interfaces for the technical components required by the domain to function. These components must be supplied to the domain layer before executing domain code. This can be done through the constructor of the class implementing the Input port or, in more complex projects, by utilizing a dependency injection framework. While the latter approach adds an extra layer of abstraction and can make reading and debugging the code more challenging, it may still be more manageable than handling a large number of dependency initializations. Output ports are implemented by adapters, with the most common example being the persistence layer responsible for storing and retrieving domain object data.
import abc
class ShoppingCartRepositoryInterface(abc.ABC):
"""
Shopping cart repository output port.
"""
@abc.abstractmethod
def add_product_to_shopping_cart(self, cart_id: int, product_id: int, amount: int):
pass
class ShoppingCartProduct:
"""
Business model of a shopping cart product.
"""
@property
def id(self) -> int:
return self._id
@property
def amount(self) -> int:
return self.amount
class ShoppingCart:
"""
Business model of a shopping cart
"""
def __init__(self, cart_id: int, shopping_cart_repository: ShoppingCartRepositoryInterface):
self._cart_id = cart_id
self._shopping_cart_repository = shopping_cart_repository
def add_product(self, product_id: int, amount: int):
"""
Some business logic required to add the product
"""
self._shopping_cart_repository.add_product_to_shopping_cart(
cart_id=self._cart_id,
product_id=product_id,
amount=amount
)
"""
Another business logic after adding cart
"""
return True
In the example above, we’ve established an output port called ShoppingCartRepositoryInterface, which enables the ShoppingCart domain model to add products to a persistence layer implemented by an adapter.
Data Transfer Object (DTO)
Ports should not expect to receive from or return to an adapter a domain object. We don’t want external technical components to be responsible for creating domain objects, as it promotes less coupling and makes the domain layer easier to maintain and modify. Ports should define a simple data structure, and in cases where a more complex structure is needed, a DTO can be used. A DTO can be a dictionary, list, or class focused on storing data without containing any logic. Using a class provides a significant advantage, as it clearly defines the data structure. In Python, an excellent choice for DTOs would be data classes or named tuples.
Let’s define a simple output port called ShoppingCartRepositoryInterface for storing and retrieving shopping cart data from a persistence layer.
import abc
import dataclasses
from decimal import Decimal
from typing import List, Literal, Optional
@dataclasses.dataclass
class ShoppingCartItemDTO:
UNIT_KG = 'kg'
UNIT_LB = 'lb'
UNIT_PIECE = 'piece'
product_id: int
price_per_unit: Decimal
units: Literal[UNIT_KG, UNIT_LB, UNIT_PIECE]
amount: Decimal
@dataclasses.dataclass
class ShoppingCartDTO:
user_id: Optional[int]
items: List[ShoppingCartItemDTO]
class ShoppingCartRepositoryInterface(abc.ABC):
@abc.abstractmethod
def get_shopping_cart(self, cart_id: str) -> ShoppingCartDTO:
pass
@abc.abstractmethod
def store_shopping_cart(self, cart_id: str, cart: ShoppingCartDTO) -> bool:
passThe persistence layer could be a database or any other storage system. The choice is an implementation detail that can be made later in an adapter implementation. Just imagine how easy it would be to replace an SQL database with a document database in this setup.
Adapters
Let’s delve into the technological aspects now. This is something that developers enjoy the most, and that’s why many of them may feel upset and disappointed that significant elements like databases and communication interfaces are reduced to a simple adapter written just to satisfy a single port of the domain. A common approach is to start by choosing a database, designing the schema and relations (if a relational database is selected), and deciding whether to use a REST API or RabbitMQ messages, etc. Indeed, we postpone these technical decisions, as database design is only a part of a project’s success. It describes data structure only, while the primary purpose of building software is its behavior, not merely storing data. By starting with the domain and business model, we gain a better understanding of the data we need to store and the communication we need to handle. This is crucial because if your technical vision does not align with the business vision and processes, the technical side must be adjusted. If your technical assumptions make adjustments very difficult or even impossible, you end up with inflexible code, even if you’ve used the most flexible programming language. Postponing technical architectural decisions allows for better decision-making.
Another advantage of using adapters is that, by defining clear, small ports, you can create small, easy-to-understand technical adapters. When you have small adapters with a clear interface, you can easily test, modify, or even rewrite them. The ability to rewrite parts of your software in just a few days when you decide that another technology could be better, more efficient, or more up-to-date is invaluable.
We can identify two types of Adapters based on the distinction between input and output Ports.
Input adapters
Input adapters invoke domain input interfaces to execute business rules. Typically, they are triggered externally. Some examples of input adapters include:
- REST API endpoints
- Event consumers
- Message queue brokers
- Database watchers
- Cron jobs
- WebSocket handlers
- GUI action handlers
- Command-line adapters
Let’s examine an example of an input adapter in Django Rest Framework that manages the addition of a product to the shopping cart. This adapter utilizes the ShoppingCartService to execute the business logic responsible for adding the product to the cart.
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from yourapp.ports.input import ShoppingCartService, SomeExceptions
class ShoppingCartProductView(APIView):
def post(self, request, id):
product_id = request.data.get('product_id')
amount = request.data.get('amount')
shopping_cart = ShoppingCartService(cart_id=id)
try:
shopping_cart.add_product(product_id, amount)
except SomeExceptions:
"""
Handle exceptions
"""
return Response(shopping_cart.get_product(product_id), status=status.HTTP_201_CREATED)
Output adapters
Output adapters implement output ports to provide the technical implementation required by the domain to process business use cases. For instance, if the domain needs to store an object in a persistence layer or call an external API, it requires technical code to accomplish that. Instead of mixing this technical design into the domain code, it defines a business port (usually an interface or abstract class) that will be implemented by the adapter. Examples of output adapters include:
- Repository of business objects
- Relational database
- Document database
- External API call
- Notifications
- Message Queue Producer
- External API Call
- WebSocket
- EmailService
- SMTP adapter
- External mailing service API
Let’s examine an example of output adapters in a simple application that generates verification codes and sends them via SMS.
"""
PORTS
"""
class SmsServiceInterface(abc.ABC):
"""
Example of domain output port to send registration code.
"""
@abc.abstractmethod
def send_sms(self, phone_number: str, content: str) -> bool:
pass
class VerificationsRepository(abc.ABC):
@abc.abstractmethod
def add(self, verification_id: str, phone_number: str, registration_time: datetime, authentication_code):
pass
"""
DOMAIN LOGIC
"""
class UserRegistrationUseCase:
def __init__(self, sms_service: SmsServiceInterface, registrations_repository: VerificationsRepository):
self._sms_service = sms_service
self._registrations_repository = registrations_repository
def execute(self, phone_number) -> str:
"""
Some business validation and logic here
"""
verification_hash_id = RandomCodes.generate(digits=32)
verification_code = RandomCodes.generate(digits=6)
self._verifications_repository.add(
id=verification_hash_id,
phone_number=phone_number,
registration_time=datetime.now(),
authentication_code=verification_code
)
self._sms_service.send_sms(
phone_number=phone_number,
content=f"Your verification code is {verification_code}"
)
return verification_hash_id
"""
ADAPTERS
"""
class SomeSmsServiceAdapter(SmsServiceInterface):
def __init__(self, api_key: str, sms_queue_url: str):
self._api_key = api_key
self._base_url = sms_queue_url
def send_sms(self, phone_number: str, content: str) -> bool:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self._api_key}",
}
data = {"phone_number": phone_number, "content": content}
requests.post(self._base_url, headers=headers, data=json.dumps(data))
class RedisRegistrationsRepository(VerificationsRepository):
def __init__(self, redis_client:StrictRedis):
self.redis_client = redis_client
def add(self, id: str, phone_number: str, registration_time: datetime, authentication_code) -> None:
data = {
"phone_number": phone_number,
"registration_time": registration_time.isoformat(),
"authentication_code": authentication_code,
}
self.redis_client.set(id, json.dumps(data))
An example implementation in Python
In this blog post, we’ll examine an example of code written using the Django Rest Framework. This simplified Django view handles an endpoint that returns a summary of the customer’s shopping cart. It also calculates a discount on the second and every subsequent product. Importantly, this discount does not overlap with other sales discounts. Instead, it compares the specific product’s sale discount value and applies the larger discount. Additionally, the code checks the delivery price for the order and returns the values for two delivery companies managed by the store.
class ShoppingCartDiscountView(APIView):
def get(self, request, shopping_cart_id):
shopping_cart = ShoppingCart.objects.get(id=shopping_cart_id)
if shopping_cart == None:
raise Http404
if shopping_cart.status in [STATUS_ORDERD, STATUS_PRE_DELIVER, STATUS_PAYMENT_IN_PROGRESS, STATUS_PAYMENT,
STATUS_SENT, STATUS_DELIVERED, STATUS_RETURNED, STATUS_REPAID]
raise Http404
products_with_amount = shopping_cart.products.annotate(
amount_available=ExpressionWrapper(F('warehouse__amount') - F('warehouse__amount_reserved'),
output_field=IntField())
).order_by("-price_base_gross")
products_output = []
weight_kg = 0
can_be_ordered = True
for index, product in enumerate(products):
sale_discount = None
if index == 0:
final_price_gross = product.price_base_gross
else:
if product.price_sale * product.tax_level < product.price_base_gross:
sale_discount = "sale"
final_price_gross = product.price_sale * product.tax_level
else:
sale_discount = "discount"
final_price_gross = product.price_base_gross * DISCOUNT_SECOND_AND_NEXT_PRODUCT_VALUE
if product.amount_available > product.max_items_to_order:
amounts_left = -2
can_be_ordered = False
elif product.amount_available < product.amount:
amounts_left = product.amount_available < product.amount
else:
amounts_left = -1
if can_be_ordered:
if product.weight_unit == 'kg':
product_weight_kg = product.weight
else:
# Weight is in pounds if is not in kg
product_weight_kg = product.weight * converters_units.POUNDS_TO_KILOGRAMS_RATIO
weight_kg = weight_kg + product.amount * product.weight
products_output.append({
"product_id": product.id,
"product_name": product.name_short,
"product_items": product.amount,
"product_price_gross": product.price_base_gross,
"product_prace_gross_sale_discount": final_price_gross
"sale_or_discount_price": sale_discount,
"amounts": amounts_left
})
if shopping_cart.address == -1:
# default user address
if shopping_cart.user.different_delivery_address:
address = shopping_cart.user.delivery_addres
else:
address = shopping_cart.user.registration_addres
else:
address = shopping_cart.address
if can_be_ordered:
data = {
"weight": weight_kg,
"state": address.state
}
response = requests.post(FEDEX_URL, data=data)
if response.status_code == 200:
response_data = response.json()
price = response_data.get("price", None)
if price is not None:
fedex_delivery_price = float(price)
else:
logger.log("Price not found in the response.")
fedex_delivery_price = -1.0
else:
logger.log(f"Request failed with status code {response.status_code}")
fedex_delivery_price = -1.0
ups_delivery_price_range = UpsDeliveryPrices.objects.filter(weight_kg__lte=weight_kg).order_by(
"-weight_kg").fetch_one()
cart_response_data = {
'cart_items': products_output,
'delivery_prices': {
'fedex': fedex_delivery_price,
'ups': ups_delivery_price_range.price
}
}
return Response(cart_response_data)This example serves as an excellent candidate for a 10,000-liner class, because in real-life scenarios, the logic would be far more complex. Who hasn’t encountered such code? Each additional logic rule makes the code increasingly difficult to understand. Consequently, it becomes challenging to understand what the code does and test it in any of both unit and integration tests. Moreover, it’s nearly impossible to reuse the code in different contexts, especially when a significant portion of the logic must be repeated during the order submission process.
Let’s examine the domain layer of the same code and compare its readability for someone encountering it for the first time. Notice how much easier it is to understand the domain rules and their purpose when they are presented in this manner.
class GenerateShopingCartSummaryUseCase(ShoppingCartSummaryInterface):
def __init__(self, shoping_cart: ShoppingCartRepository, delivery_service: IDeliveryService):
self._shoping_cart_repository = shoping_cart
self._delivery_service = delivery_service
def execute(self, shopping_cart_id) -> ShoppingCartSummaryDTO:
shopping_cart = self._shoping_cart_repository.get_shopping_card(shopping_cart_id)
if shopping_cart == None or not shopping_cart.await_for_purchase():
raise ShoppingCartNotFound()
shopping_cart.apply_discounts(
discounts_merge_strategy=SingleHighestDiscount(),
discounts = [
FromSecondChepperProductsDiscount(
discount_percentage=10,
)
]
)
delivery_options = self._delivery_service.calculate(
services=[DELIVERY_FED, DELIVERY_UPS],
delivery_attributes=DeliveryAttributes(
address=shopping_cart.get_delivery_address,
weight=shopping_cart.get_summary_weight()
)
)
return ShoppingCartSummaryDTO(
shopping_cart_id=shopping_cart.id,
products=shopping_cart.products,
delivery_options=delivery_options
)Rewrite or refactor?
Although this article is not specifically about refactoring, it is essential to address the topic briefly. How often have you heard someone say about a several-years-old codebase, “I would love to start from scratch”? In most cases, these statements come from people who continue to write legacy code. Starting with a clean slate may seem attractive, but in most cases, the project would likely encounter similar issues after some time. Of course, we cannot avoid replacing old code with new code, but it is generally inadvisable to do so all at once.
Consider a scenario where you have a large warehouse handling deliveries for a factory. Production has grown rapidly, and the warehouse has expanded multiple times without any reorganization of workflow. Workers still operate as if it were a small warehouse, placing items on various shelves and relying on their memory to locate them. As a result, the warehouse becomes increasingly disorganized, fulfilling orders takes longer, and sometimes it becomes nearly impossible to find products in a timely manner.
While the idea of emptying the warehouse and starting anew with a well-organized system might be tempting, the most reasonable solution is to tackle the problem step by step without disrupting the production and delivery processes. The first step would be to stop adding to the chaos and begin placing new items in their proper locations. By doing so, order will gradually replace disorder, leading to a more efficient warehouse operation.
Usage with libraries
Often, there are misunderstandings and questions such as: “Do I have to start implementing everything from scratch and forget about my favorite tools like Django Rest Framework, Flask, or SQLAlchemy?” The answer is no, there’s no need to write SQL queries manually or lose the advantages of Django modules. Instead, these tools should be employed at the technical level of the Adapter.
Let’s examine an example of an output adapter using SQLAlchemy to illustrate this point:
"""
DOMAIN PORTS
"""
@dataclass
class NewProductDTO:
id: str
category_name: str
code: str
name: str
description: str
weight_kg: int
class ProductsRepository(abc.ABC):
@abc.abstractmethod
def add_product(self, product: NewProductDTO):
pass
"""
ADAPTERS
"""
Base = declarative_base()
class Category(Base):
__tablename__ = 'categories'
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
products = relationship('Product', back_populates='category')
class Product(Base):
__tablename__ = 'products'
id = Column(String, primary_key=True)
code = Column(String)
name = Column(String)
description = Column(String)
weight_kg = Column(Integer)
category_id = Column(Integer, ForeignKey('categories.id'))
category = relationship('Category', back_populates='products')
class SQLAlchemyProductsRepository(ProductsRepository):
def __init__(self, session):
self.session = session
def add_product(self, product_data: NewProductDTO):
category = self.session.query(Category).filter_by(name=product_data.category_name).first()
if not category:
category = Category(name=product_data.category_name)
self.session.add(category)
product = Product(
id=product_data.id,
code=product_data.code,
name=product_data.name,
description=product_data.description,
weight_kg=product_data.weight_kg,
category=category
)
self.session.add(product)
self.session.commit()
Although Ports and Adapters does not prohibit having SQL relations between tables managed by different adapters, you should carefully consider your approach. Do you need database consistency in that specific area? You may want to contemplate whether it’s necessary, or if another solution like data redundancy might be more suitable. While cross-adapter constraints don’t directly break the rule of separating business logic from technical details, your system will become more coupled, and hidden business rules may be implemented on the technical level (e.g., database queries) such as “You cannot add a product to a category that doesn’t exist” or “What should happen when you delete a category that contains products – delete products, move them to the default category, or leave them with no category?” These concerns should typically be addressed by domain logic, and applying redundant rules on the implementation level could introduce unexpected behavior. Knowledge about these rules and potential database errors will need to be managed on both sides of the adapters involved in the relationship. Each case should be considered individually, but it’s certainly a good idea to use relations within a single entity that is managed by a single adapter.
When should you NOT use it.
The Ports and Adapters pattern, along with the entire Domain-Driven Design approach, is a solution best suited for addressing the complexities arising from intricate business logic. It is most applicable in software systems with deep business logic, numerous rules, and dependencies. However, it is not advisable to use this pattern for simple CRUD applications with limited domain logic, as the effort to model such systems would outweigh the benefits. In these cases, MVC frameworks would be a more suitable choice.
Similarly, the Ports and Adapters pattern may not be the best fit for rapid prototyping or applications with a short lifespan that are created for a specific task and then discarded. Additionally, in applications that are tightly coupled to a particular technology, separating domain concerns from technology-related concerns might prove challenging, making this pattern less effective in such cases.
As there are various types of software, applications may also have different components or modules. In large, complex applications where only some parts involve intricate business logic, it is not necessary to implement the Ports and Adapters pattern throughout the entire system. What’s important is to separate the domain model from these parts using ports and avoid mixing different application components. For instance, if you have a group of tables with CRUD-only operations, then Django’s model serializer would be an ideal solution for handling them. However, when you need to use this data in a domain model, it’s essential to define a port and create a simple adapter. In straightforward cases, you can even use or extend a Django model class to implement the output port and provide required data.
Where to use it?
Let’s take a quick look at cases where Ports and Adapters typically fit very well:
- Enterprise applications with complex and deep business layers.
- Software that requires high test coverage. The separation of business logic and injectable adapters makes it easy to mock and perform fast unit tests. The technical layer, divided into small adapters with clear interfaces, is easy to test in integration tests.
- Software that undergoes rapid changes. It helps to avoid clutter in the code caused by shortcuts.
- Microservices and modular applications – using ports to communicate between microservices or modules can help create clear communication interfaces.
- Applications with multiple interfaces (HTML Templates + REST API) – different interfaces usually require different input adapters. The rest of the application (Domain, output adapters) remains the same. Even if not, large parts of the system components can be reused.
