In this post we will explain the approach to apply Clean Architecture in a Django Restful API backend project. It will be useful for this reading to be familiarized with Django and Django Rest Framework as well as with Clean Architecture.
In no case does this post intend to reject Django components and apps architecture, nor misuse it, but to propose an alternative software architecture for Django projects that may be more convenient in certain cases.
The Clean Architecture is an alternative pattern intended to reduce the cost of changes that may be useful in some contexts:
- Easier and faster to build new features.
- Less cost overhead for software development.
- More independence of solution components meaning less refactoring.
- Improved testability leading a better quality software.
These objectives are achieved by dividing the software into layers, each of them with its own concerns.
We will use as an example the code from Forex API backend. This application aims to provide information for all currencies, latest exchange rates and historical time series exchange rates for currency pairs, currency conversion and calculation of time-weighted rates. The explanation of the architecture is structured using the same layers in the diagram.
The code snippets used in this post have been taken from the source code repository of the corresponding project. The full code is available in the Github repository.
1. Entities Layer
This is the innermost layer where we define entities. These entities encapsulate our enterprise wide business logic and high level rules which are the least likely to change.
domain/exchange_rate
from dataclasses import dataclass
from datetime import date
@dataclass
class CurrencyEntity:
code: str = None
name: str = None
symbol: str = None
@dataclass
class CurrencyExchangeRateEntity:
source_currency: CurrencyEntity = None
exchanged_currency: CurrencyEntity = None
valuation_date: str = None
rate_value: float = None
def __post_init__(self):
if self.valuation_date and isinstance(self.valuation_date, date):
self.valuation_date = self.valuation_date.strftime('%Y-%m-%d')
if self.rate_value:
self.rate_value = round(float(self.rate_value), 6)
def calculate_amount(self, amount: float) -> float:
return round(amount * self.rate_value, 2)
2. Use Cases Layer
The use cases, or interactors as we will call them, contain the application specific business rules for each use case. Therefore this layer contains all the application logic responsible of communicate with repositories, manage permissions, trigger side effects or handle exceptions.
usecases/exchange_rate
import datetime
from typing import List
from domain.exchange_rate import CurrencyExchangeRateEntity
class CurrencyExchangeRateInteractor:
def __init__(self, exchange_rate_repo: object):
self.exchange_rate_repo = exchange_rate_repo
def get(self, source_currency: str, exchanged_currency: str,
valuation_date: str) -> CurrencyExchangeRateEntity:
return self.exchange_rate_repo.get(
source_currency, exchanged_currency, valuation_date)
def get_time_series(self, source_currency: str, exchanged_currency: str,
date_from: str, date_to: str) -> List[CurrencyExchangeRateEntity]:
return self.exchange_rate_repo.get_time_series(
source_currency, exchanged_currency, date_from, date_to)
In this simple example the interactor is responsible for retrieving the storaged exchange rate information from the corresponding repository.
3. Interface Adapters Layer
This layer contains the pieces which are decoupled from the framework, like routes, controllers, serializers and repositories.
The repositories in this layer know about the data sources. They are responsible for selecting the data source, but do not access data directly.
interface/repositories/exchange_rate
from typing import List
from domain.exchange_rate import CurrencyExchangeRateEntity
class CurrencyExchangeRateRepository:
def __init__(self, db_repo: object, cache_repo: object):
self.db_repo = db_repo
self.cache_repo = cache_repo
def get(self, source_currency: str, exchanged_currency: str,
valuation_date: str) -> CurrencyExchangeRateEntity:
exchange_rate = self.cache_repo.get(
source_currency, exchanged_currency, valuation_date)
if not exchange_rate:
exchange_rate = self.db_repo.get(
source_currency, exchanged_currency, valuation_date)
self.cache_repo.save(exchange_rate)
return exchange_rate
def get_time_series(self, source_currency: str, exchanged_currency: str,
date_from: str, date_to: str) -> List[CurrencyExchangeRateEntity]:
return self.db_repo.get_time_series(
source_currency, exchanged_currency, date_from, date_to)
The controllers handle the REST API entry-points. In this case they follow Django's view structure but are completely decoupled from it. Each controller gets the needed interactors from a factory. They are responsible of parsing and/or validating the input data, calling the use cases and formating the output result or error with the defined serializers.
interface/controllers/exchange_rate
import logging
from http import HTTPStatus
from typing import Tuple
from domain.exchange_rate import CurrencyExchangeRateEntity
from interface.controllers.utils import calculate_exchanged_amount
from interface.repositories.exceptions import EntityDoesNotExist
from interface.serializers.exchange_rate import (
CurrencyExchangeRateAmountSerializer, CurrencyExchangeRateConvertSerializer)
from usecases.exchange_rate import CurrencyExchangeRateInteractor
from usecases.provider import ProviderClientInteractor
logger = logging.getLogger(__name__)
class CurrencyExchangeRateController:
def __init__(self, exchange_rate_interactor: CurrencyExchangeRateInteractor,
provider_client_interactor: ProviderClientInteractor):
self.exchange_rate_interactor = exchange_rate_interactor
self.provider_client_interactor = provider_client_interactor
def convert(self, params: dict) -> Tuple[dict, int]:
logger.info('Converting currency for params: %s', str(params))
data = CurrencyExchangeRateConvertSerializer().load(params)
if 'errors' in data:
logger.error('Error deserializing params: %s', str(data['errors']))
return data, HTTPStatus.BAD_REQUEST.value
amount = data.pop('amount')
try:
exchange_rate = self.exchange_rate_interactor.get_latest(**data)
except EntityDoesNotExist as err:
exchange_rate = self.provider_client_interactor.fetch_data(
'exchange_rate_convert', **data)
if not isinstance(exchange_rate, CurrencyExchangeRateEntity):
logger.error('Failure converting currency: %s', err.message)
return {'error': err.message}, HTTPStatus.NOT_FOUND.value
exchanged_amount = calculate_exchanged_amount(exchange_rate, amount)
logger.info('Currency successfully converted: %s', str(exchanged_amount))
return (
CurrencyExchangeRateAmountSerializer().dump(exchanged_amount),
HTTPStatus.OK.value
)
interface/serializers/exchange_rate
from marshmallow import Schema, fields
from marshmallow.decorators import post_load
from marshmallow.exceptions import ValidationError
class CurrencyExchangeRateConvertSerializer(Schema):
source_currency = fields.String(required=True)
exchanged_currency = fields.String(required=True)
amount = fields.Float(required=True)
def load(self, data: dict) -> dict:
try:
data = super().load(data)
except ValidationError as err:
data = {'errors': err.messages}
return data
@post_load
def make_upper_code(self, data: dict, **kwargs) -> dict:
data['source_currency'] = data['source_currency'].upper()
data['exchanged_currency'] = data['exchanged_currency'].upper()
return data
class CurrencyExchangeRateAmountSerializer(Schema):
exchanged_currency = fields.String(required=True)
exchanged_amount = fields.Float(required=True)
rate_value = fields.Float(required=True)
interface/routes/exchange_rate
from domain.core.routing import Route, Router
from interface.controllers.exchange_rate import CurrencyExchangeRateController
exchange_rate_router = Router()
exchange_rate_router.register([
Route(
http_verb='get',
path=r'^exchange-rate/time-weighted/$',
controller=CurrencyExchangeRateController,
method='calculate_twr',
name='exchange_rate_calculate_twr',
),
Route(
http_verb='get',
path=r'^exchange-rate/convert/$',
controller=CurrencyExchangeRateController,
method='convert',
name='exchange_rate_convert',
),
Route(
http_verb='get',
path=r'^exchange-rate/$',
controller=CurrencyExchangeRateController,
method='list',
name='exchange_rate_list',
),
])
4. Infrastructure Layer
This is the outermost layer on which the code related to those parts to abstract their implementations is placed, that is the framework, in our case Django and Django Rest Framework, and the drivers for third party services.
In our example, we have three main parts which are the api server, the repositories (database and cache), and the clients.
This is the only layer that knows all about data sources and is responsible of the correct communication with them. For the postgres relational database we have created a repository tied to Django ORM.
infrastructure/orm/db/exchange_rate/models
from django.db import models
class Currency(models.Model):
code = models.CharField(max_length=3, primary_key=True, unique=True)
name = models.CharField(max_length=50, blank=True, null=True)
symbol = models.CharField(max_length=1, blank=True, null=True)
class Meta:
verbose_name = 'currency'
verbose_name_plural = 'currencies'
ordering = ('code',)
class CurrencyExchangeRate(models.Model):
source_currency = models.ForeignKey(
Currency, db_index=True, on_delete=models.CASCADE, related_name='exchanges')
exchanged_currency = models.ForeignKey(
Currency, db_index=True, on_delete=models.CASCADE)
valuation_date = models.DateField(db_index=True)
rate_value = models.DecimalField(decimal_places=6, max_digits=18)
class Meta:
verbose_name = 'currency exchange rate'
verbose_name_plural = 'currency exchange rates'
ordering = ('-valuation_date', 'source_currency')
infrastructure/orm/db/exchange_rate/repositories
from typing import List
from domain.exchange_rate import CurrencyExchangeRateEntity
from infrastructure.orm.db.exchange_rate.models import CurrencyExchangeRate
from interface.repositories.exceptions import EntityDoesNotExist
class CurrencyExchangeRateDatabaseRepository:
def get(self, source_currency: str, exchanged_currency: str,
valuation_date: str) -> CurrencyExchangeRateEntity:
exchange_rate = CurrencyExchangeRate.objects.filter(
source_currency=source_currency,
exchanged_currency=exchanged_currency,
valuation_date=valuation_date
).values(
'source_currency', 'exchanged_currency', 'valuation_date', 'rate_value'
).first()
if not exchange_rate:
raise EntityDoesNotExist(
f'Exchange rate {source_currency}/{exchanged_currency} '
f'for {valuation_date} does not exist')
return CurrencyExchangeRateEntity(**exchange_rate)
def get_time_series(self, source_currency: str, exchanged_currency: str,
date_from: str, date_to: str) -> List[CurrencyExchangeRateEntity]:
timeseries = CurrencyExchangeRate.objects.filter(
source_currency=source_currency,
exchanged_currency__in=exchanged_currency.split(','),
valuation_date__range=[date_from, date_to]
).values(
'source_currency', 'exchanged_currency', 'valuation_date', 'rate_value')
return list(map(lambda x: CurrencyExchangeRateEntity(**x), timeseries))
As we can see, both the returned entity and the raised exception are custom designed objects, thus we hide all the Django ORM details.
For the Django views we define a view wrapper to hide the business logic details and decouple our views from the framework. With this view wrapper we achieve two main goals:
- Convert the incoming request data to pure Python objects.
- Format response so the view also returns pure Python objects.
As we will see later, the corresponding factory handles the creation of the view with all its dependencies.
infrastructure/api/views/exchange_rate
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import ViewSet
from interface.controllers.exchange_rate import CurrencyExchangeRateController
class CurrencyExchangeRateViewSet(ViewSet):
viewset_factory = None
@property
def controller(self) -> CurrencyExchangeRateController:
return self.viewset_factory.create()
def convert(self, request: Request, *args, **kwargs) -> Response:
query_params = request.query_params
payload, status = self.controller.convert(query_params)
return Response(data=payload, status=status)
In this layer we also encounter the rest of the components related to Django and Django Rest Framework like routes, urls, migrations, settings, admin,... Then, as we can see, this layer is completely coupled to Django framework and other third party libraries like Redis for cache or Celery for running asynchronous tasks.
infrastructure/api/routes/exchange_rate/routers
from rest_framework.routers import SimpleRouter, Route
from infrastructure.factories.exchange_rates import CurrencyExchangeRateViewSetFactory
from interface.routes.exchange_rate import exchange_rate_router
class CurrencyExchangeRateRouter(SimpleRouter):
routes = [
Route(
url=exchange_rate_router.get_url('exchange_rate_convert'),
mapping=exchange_rate_router.map('exchange_rate_convert'),
initkwargs={'viewset_factory': CurrencyExchangeRateViewSetFactory},
name='{basename}-convert',
detail=False
)
]
infrastructure/api/routes/exchange_rate/urls
from django.conf.urls import include
from django.urls import path
from infrastructure.api.routes.exchange_rate.routers import CurrencyExchangeRateRouter
from infrastructure.api.views.exchange_rate import CurrencyExchangeRateViewSet
exchange_rate_router = CurrencyExchangeRateRouter()
exchange_rate_router.register(
'', viewset=CurrencyExchangeRateViewSet, basename='exchange-rate')
urlpatterns = [
path('', include(exchange_rate_router.urls))
]
Now, the way we join all the described layers is by dependency injection. We achieve this defining factories which are in charge of solving these dependencies recursively, giving the responsability of each element to its own factory resolver.
infrastructure/factories/exchange_rate
from infrastructure.factories.provider import ProviderClientInteractorFactory
from infrastructure.orm.cache.exchange_rate.repositories import CurrencyExchangeRateCacheRepository
from infrastructure.orm.db.exchange_rate.repositories import CurrencyExchangeRateDatabaseRepository
from interface.controllers.exchange_rate import CurrencyExchangeRateController
from interface.repositories.exchange_rate import CurrencyExchangeRateRepository
from usecases.exchange_rate import CurrencyExchangeRateInteractor
class CurrencyExchangeRateDatabaseRepositoryFactory:
@staticmethod
def get() -> CurrencyExchangeRateDatabaseRepository:
return CurrencyExchangeRateDatabaseRepository()
class CurrencyExchangeRateCacheRepositoryFactory:
@staticmethod
def get() -> CurrencyExchangeRateCacheRepository:
return CurrencyExchangeRateCacheRepository()
class CurrencyExchangeRateRepositoryFactory:
@staticmethod
def get() -> CurrencyExchangeRateRepository:
db_repo = CurrencyExchangeRateDatabaseRepositoryFactory.get()
cache_repo = CurrencyExchangeRateCacheRepositoryFactory.get()
return CurrencyExchangeRateRepository(db_repo, cache_repo)
class CurrencyExchangeRateInteractorFactory:
@staticmethod
def get() -> CurrencyExchangeRateInteractor:
exchange_rate_repo = CurrencyExchangeRateRepositoryFactory.get()
return CurrencyExchangeRateInteractor(exchange_rate_repo)
class CurrencyExchangeRateViewSetFactory:
@staticmethod
def create() -> CurrencyExchangeRateController:
exchange_rate_interactor = CurrencyExchangeRateInteractorFactory.get()
provider_client_interactor = ProviderClientInteractorFactory.get()
return CurrencyExchangeRateController(exchange_rate_interactor, provider_client_interactor)
As a flexible platform Forex API backend is designed to use several external providers to retrieve and store daily currency exchange rates. The decision of which third party provider to use, among those available, depends on a priority criteria that has each provider in the platform following a fallback strategy managed by a master driver for providers. Therefore, in this layer we also put the code related to drivers for the integration with third party services.
infrastructure/clients/provider/drivers
from http import HTTPStatus
from typing import Any, List
from domain.provider import ProviderEntity
from infrastructure.clients.provider.base import ProviderBaseDriver
from infrastructure.clients.provider.utils import get_available_drivers
from usecases.provider import ProviderInteractor
class ProviderMasterDriver:
ACTIONS = {
'currency_get': 'get_currencies',
'currency_list': 'get_currencies',
'exchange_rate_calculate_twr': 'get_time_series',
'exchange_rate_convert': 'get_exchange_rate',
'exchange_rate_list': 'get_time_series',
}
def __init__(self, provider_interactor: ProviderInteractor):
self.provider_interactor = provider_interactor
self.drivers = get_available_drivers()
@property
def providers(self) -> List[ProviderEntity]:
return self.provider_interactor.get_by_priority()
def _get_driver_class(self, provider: ProviderEntity) -> ProviderBaseDriver:
driver_name = provider.driver
return self.drivers.get(driver_name)
def _get_driver_by_priority(self) -> ProviderBaseDriver:
for provider in self.providers:
driver_class = self._get_driver_class(provider)
yield driver_class(provider)
def fetch_data(self, action: str, **kwargs: dict) -> Any:
error = 'Unable to fetch data from remote server'
for driver in self._get_driver_by_priority():
method = getattr(driver, self.ACTIONS.get(action))
try:
response = method(**kwargs)
except Exception as err:
error = err
else:
if response:
break
else:
response = {
'error': error.message if hasattr(
error, 'message') else str(error),
'status_code': error.code if hasattr(
error, 'code') else HTTPStatus.INTERNAL_SERVER_ERROR.value
}
return response
More on Github:
github.com/sdediego/forex-django-clean-arch..
Summary
Django is designed with a default way of doing things and a Django project itself is a collection of apps which are self-contained packages with all the logic to do just one thing. The objective of the Clean Architecture is the separation of concerns which is achieved by dividing the software into layers, each of them depending on the inner layers thanks to dependency injection. This alternative architecture produces software that is independent of frameworks, more testable, independent of web UI, databases or any external agency.
I hope you enjoyed this post.
Please leave your feedback so that I can improve and share better. Remember to like and share so that others will learn from it.