Pure Python Architecture: A Walkthrough. Part 5

REST layer (part 1)

Git tag: Step12

The final stage of our adventure has come in search of pure architecture. We created domain models, serializers, scripts, and storage. But while there is no interface that sticks together all together: it receives call parameters from the user, initializes the script with the repository, executes the script, which gets the domain models from the repository, and converts them into a standard format. This layer can be represented using a variety of interfaces and technologies. For example, using the command line interface (CLI): retrieve parameters using command line keys and return the result as text on the console. But the same basic system can also be used for a web page that receives call parameters from a set of widgets, performs the steps described above, and parses the returned data in JSON format to display the result on the same page.

Regardless of the technology chosen, to interact with the user, collect input data and provide output results, we need to interact with the newly created pure architecture. Therefore, now we will create a layer to bring out the API for working with HTTP. This will be implemented with the help of a server that provides a set of HTTP addresses (API endpoints), when accessing which some data is returned. This layer is usually called the REST layer, because, as a rule, the semantics of addresses is similar to the REST recommendations.

Flask is a lightweight web server with a modular structure that provides only the parts needed by the user. In particular, we will not use any database / ORM, since we already have our own implemented storage layer.

I note that usually this layer together with the storage layer is implemented as a separate package, but in the framework of this lesson I placed them together.

Update the dependency file. The prod.txt file must contain the Flask module.

 Flask 

The dev.txt file contains the Flask-Script extension.

 -r test.txt pip wheel flake8 Sphinx Flask-Script 

And in the test.txt file we test.txt add the pytest extension to work with Flask (more on this later)

 -r prod.txt pytest tox coverage pytest-cov pytest-flask 

After these changes, be sure to run pip install -r requirenments/dev.txt to install the new packages into the virtual environment.

Setting up a Flask application is simple, but includes many features. Since this is not a Flask tutorial, I’ll take a quick look at these steps. However, I will provide links to Flask documentation for each feature.

I usually define individual configurations for my test, development, and production environments. Since the Flask application can be configured using a regular Python object ( documentation ), I create a file rentomatic/settings.py to accommodate these objects.

 import os class Config(object): """Base configuration.""" APP_DIR = os.path.abspath(os.path.dirname(__file__)) # This directory PROJECT_ROOT = os.path.abspath(os.path.join(APP_DIR, os.pardir)) class ProdConfig(Config): """Production configuration.""" ENV = 'prod' DEBUG = False class DevConfig(Config): """Development configuration.""" ENV = 'dev' DEBUG = True class TestConfig(Config): "" " ". "" ENV = 'test' TESTING = True DEBUG = True 

To learn more about Flask configuration options, see this page . Now we need a function that initializes the Flask application ( documentation ), sets it up and registers the drawings ( documentation ). The rentomatic/app.py contains the following code:

 from flask import Flask from rentomatic.rest import storageroom from rentomatic.settings import DevConfig def create_app(config_object=DevConfig): app = Flask(__name__) app.config.from_object(config_object) app.register_blueprint(storageroom.blueprint) return app 

Application endpoints should return a Flask Response object with current results and HTTP status. The content of the response, in this case, is the JSON serialization of the script response.

We'll start writing tests step by step so that you can understand well what will happen at the REST endpoint. The basic structure of the test is as follows.

 [SOME PREPARATION] [CALL THE API ENDPOINT] [CHECK RESPONSE DATA] [CHECK RESPONDSE STATUS CODE] [CHECK RESPONSE MIMETYPE] 

Therefore, our first test - tests/rest/test_get_storagerooms_list.py consists of the following parts

 @mock.patch('rentomatic.use_cases.storageroom_use_cases.StorageRoomListUseCase') def test_get(mock_use_case, client): mock_use_case().execute.return_value = res.ResponseSuccess (storagerooms) 

Since here we are not testing the script itself, we can lock it up. We force the script to return an instance of ResponseSuccess containing a list of domain models (which we have not yet defined).

 def test_get(client, monkeypatch): def monkey_execute(self, rqst): return res.ResponseSuccess(storagerooms) monkeypatch.setattr(StorageRoomListUseCase, 'execute', monkey_execute) 

  http_response = client.get('/storagerooms') 

This is the current API call. We set the end point at /storagerooms . /storagerooms attention to using the client fixture provided by pytest-Flask.

 assert json.loads(http_response.data.decode('UTF-8')) == [storageroom1_dict] assert http_response.status_code == 200 assert http_response.mimetype == 'application/json' 

These are the three checks mentioned above. The second and third are fairly simple, while the first needs an explanation. We want to compare http_response.data with [storageroom1_dict] , which is a list of Python dictionaries containing storageroom1_domain_model object storageroom1_domain_model . Flask Response objects contain a binary representation of the data, so we first decode the bytes using UTF-8, and then convert them to a Python object. It is much more convenient to compare Python objects, since pytest may have problems with dictionaries, due to the lack of order of keys. But if you compare the lines, then there will be no such difficulties.

The final test file with the test of the domain model and its dictionary:

 import json from unittest import mock from rentomatic.domain.storageroom import StorageRoom from rentomatic.shared import response_object as res storageroom1_dict = { 'code': '3251a5bd-86be-428d-8ae9-6e51a8048c33', 'size': 200, 'price': 10, 'longitude': -0.09998975, 'latitude': 51.75436293 } storageroom1_domain_model = StorageRoom.from_dict(storageroom1_dict) storagerooms = [storageroom1_domain_model] @mock.patch('rentomatic.use_cases.storageroom_use_cases.StorageRoomListUseCase') def test_get(mock_use_case, client): mock_use_case().execute.return_value = res.ResponseSuccess(storagerooms) http_response = client.get('/storagerooms') assert json.loads(http_response.data.decode('UTF-8')) == [storageroom1_dict] assert http_response.status_code == 200 assert http_response.mimetype == 'application/json' 

It's time to write the end point, where we finally see the work of all parts of the architecture.

The minimum Flask endpoint can be placed at rentomatic/rest/storageroom.py

 blueprint = Blueprint('storageroom', __name__) @blueprint.route('/storagerooms', methods=['GET']) def storageroom(): [LOGIC] return Response([JSON DATA], mimetype='application/json', status=[STATUS]) 

The first thing we create is a StorageRoomListRequestObject . At this time, you can ignore the optional query string parameters and use an empty dictionary.

 def storageroom(): request_object = ro.StorageRoomListRequestObject. from_dict ({}) 

As you can see, I create an object from an empty dictionary, so query string parameters are not taken into account. The second thing to do is initialize the repository.

  repo = mr.MemRepo () 

The third is the initialization of the script endpoint

 use_case = uc.StorageRoomListUseCase(repo) 

Finally, we execute the script by passing the request object

 response = use_case.execute(request_object) 

But this response is not yet an HTTP response. We must explicitly build it. The HTTP response will contain a JSON representation of the response.value attribute.

 return Response(json.dumps(response.value, cls=ser.StorageRoomEncoder), mimetype='application/json', status=200) 

Note that this function is not yet complete, as it always returns a successful response (code 200). But this is enough to pass the written tests. The whole file looks like this:

 import json from flask import Blueprint, Response from rentomatic.use_cases import request_objects as req from rentomatic.repository import memrepo as mr from rentomatic.use_cases import storageroom_use_cases as uc from rentomatic.serializers import storageroom_serializer as ser blueprint = Blueprint('storageroom', __name__) @blueprint.route('/storagerooms', methods=['GET']) def storageroom(): request_object = req.StorageRoomListRequestObject.from_dict({}) repo = mr.MemRepo() use_case = uc.StorageRoomListUseCase(repo) response = use_case.execute(request_object) return Response(json.dumps(response.value, cls=ser.StorageRoomEncoder), mimetype='application/json', status=200) 

This code demonstrates how clean architecture works. True, the written function is not yet complete, since it does not take into account the parameters of string queries and cases with errors.

Server in action

Git tag: Step13

Before adding the missing parts of the end point, let's look at the server in work and see our project in action.

In order to see the results when accessing the endpoint, we must fill the repository with test data. Obviously, we have to do this because of the inconstancy of the storage we use. The actual storage will wrap the persistent data source, and this test data will no longer be needed. We define them to initialize the repository:

 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', } 

And we transfer them to our storage.

 repo = mr.MemRepo ([storageroom1, storageroom2, storageroom3]) 

Now we can run Flask through the manage.py file and check the published URLs:

 $ python manage.py urls Rule Endpoint ------------------------------------------------ /static/<path:filename> static /storagerooms storageroom.storageroom 

And start the development server

 $ python manage.py server 

If you open a browser and go to http: //127.0.0.1.1000000/storagerooms , you will see the result of the call API. I recommend installing the browser formatting extension so that the answer is readable. If you're using Chrome, try the JSON Formatter .

REST layer (part 2)

Git tag: Step14

Let's look at two unimplemented cases at the end point. First, I will enter a test that checks the correctness of processing the parameters of the query string by the end point

 @mock.patch('rentomatic.use_cases.storageroom_use_cases.StorageRoomListUseCase') def test_get_failed_response(mock_use_case, client): mock_use_case().execute.return_value = res.ResponseFailure.build_system_error('test message') http_response = client.get('/storagerooms') assert json.loads(http_response.data.decode('UTF-8')) == {'type': 'SYSTEM_ERROR', 'message': 'test message'} assert http_response.status_code == 500 assert http_response.mimetype == 'application/json' 

Now we check that the script returns an error response, and we also see that the HTTP response contains an error code. To pass the test, we must correlate the domain response codes with HTTP response codes.

 from rentomatic.shared import response_object as res STATUS_CODES = { res.ResponseSuccess.SUCCESS: 200, res.ResponseFailure.RESOURCE_ERROR: 404, res.ResponseFailure.PARAMETERS_ERROR: 400, res.ResponseFailure.SYSTEM_ERROR: 500 } 

Then we need to create a Flask response with the correct code.

  return Response(json.dumps(response.value, cls=ser.StorageRoomEncoder), mimetype='application/json', status=STATUS_CODES[response.type]) 

The second and last test is a little more difficult. As before, we lock the script, but this time we also patch StorageRoomListRequestObject . We need to know that the request object is initialized with the correct parameters from the command line. So, we move step by step:

 @mock.patch('rentomatic.use_cases.storageroom_use_cases.StorageRoomListUseCase') def test_request_object_initialisation_and_use_with_filters(mock_use_case, client): mock_use_case().execute.return_value = res.ResponseSuccess([]) 

Here, as before, the patch of the script class ensures that the instance of the ResponseSuccess object is returned by precedent.

  internal_request_object = mock.Mock() 

The request object will be created inside the StorageRoomListRequestObject.from_dict , and we want the function to return the initialized here mock object.

 request_object_class = 'rentomatic.use_cases.request_objects.StorageRoomListRequestObject' with mock.patch(request_object_class) as mock_request_object: mock_request_object.from_dict.return_value = internal_request_object client.get('/storagerooms?filter_param1=value1&filter_param2=value2') 

We patch StorageRoomListRequestObject and assign a pre-known result to the from_dict() method. Then we access the endpoint with some parameters of the query string. The following happens: the from_dict() method of the request is called with filter parameters, and the execute() method of the script instance is called with internal_request_object .

 mock_request_object.from_dict.assert_called_with( {'filters': {'param1': 'value1', 'param2': 'value2'}} ) mock_use_case().execute.assert_called_with(internal_request_object) 

The endpoint function must be changed to reflect this new behavior and make the test valid. The final code of the new Flask-method storageroom() as follows

 import json from flask import Blueprint, request, Response from rentomatic.use_cases import request_objects as req from rentomatic.shared import response_object as res from rentomatic.repository import memrepo as mr from rentomatic.use_cases import storageroom_use_cases as uc from rentomatic.serializers import storageroom_serializer as ser blueprint = Blueprint('storageroom', __name__) STATUS_CODES = { res.ResponseSuccess.SUCCESS: 200, res.ResponseFailure.RESOURCE_ERROR: 404, res.ResponseFailure.PARAMETERS_ERROR: 400, res.ResponseFailure.SYSTEM_ERROR: 500 } 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', } @blueprint.route('/storagerooms', methods=['GET']) def storageroom(): qrystr_params = { 'filters': {}, } for arg, values in request.args.items(): if arg.startswith('filter_'): qrystr_params['filters'][arg.replace('filter_', '')] = values request_object = req.StorageRoomListRequestObject.from_dict(qrystr_params) repo = mr.MemRepo([storageroom1, storageroom2, storageroom3]) use_case = uc.StorageRoomListUseCase(repo) response = use_case.execute(request_object) return Response(json.dumps(response.value, cls=ser.StorageRoomEncoder), mimetype='application/json', status=STATUS_CODES[response.type]) 

Note that we retrieve the parameters from the query string of the global request Flask object. After the query string parameters appear in the dictionary, you only need to create a query object from it.

Conclusion

Well that's all! Some tests in the REST layer are missing, but, as I said, this is only a working implementation to demonstrate pure architecture, and not a fully developed project. I think you should try to add some changes yourself, such as:

• another endpoint, for example, access to a single resource ( /storagerooms/<code> )
• real storage connected to a real database (alternatively, SQLite )
• implementation of a new query string parameter, for example, the distance from a given point on the map (I recommend using geopy to simply calculate the distance)

While developing your code, always try to follow the TDD approach. Testability is one of the main features of a clean architecture, it is very important to write tests, do not ignore them.

Regardless of whether you decide to use pure architecture or not, I hope that this post will help you get a fresh look at software architecture, as happened to me when I first learned about the concepts presented here.