Code Architecture

The main concept is to assign and separate the responsibilities of each layer.

Resources:

• Clean architecture by Uncle Bob
• Django clean architecture
• Python clean architecture

Why?

• Improve code maintainability
• Add unit testing

• Separate concerns and responsibilities

Layer and component definitions

• Entities
• Use Cases
• Repositories
• Views
• Admin

Entities

Entities encapsulate Enterprise wide business rules. An entity can be an object with methods, or it can be a set of data structures and functions.

@dataclass
class User:
    id: str
    name: str
    email: str

Use Cases

The code in this layer contains application specific business rules. It encapsulates and implements all of the use cases of the system. These use cases orchestrate the flow of data to and from the entities, and direct those entities to use their enterprise wide business rules to achieve the goals of the use case.

# use_cases.py
class CreateUserUseCase:
    def __init__(user_repo):
        self._user_repo = user_repo
    
    def execute(user_dto: UserDTO) -> User:
        # do stuff
        user_repo.create(user_dto)
        # do more stuff

Testing

The aim of testing is to guarantee that the core logic is working as expected. The way use cases are written its important and by doing it this way (OO & DI) the greatest benefit is how clean the tests become.

Imagine a case without DI nor dependency injection:

# use_cases.py
def create_user_use_case(user_dto):
    UsersRepo().get(...)
    ...

# tests_use_cases.py
@patch('users.repositories.UsersRepo')
@patch('users.events.EventManager')
@patch('users.tasks.UserTasks')
@patch('users.services.UsersService')
def test_repo_called():
    ...

Versus:

# tests_use_cases.py
from users.use_cases import CreateUserUseCase

@pytest.mark.unit       # <- NICE TO HAVE in order to later, run only unit or integration
class TestCreateUserUseCase:
    def test_repo_called():
        repo = Mock()
        event_manager = Mock()
        service = Mock()
        tasks = Mock()
        use_case = CreateUserUseCase(repo, event_manager, service, tasks)
        
        # All the mocking could be even done in a different method (setup_method) or fixture
        ...

Services

The aim of the services is keep the code more DRY, so reduce the boilerplate between use cases. A good example could be the permissions:

# services.py
class UserService:
    def has_edit_permission(request_user_id: int, user: User):
        # Check is-self or is admin for example        

Comparison with microservices

Thinking about microservices we have two common cases:

• Read: If you need to retreive something in a MS architecture you probably need data from more than one service. In this case you have two options: duplicate the data or call the two services. If you choose the second one the one in charge of calling the two services would be the use case, working as a gateway.
• Non-safe operations: It's very common, in a MS architecture, that you have side-effects when you create/edit/delete some data. This is managed by the use of events and eventual consistency. Again, the one in charge of triggering those events is the use case.

Events

In order to decouple logic from one app/component to another events should be used. For example, a user creation must create a DownloadLeadRegister. The code for creating a DownloadLeadRegister and computing the reward for the user leading the new registration must be in the component download_registers! Never into the component users that only has to worry about triggering the event. Here's an example using Razemax as library:

# users/events.py
@dataclass
class UserCreatedEvent:
    user: User

# download_registers/subscribers.py
def compute_rewards(event: UserCreatedEvent):
    # run task to compute rewards

# download_registers/apps.py
# ⚠️ Remember to add this on the __init__.py of the app!!!
from django.apps import AppConfig

class DownloadRegistersConfig(AppConfig):
    name = 'download_registers'
    verbose_name = "Download Registers"

    def ready(self):    # Wait for ready to allow django to load apps
        from .subscribers import compute_rewards
        from users.events import UserCreatedEvent
        EventManager.subscribe(compute_rewards, UserCreatedEvent)

# use_cases.py
class CreateUserUseCase:
    ...
    def execute(user_dto) -> User:
        ...
        event_manager.trigger(UserCreatedEvent(user))

Tasks

Tasks should only queue use cases using the desired backend/lib: RQ, celery with RabbitMQ...

Factories

In order to wire the differents components factories are used. Factories contain inicialization of use cases as well as wiring with the different infrastructure components.

Why use factories?

• Give an overview of how all components are connected; such as the navigator for files but for classes.
• Centralize the dependency injections

# factories.py
from .use_cases import CreateUserUseCase
from .repositories import UsersRepo

def build_create_user_use_case() -> CreateUserUseCase:
    return CreateUserUseCase(UsersRepo())   # Also, EventManager(), UserService(), whatever needed to execute the usecase

Repositories

The main responsability of this layer is to include everything related with the access to any data, either a file, database or another MS. In case of multiple data sources a frontal repository should be added to redirect to the real implementations, as a repository gateway. They must return entities.

# repositories.py
class UserRepository:
    def __init__(db_repo, elastic_repo):
        self._db_repo = db_repo
        self._elastic_repo = elastic_repo
        
    def get(self, id):
        return self._db_repo.get(id)
    
    def search(self, query):
        return self._elastic_repo.search(query)

class UserDBRepository:
    def get(self, id):
        user = User.objects.get(id=id)
        # parse user to entity
        return user

class UserElasticRepository:
    def search(self, query):
        users = elasticsearch_dsl.query()
        # parse elastic users to entities
        return users

Testing

Repository tests are usually integration tests in order to test it's connection with the database, file or whatever. BUT, in the case above would be nice to unit test the UserRepository (just like a use case is tested, with a @pytest.mark.unit) and then create the integration test for the others:

# test_respositories.py
@pytest.mark.integration
@pytest.mark.django_db
class TestUserDBRepository:
    def test_get_user():
        # create user to db (using fixtures for example)
        u = UserDBRepository().get(user_id)
        assert u == <your user>

Views

Views are the entrypoint for the API users and it's main responsabilities are:

• Deserialize/Validate: Data provided by API users MUST be validated and deserialized to an object (DTO), no matter if the data cames into the request body or in the query parameters. It's very recommended to use marshmallow to accomplish this. Example:

# dtos.py
@dataclass
class CreateUserRequestDTO:
        username: str
        email: str
        
# validators.py
from marshmallow import Schema, fields, validate, post_load

class CreateUserSerializer(Schema):
    username = fields.String(required=True)
    email = fields.Email(required=True)

    @post_load
    def build_create_user_dto(self, data):
        return CreateUserRequestDTO(**data)

• Build & call Use Case: The view is responsible of calling the use case with the correct parameters and handle its exceptions.

# views.py
from users.factories import build_create_user_use_case
...
use_case = build_create_user_use_case()
try:
    response = use_case.execute(user_dto)
except UserBadDataException as e:
    ...

• Return response + status code: Once the use case has been executed the view MUST return a response. In case that data is retrieved a serializer MUST be defined.

# serializers.py
from marshmallow import Schema, fields, validate, post_load

class UserSerializer(Schema):
    id = fields.String()
    username = fields.String(required=True)
    email = fields.Email(required=True)
    is_active = fields.Boolean()

# ...or use serpy since it's supah fast
import serpy

class UserSerializer(serpy.Serializer):
    id = serpy.IntField()
    username = serpy.StringField()
    email = serpy.StringField()
    is_active = serpy.BooleanField()

In order to understand how to return the different status codes here is a full view example (using the above declared classes):

from .factories import build_create_user_use_case
from .serializers import UserSerializer
from .validators import CreateUserSerializer

class CreateUserView(APIView):
    def post(self, request):
        use_case = build_create_user_use_case()

        user_dto, errors = CreateUserSerializer().load(request.data)
        if errors:
            return Response(errors, status=status.HTTP_400_BAD_REQUEST)

        try:
            user = use_case.execute(user_dto)
        except (UsernameAlreadyExistsError, EmailAlreadyExistsError) as e:
            return Response(str(e), status=status.HTTP_422_UNPROCESSABLE_ENTITY)
        except PermissionsInsuficientException as e:
            return Response(str(e), status=status.HTTP_403_FORBIDDEN)
        else:
            return Response(UserSerializer().dump(user).data, status=status.HTTP_201_CREATED)

Following this pattern it's pretty easy to know what status code return:

• Deserialization errors lead to status 400 BAD REQUEST since the API user is breaking the contract/API docs
• Exceptions when calling the use case create another status 4xx even a 5xx. For example 422 UNPROCESSABLE ENTITY in case request data cannot be processed by the use case

Testing

In this case, even though the test affects different classes it's called unitary since those classes conform a bigger piece. Tests must assure that:

• Validation errors are raised
• Use case is called
• Given use case exceptions an specific status code is returned
• The response matches the JSON Schema provided on the documentation

A small example would be:

# test_views.py

@pytest.mark.unit
class TestCreateUserView:
    @patch('users.views.build_create_user_use_case')
    def test_patch_failed_404(self, use_case_factory):
        use_case = MagicMock()
        use_case_factory.return_value = use_case
        use_case.execute.side_effect = BadContentUserDoesNotExist()
        request = MagicMock()

        response = BadContentUserDetailView().patch(request, user_id=1)

        assert response.status_code == 404

Admin

Django admin should be at the same level than views/controllers so all methods of the admin must be overriden.

@admin.register(UserModel)
class UserAdmin(admin.ModelAdmin):
    list_display = ['username', 'email']
    search_fields = ['=user__username']
    raw_id_fields = ('user',)

    def save_model(self, request, obj, form, change):
        # map obj -> dto
        use_case = build_create_user_use_case()
        use_case.execute(user_dto)

    def delete_model(self, request, obj):
        ...
    
    # for multiple deletions and any actions you'd like to do
    def get_actions(self, request):
        return {
            'delete_selected': (
                self.delete_selected,
                'delete_selected',
                'Delete selected %(verbose_name_plural)s'
            )
        }
    
    def delete_selected(self, admin_model, request, queryset):
        # call the same use case of the delete_model
        for u in queryset:
            self.delete_model(request, obj)
        
        # do something else, like queue a task or call specific use case for batching

Integration API tests

Integration tests are necessary but it's better not to over-use them. An integration test should test that all the wiring between the layers is working fine.

from jairowt_auth.rest_framework import APITestCase

class UsersTestCase(APITestCase):
    def setUp(self):
        self.client = JWTAPIClient()
        self.client.force_authenticate(user_id=1, user_country='ES')
        
    def test_get_user(self, generate_discover_posts):
        # make a request to the API
        response = self.client.get('/v1/users/1')
        
        # check the response
        assert response.status_code == 200
        
        # as test_views have already tested the response format only data checks are need
        assert response.data['id'] == 1