Pure Python Architecture: A Walkthrough. Part 4

Scripts (Part 3)

Git tag: Step09

Our implementation of responses and requests is finally complete. And now we can implement the latest version of our script . The script correctly returns the ResponseSuccess object, but still does not check the validity of the incoming request.

Let's change the test in the tests/use_cases/test_storageroom_list_use_case.py and add 2 more tests. The resulting test suite (after the domain_storagerooms ) looks like this:

 import pytest from unittest import mock from rentomatic.domain.storageroom import StorageRoom from rentomatic.shared import response_object as res from rentomatic.use_cases import request_objects as req from rentomatic.use_cases import storageroom_use_cases as uc @pytest.fixture def domain_storagerooms(): […] def test_storageroom_list_without_parameters(domain_storagerooms): repo = mock.Mock() repo.list.return_value = domain_storagerooms storageroom_list_use_case = uc.StorageRoomListUseCase(repo) request_object = req.StorageRoomListRequestObject.from_dict({}) response_object = storageroom_list_use_case.execute(request_object) assert bool(response_object) is True repo.list.assert_called_with(filters=None) assert response_object.value == domain_storagerooms 

This test differs from the previous one in that now the assert_called_with() method is called with the filters=None parameter. In the lines with the import, response_objects and request_objects . The fixture Domain_storagerooms was excluded from the code for brevity.

 def test_storageroom_list_with_filters(domain_storagerooms): repo = mock.Mock() repo.list.return_value = domain_storagerooms storageroom_list_use_case = uc.StorageRoomListUseCase(repo) qry_filters = {'a': 5} request_object = req.StorageRoomListRequestObject.from_dict({'filters': qry_filters}) response_object = storageroom_list_use_case.execute(request_object) assert bool(response_object) is True repo.list.assert_called_with(filters=qry_filters) assert response_object.value == domain_storagerooms 

This test verifies that when calling the repository , the filters key value is used, which is also used to create queries .

 def test_storageroom_list_handles_generic_error(): repo = mock.Mock() repo.list.side_effect = Exception('Just an error message') storageroom_list_use_case = uc.StorageRoomListUseCase(repo) request_object = req.StorageRoomListRequestObject.from_dict({}) response_object = storageroom_list_use_case.execute(request_object) assert bool(response_object) is False assert response_object.value == { 'type': res.ResponseFailure.SYSTEM_ERROR, 'message': "Exception: Just an error message" } def test_storageroom_list_handles_bad_request(): repo = mock.Mock() storageroom_list_use_case = uc.StorageRoomListUseCase(repo) request_object = req.StorageRoomListRequestObject.from_dict({'filters': 5}) response_object = storageroom_list_use_case.execute(request_object) assert bool(response_object) is False assert response_object.value == { 'type': res.ResponseFailure.PARAMETERS_ERROR, 'message': "filters: Is not iterable" } 

These last two tests check the behavior of the script when an exception occurs in the repository or if an incorrect request is made .

rentomatic/use_cases/storageroom_use_cases.py so that it contains a new script implementation that allows tests to pass successfully.

 from rentomatic.shared import response_object as res class StorageRoomListUseCase(object): def __init__(self, repo): self.repo = repo def execute(self, request_object): if not request_object: return res.ResponseFailure.build_from_invalid_request_object(request_object) try: storage_rooms = self.repo.list(filters=request_object.filters) return res.ResponseSuccess(storage_rooms) except Exception as exc: return res.ResponseFailure.build_system_error( "{}: {}".format(exc.__class__.__name__, "{}".format(exc))) 

As you can see, the execute() method checks the correctness of the request; otherwise, it returns a ResponseFailure created based on the same request object . Now the business logic is implemented, calling the repository and returning a successful response. If something goes wrong in this phase, an exception is caught and the ResponseFailure that is formed as necessary is returned.

Intermission: refactoring

Git tag: Step10

Pure architecture is not a framework. You should not expect as much from it as from products such as Django, which provides models, ORM, various structures and libraries. However, some classes can be separated from our code and presented as a library, so that this code can be reused. In this section, I will introduce you to the refactoring of already existing code, with which we isolate the main features for requests, responses and scripts .

We have already identified the object of the response . Since test_valid_request_object_cannot_be_used tests the overall behavior, not the relationship between the StorageRoom model and the scripts , we can move it from tests/use_cases/test_storageroom_list_request_objects.py to tests/shared/test_response_object.py .

Then we can move the InvalidRequestObject and ValidRequestObject from rentomatic/use_cases/request_objects.py to rentomatic/shared/request_object.py , making the necessary changes to the StorageRoomListRequestObject class, which is now inherited from the external class.

But the script class is undergoing significant changes. The UseCase class is tested with the file code tests/shared/test_use_case.py :

 from unittest import mock from rentomatic.shared import request_object as req, response_object as res from rentomatic.shared import use_case as uc def test_use_case_cannot_process_valid_requests(): valid_request_object = mock.MagicMock() valid_request_object.__bool__.return_value = True use_case = uc.UseCase() response = use_case.execute(valid_request_object) assert not response assert response.type == res.ResponseFailure.SYSTEM_ERROR assert response.message == \ 'NotImplementedError: process_request() not implemented by UseCase class' 

This test verifies that the UseCase class cannot be used to process incoming requests.

 def test_use_case_can_process_invalid_requests_and_returns_response_failure(): invalid_request_object = req.InvalidRequestObject() invalid_request_object.add_error('someparam', 'somemessage') use_case = uc.UseCase() response = use_case.execute(invalid_request_object) assert not response assert response.type == res.ResponseFailure.PARAMETERS_ERROR assert response.message == 'someparam: somemessage' 

The test executes the script with an incorrect request and checks the response. Since the request is incorrect, the type of response is PARAMETERS_ERROR , thus speaking about the presence of a problem in the request parameters.

 def test_use_case_can_manage_generic_exception_from_process_request(): use_case = uc.UseCase() class TestException(Exception): pass use_case.process_request = mock.Mock() use_case.process_request.side_effect = TestException('somemessage') response = use_case.execute(mock.Mock) assert not response assert response.type == res.ResponseFailure.SYSTEM_ERROR assert response.message == 'TestException: somemessage' 

This test causes the script to raise an exception . This type of error is classified as SYSTEM_ERROR , which is the general name for an exception that is not related to the request parameters or the actual entities.

As you can see from the last test, the idea is to provide the execute() method of the UseCase class and call the process_request() method, which is defined in each child class that is our script.

The rentomatic/shared/use_case.py contains the following code for passing the test successfully:

 from rentomatic.shared import response_object as res class UseCase(object): def execute(self, request_object): if not request_object: return res.ResponseFailure.build_from_invalid_request_object(request_object) try: return self.process_request(request_object) except Exception as exc: return res.ResponseFailure.build_system_error( "{}: {}".format(exc.__class__.__name__, "{}".format(exc))) def process_request(self, request_object): raise NotImplementedError( "process_request() not implemented by UseCase class") 

And now the rentomatic/use_cases/storageroom_use_cases.py contains the following code:

 from rentomatic.shared import use_case as uc from rentomatic.shared import response_object as res class StorageRoomListUseCase(uc.UseCase): def __init__(self, repo): self.repo = repo def process_request(self, request_object): domain_storageroom = self.repo.list(filters=request_object.filters) return res.ResponseSuccess(domain_storageroom) 

Storage layer

Git tag: Step11

The storage layer is the layer in which there is data storage. As you saw, when we implemented the script , we accessed the data store through the API, in this case through the list() store method. The level of abstraction provided by the storage level is higher than the level provided by ORM or a tool such as SQLAlchemy. The storage layer provides only application - specific endpoints with an interface that is tailored to specific business objectives and application goals.

For clarity, speaking in terms of specific technologies, SQLAlchemy is an excellent tool for abstract access to an SQL database. The internal implementation of the storage layer can use it to access the PostgreSQL database. But the outer layer API is not the same as SQLAlchemy. An API is a (usually diminished) set of functions that call scripts to get data. In fact, the internal implementation of the API can use raw SQL queries. The storage may not even be database based. We can have a storage layer that retrieves data from a REST service or makes remote procedure calls via RabbitMQ.

A very important feature of the storage layer is that it always returns domain models (as well as ORM frameworks).

In this article, I will not deploy a real database. Perhaps I will write this in other articles, in which enough space will be allocated for implementing two different solutions, and in which it will turn out to show how the repository API can hide the actual implementation.

Instead, I'm going to create a very simple in-memory storage system with some predefined data. I think this will be enough to demonstrate the concept of storage.

The first thing to do is to write a few tests that describe the public repository API. The file tests/repository/test_memrepo.py contains tests for this task.

First we will add the data that we will use in the tests. We import the domain model to check the correct type of the result of the API call:

 import pytest from rentomatic.shared.domain_model import DomainModel from rentomatic.repository import memrepo storageroom1 = { 'code': 'f853578c-fc0f-4e65-81b8-566c5dffa35a', 'size': 215, 'price': 39, 'longitude': '-0.09998975', 'latitude': '51.75436293', } storageroom2 = { 'code': 'fe2c3195-aeff-487a-a08f-e0bdc0ec6e9a', 'size': 405, 'price': 66, 'longitude': '0.18228006', 'latitude': '51.74640997', } storageroom3 = { 'code': '913694c6-435a-4366-ba0d-da5334a611b2', 'size': 56, 'price': 60, 'longitude': '0.27891577', 'latitude': '51.45994069', } storageroom4 = { 'code': 'eed76e77-55c1-41ce-985d-ca49bf6c0585', 'size': 93, 'price': 48, 'longitude': '0.33894476', 'latitude': '51.39916678', } @pytest.fixture def storagerooms(): return [storageroom1, storageroom2, storageroom3, storageroom4] 

Since the repository object will return domain models , we need a helper function to validate the results. The following function checks the length of two lists and ensures that all returned model elements are from the domain, and compares the codes. Note that we can use the isinstance() built-in function, since DomainModel is an abstract base class and our models are registered (see rentomatic/domian/storagerooms.py )

 def _check_results(domain_models_list, data_list): assert len(domain_models_list) == len(data_list) assert all([isinstance(dm, DomainModel) for dm in domain_models_list]) assert set([dm.code for dm in domain_models_list]) == set([d['code'] for d in data_list]) 

We need to be able to initialize the repository with a list of dictionaries, and so that calling the list() method without any parameter returns the same list of entries.

 def test_repository_list_without_parameters(storagerooms): repo = memrepo.MemRepo(storagerooms) assert repo.list() == storagerooms 

The list() method must accept the filters parameter, which is a dictionary. Dictionary keys should be in the form of <attribute>__<operator> , similar to the syntax used in Django ORM. Thus, to express that the price should be less than 65, you can write filters={'price__lt': 60} .

There are several error conditions that need to be checked: an unknown key should raise a KeyError exception, and using an invalid operator should cause a ValueError exception.

 def test_repository_list_with_filters_unknown_key(storagerooms): repo = memrepo.MemRepo(storagerooms) with pytest.raises(KeyError): repo.list(filters={'name': 'aname'}) def test_repository_list_with_filters_unknown_operator(storagerooms): repo = memrepo.MemRepo(storagerooms) with pytest.raises(ValueError): repo.list(filters={'price__in': [20, 30]}) 

Let's check that the filtering mechanism works. We want the default operator to be __eq . This means that if we do not pass any operator, the equality operator must be executed.

 def test_repository_list_with_filters_price(storagerooms): repo = memrepo.MemRepo(storagerooms) _check_results(repo.list(filters={'price': 60}), [storageroom3]) _check_results(repo.list(filters={'price__eq': 60}), [storageroom3]) _check_results(repo.list(filters={'price__lt': 60}), [storageroom1, storageroom4]) _check_results(repo.list(filters={'price__gt': 60}), [storageroom2]) def test_repository_list_with_filters_size(storagerooms): repo = memrepo.MemRepo(storagerooms) _check_results(repo.list(filters={'size': 93}), [storageroom4]) _check_results(repo.list(filters={'size__eq': 93}), [storageroom4]) _check_results(repo.list(filters={'size__lt': 60}), [storageroom3]) _check_results(repo.list(filters={'size__gt': 400}), [storageroom2]) def test_repository_list_with_filters_code(storagerooms): repo = memrepo.MemRepo(storagerooms) _check_results( repo.list(filters={'code': '913694c6-435a-4366-ba0d-da5334a611b2'}), [storageroom3]) 

The implementation of the MemRepo class MemRepo fairly simple, and I will not explain it line by line.

 from rentomatic.domain import storageroom as sr class MemRepo: def __init__(self, entries=None): self._entries = [] if entries: self._entries.extend(entries) def _check(self, element, key, value): if '__' not in key: key = key + '__eq' key, operator = key.split('__') if operator not in ['eq', 'lt', 'gt']: raise ValueError('Operator {} is not supported'.format(operator)) operator = '__{}__'.format(operator) return getattr(element[key], operator)(value) def list(self, filters=None): if not filters: return self._entries result = [] result.extend(self._entries) for key, value in filters.items(): result = [e for e in result if self._check(e, key, value)] return [sr.StorageRoom.from_dict(r) for r in result]