Pure Python Architecture: A Walkthrough. Part 3

Scripts (Part 2)

Git tag: Step06

Now that we have implemented the request and response objects, we add them. We tests/use_cases/test_storageroom_list_use_case.py following code in the tests/use_cases/test_storageroom_list_use_case.py file:

 import pytest from unittest import mock from rentomatic.domain.storageroom import StorageRoom from rentomatic.use_cases import request_objects as ro from rentomatic.use_cases import storageroom_use_cases as uc @pytest.fixture def domain_storagerooms(): storageroom_1 = StorageRoom( code='f853578c-fc0f-4e65-81b8-566c5dffa35a', size=215, price=39, longitude='-0.09998975', latitude='51.75436293', ) storageroom_2 = StorageRoom( code='fe2c3195-aeff-487a-a08f-e0bdc0ec6e9a', size=405, price=66, longitude='0.18228006', latitude='51.74640997', ) storageroom_3 = StorageRoom( code='913694c6-435a-4366-ba0d-da5334a611b2', size=56, price=60, longitude='0.27891577', latitude='51.45994069', ) storageroom_4 = StorageRoom( code='eed76e77-55c1-41ce-985d-ca49bf6c0585', size=93, price=48, longitude='0.33894476', latitude='51.39916678', ) return [storageroom_1, storageroom_2, storageroom_3, storageroom_4] 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 = ro.StorageRoomListRequestObject.from_dict({}) response_object = storageroom_list_use_case.execute(request_object) assert bool(response_object) is True repo.list.assert_called_with() assert response_object.value == domain_storagerooms 

The new version of the rentomatic/use_case/storageroom_use_cases.py now looks like this:

 from rentomatic.shared import response_object as ro class StorageRoomListUseCase(object): def __init__(self, repo): self.repo = repo def execute(self, request_object): storage_rooms = self.repo.list() return ro.ResponseSuccess (storage_rooms) 

Let's see what we do with this pure architecture. We have a very lightweight model, serialized into JSON and completely independent from other parts of the system. The code also contains a script in which the storage provided by this API retrieves all models and returns them inside a structured object.

True, we still have not implemented everything. For example, there is no negative response or validated incoming request object .

Let's try to correct these omissions by changing the current script to accept the filters parameter, which represents the filters applied to the model list retrieval. When passing this parameter, errors may occur, so we have to implement a check for the incoming request object .

Requests and validation

Git tag: Step07

I want to add a filters parameter for the query. This option will allow the caller to add different filters, specifying a name and a value for each (for example, { 'price_lt': 100} to get all the results with a price less than 100).

The first where we start our edits is a test. The new version of the file tests/use_cases/test_storageroom_list_request_objects.py looks like this:

 import pytest from rentomatic.use_cases import request_objects as ro def test_valid_request_object_cannot_be_used(): with pytest.raises(NotImplementedError): ro.ValidRequestObject.from_dict({}) def test_build_storageroom_list_request_object_without_parameters(): req = ro.StorageRoomListRequestObject() assert req.filters is None assert bool(req) is True def test_build_file_list_request_object_from_empty_dict(): req = ro.StorageRoomListRequestObject.from_dict({}) assert req.filters is None assert bool(req) is True def test_build_storageroom_list_request_object_with_empty_filters(): req = ro.StorageRoomListRequestObject(filters={}) assert req.filters == {} assert bool(req) is True def test_build_storageroom_list_request_object_from_dict_with_empty_filters(): req = ro.StorageRoomListRequestObject.from_dict({'filters': {}}) assert req.filters == {} assert bool(req) is True def test_build_storageroom_list_request_object_with_filters(): req = ro.StorageRoomListRequestObject(filters={'a': 1, 'b': 2}) assert req.filters == {'a': 1, 'b': 2} assert bool(req) is True def test_build_storageroom_list_request_object_from_dict_with_filters(): req = ro.StorageRoomListRequestObject.from_dict({'filters': {'a': 1, 'b': 2}}) assert req.filters == {'a': 1, 'b': 2} assert bool(req) is True def test_build_storageroom_list_request_object_from_dict_with_invalid_filters(): req = ro.StorageRoomListRequestObject.from_dict({'filters': 5}) assert req.has_errors() assert req.errors[0]['parameter'] == 'filters' assert bool(req) is False 

Check assert req.filters is None for the initial two tests, and then add 5 more tests to check if filters can be refined and to check the behavior of an object with an invalid filter parameter.

In order for the tests to pass, you must change our StorageRoomListRequestObject class. Naturally, there are many possible solutions that you can come up with, and I recommend that you try to find your own. It also describes the solution that I usually use myself. The rentomatic/use_cases/request_object.py file now looks like

 import collections class InvalidRequestObject(object): def __init__(self): self.errors = [] def add_error(self, parameter, message): self.errors.append({'parameter': parameter, 'message': message}) def has_errors(self): return len(self.errors) > 0 def __nonzero__(self): return False __bool__ = __nonzero__ class ValidRequestObject(object): @classmethod def from_dict(cls, adict): raise NotImplementedError def __nonzero__(self): return True __bool__ = __nonzero__ class StorageRoomListRequestObject(ValidRequestObject): def __init__(self, filters=None): self.filters = filters @classmethod def from_dict(cls, adict): invalid_req = InvalidRequestObject() if 'filters' in adict and not isinstance(adict['filters'], collections.Mapping): invalid_req.add_error('filters', 'Is not iterable') if invalid_req.has_errors(): return invalid_req return StorageRoomListRequestObject(filters=adict.get('filters', None)) 

Let me explain these code changes.

First, two helper objects were introduced, ValidRequestObject and InvalidRequestObject . They differ from each other because an invalid request must contain validation errors, but both must be converted to a boolean value.

Secondly, at the time of creation, the StorageRoomListRequestObject takes an optional filters parameter. In the __init __ () method, there are no checks for validity, as it is considered an internal method, which is called after the parameters have already been confirmed.

As a result, the from_dict() method checks the presence of the filters parameter. I use the abstract collections.Mapping class to check that the input parameters are dictionaries, and that an instance of an InvalidRequestObject or ValidRequestObject .

Since we can now report the presence of correct or incorrect requests , we need to introduce a new type of response to manage responses to incorrect requests or errors in the script.

Answers and failures

Git tag: Step08

What happens if an error occurs in the script? A large number of errors can occur in them: not only validation errors, which we talked about in the previous section, but also business errors or errors from the storage layer. Whatever the error, the script should always return an object with a known structure (the answer), so we need a new object that provides good support for various types of dips.

As with queries, there is no one correct way to represent such an object, and the following code is just one of the possible solutions.

The first thing to do is expand the tests/shared/test_response_object.py , adding tests for failures.

 import pytest from rentomatic.shared import response_object as res from rentomatic.use_cases import request_objects as req @pytest.fixture def response_value(): return {'key': ['value1', 'value2']} @pytest.fixture def response_type(): return 'ResponseError' @pytest.fixture def response_message(): return 'This is a response error' 

This is template code based on pytest , which we will use in the following tests.

 def test_response_success_is_true(response_value): assert bool(res.ResponseSuccess(response_value)) is True def test_response_failure_is_false(response_type, response_message): assert bool(res.ResponseFailure(response_type, response_message)) is False 

Two basic tests to verify that the old ResponseSuccess and the new ResponseFailure behave consistently when converted to a boolean value.

 def test_response_success_contains_value(response_value): response = res.ResponseSuccess(response_value) assert response.value == response_value 

The ResponseSuccess object contains the result of the call in the value attribute.

 def test_response_failure_has_type_and_message(response_type, response_message): response = res.ResponseFailure(response_type, response_message) assert response.type == response_type assert response.message == response_message def test_response_failure_contains_value(response_type, response_message): response = res.ResponseFailure(response_type, response_message) assert response.value == {'type': response_type, 'message': response_message} 

These two tests ensure that the ResponseFailure object provides the same interface as on success, and that this object has type and message parameters.

 def test_response_failure_initialization_with_exception(): response = res.ResponseFailure(response_type, Exception('Just an error message')) assert bool(response) is False assert response.type == response_type assert response.message == "Exception: Just an error message" def test_response_failure_from_invalid_request_object(): response = res.ResponseFailure.build_from_invalid_request_object(req.InvalidRequestObject()) assert bool(response) is False def test_response_failure_from_invalid_request_object_with_errors(): request_object = req.InvalidRequestObject() request_object.add_error('path', 'Is mandatory') request_object.add_error('path', "can't be blank") response = res.ResponseFailure.build_from_invalid_request_object(request_object) assert bool(response) is False assert response.type == res.ResponseFailure.PARAMETERS_ERROR assert response.message == "path: Is mandatory\npath: can't be blank" 

Sometimes it is necessary to create responses from Python exceptions that can occur in scripts , so we check that ResponseFailure objects can be initialized with an exception.

And build_from_invalid_request_object() , we have tests for the build_from_invalid_request_object() method that automates the initialization of the response from a non-invalid request. If the request contains errors (remember, the request checks itself), we must send them in the response message.

The last test uses the class attribute to classify the error. The ResponseFailure class will contain three predefined errors that can occur during the execution of the script: RESOURCE_ERROR , PARAMETERS_ERROR and SYSTEM_ERROR . With a similar separation, we try to cover various types of errors that can occur when working with an external system through an API. RESOURCE_ERROR contains errors related to the resources contained in the repository, for example, when you cannot find a record by its unique identifier. PARAMETERS_ERROR describes errors that occur with incorrect or missing query parameters. SYSTEM_ERROR covers errors that occur in the base system at the operating system level, such as a file system failure or a network connection error while fetching data from the database.

The script is responsible for interacting with various errors that occur in the Python code, and converts them into one of the three types of messages just described that have a description of this error.

Let's write the ResponseFailure class, which allows tests to run successfully. Create it in rentomatic/shared/response_object.py

 class ResponseFailure(object): RESOURCE_ERROR = 'RESOURCE_ERROR' PARAMETERS_ERROR = 'PARAMETERS_ERROR' SYSTEM_ERROR = 'SYSTEM_ERROR' def __init__(self, type_, message): self.type = type_ self.message = self._format_message(message) def _format_message(self, msg): if isinstance(msg, Exception): return "{}: {}".format(msg.__class__.__name__, "{}".format(msg)) return msg 

Using the _format_message() method, we allow the class to accept both a string message and a Python exception, which is very convenient when working with external libraries that can cause unknown or uninteresting exceptions.

 @property def value(self): return {'type': self.type, 'message': self.message} 

This property makes the class consistent with the API ResponseSuccess , providing the value attribute, which is a dictionary.

 def __bool__(self): return False @classmethod def build_from_invalid_request_object(cls, invalid_request_object): message = "\n".join(["{}: {}".format(err['parameter'], err['message']) for err in invalid_request_object.errors]) return cls(cls.PARAMETERS_ERROR, message) 

As explained above, the PARAMETERS_ERROR type covers all those errors that occur when an incorrect set of transmitted parameters, that is, some parameters contain errors or are missing.

Since we often have to create error responses, it is helpful to have helper methods. Add three tests for building functions in the file tests/shared/test_response_object.py

 def test_response_failure_build_resource_error(): response = res.ResponseFailure.build_resource_error("test message") assert bool(response) is False assert response.type == res.ResponseFailure.RESOURCE_ERROR assert response.message == "test message" def test_response_failure_build_parameters_error(): response = res.ResponseFailure.build_parameters_error("test message") assert bool(response) is False assert response.type == res.ResponseFailure.PARAMETERS_ERROR assert response.message == "test message" def test_response_failure_build_system_error(): response = res.ResponseFailure.build_system_error("test message") assert bool(response) is False assert response.type == res.ResponseFailure.SYSTEM_ERROR assert response.message == "test message" 

We added the corresponding methods in the class and added the use of the new build_parameters_error() method in the build_from_invalid_request_object() method. The rentomatic/shared/response_object.py file should now contain this code.

  @classmethod def build_resource_error(cls, message=None): return cls(cls.RESOURCE_ERROR, message) @classmethod def build_system_error(cls, message=None): return cls(cls.SYSTEM_ERROR, message) @classmethod def build_parameters_error(cls, message=None): return cls(cls.PARAMETERS_ERROR, message) @classmethod def build_from_invalid_request_object(cls, invalid_request_object): message = "\n".join(["{}: {}".format(err['parameter'], err['message']) for err in invalid_request_object.errors]) return cls.build_parameters_error(message) 

To be continued in Part 4 .