Skip to content

23. Practice Exercise: Version 6

23.1. Introduction

We now return to our tax calculation application. We will build various web applications around it.

In version 5 of our application exercise, the tax authority’s data was stored in a database. This version 5 consisted of two separate applications that shared common layers:

  • an application that calculated taxes in |batch| mode for taxpayers stored in a text file;
  • an application that calculated taxes in |interactive| mode for taxpayers whose information was entered via the keyboard;

Version 5 of the batch tax calculation application had the following architecture:

Image

Ultimately, the web version of this application will have the following architecture:

Image

  • the web client [1] communicates with the web server [2], which communicates with the DBMS [3];
  • the web server [2] retains the [business] [8] and [DAO] [9] layers of the original application;
  • The original application retains its main script [4] and its [business] layer [15]. The [business] layers [8] and [15] are identical;
  • client/server communication requires two additional layers:
    • the [web] layer [7], which implements the web application;
    • the [DAO] layer [5], which acts as a client for the web application [7];

In the final version, batch tax calculation can be performed in two ways:

  • the business logic for tax calculation is handled by the server’s [business] layer. The [main] script will use this method;
  • the business logic for tax calculation is handled by the client’s [business] layer. The [main2] script will use this method;

From now on, we will develop several client/server applications of the type described above, each illustrating one or more new web development technologies.

23.2. The tax calculation web server

23.2.1. Version 1

Image

The [server_01] script is the following web application:

Image

  • In [1], we use a parameterized URL in which we pass three values:
    • [married] (yes/no) to indicate whether the taxpayer is married;
    • [children]: the number of children the taxpayer has;
    • [salary]: the taxpayer’s annual salary;
  • In [2], the web server returns a JSON string that provides the amount of tax due along with its various components;

The application architecture is as follows:

Image

  • the browser [1] queries the server [2]. The script [server_01] implements the [web] layer [2] of the server;
  • layers [3-8] are those already used in |version 5| of the tax calculation application. We reuse them as-is;
    • The [business] layer [3] is defined |here|;
    • the [DAO] layer [4] is defined |here|;

The [server_01] web application is configured using three scripts:

  • [config], which configures the entire application;
  • [config_database], which configures database access. We will be working with the MySQL and PostgreSQL DBMS;
  • [config_layers], which configures the application layers;

The [config] script is as follows:


def configure(config: dict) -> dict:
    import os

    # step 1 ------
    # directory of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))
    # root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    # absolute dependencies
    absolute_dependencies = [
        # project folders
        # BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        # TaxDaoInterface, TaxBusinessInterface, TaxUiInterface
        f"{root_dir}/taxes/v04/interfaces",
        # AbstractTaxDao, TaxConsole, TaxBusiness
        f"{root_dir}/taxes/v04/services",
        # TaxDaoWithAdminDataInDatabase
        f"{root_dir}/taxes/v05/services",
        # AdminData, TaxError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        # Constants, tax brackets
        f"{root_dir}/taxes/v05/entities",
        # IndexController
        f"{script_dir}/../controllers",
        # scripts [database_config, layers_config]
        script_dir,
    ]
    # Set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # step 2 ------
    # application configuration
    # list of users authorized to use the application
    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

    # Step 3 ------
    # database configuration
    import config_database
    config["database"] = config_database.configure(config)

    # Step 4 ------
    # instantiating the application layers
    import config_layers
    config['layers'] = config_layers.configure(config)

    # we return the configuration
    return config
  • The [configure] function takes a [config] dictionary as an argument (line 1) and returns it as the result (line 54) after enriching its contents. It could have been pointed out long ago that it was not necessary to return the result [config]. Indeed, [config] is a dictionary reference that the calling code shares with the called code. The calling code therefore already possesses this reference (line 1), and there is no need to return it again (line 54). Thus, writing: 

config=[module].configure(config) (1)

is redundant. It is sufficient to write:


[module].configure(config) (2)

Nevertheless, I kept the (1) style of writing because I thought it might better illustrate that the called code modifies the [config] dictionary.

  • line 1: the [config] dictionary received by the [configure] function has a ‘sgbd’ key whose value is taken from the list [‘mysql’, ‘pgres’]. [mysql] means that the database used is managed by MySQL, while ‘pgres’ means that the database used is managed by PostgreSQL;
  • Lines 4–27: We list all the directories containing elements necessary for the web application. They will be part of the application’s Python Path (lines 30–31);
  • lines 33–40: only certain users will be allowed to access the application. Here, we have a list with a single user;
  • lines 43–46: the [config_database] script builds the configuration for the database being used;
  • line 46: the configuration built by the [config_database] script is a dictionary that we store in the general configuration associated with the ‘database’ key;
  • lines 48–51: the [config_layers] script instantiates the web application layers. It returns a dictionary that is stored in the general configuration under the ‘layers’ key;

The [config_database] script is the one already used in |version 5|. We include it here for reference:


def configure(config: dict) -> dict:
    # SQLAlchemy configuration
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
    from sqlalchemy.orm import mapper, sessionmaker

    # connection strings for the databases in use
    connection_strings = {
        'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
        'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
    }
    # connection string for the database in use
    engine = create_engine(connection_strings[config['sgbd']])

    # metadata
    metadata = MetaData()

    # the constants table
    constants_table = Table("tbconstants", metadata,
                             Column('id', Integer, primary_key=True),
                             Column('half-share_income_limit', Float, nullable=False),
                             Column('income_limit_single_for_reduction', Float, nullable=False),
                             Column('couple_income_limit_for_reduction', Float, nullable=False),
                             Column('half-share_reduction_value', Float, nullable=False),
                             Column('single_discount_limit', Float, nullable=False),
                             Column('couple_discount_limit', Float, nullable=False),
                             Column('single_tax_threshold_for_discount', Float, nullable=False),
                             Column('couple_tax_ceiling_for_discount', Float, nullable=False),
                             Column('maximum_10_percent_deduction', Float, nullable=False),
                             Column('minimum_10_percent_deduction', Float, nullable=False)
                             )

    # the tax bracket table
    bracket_table = Table("tbtranches", metadata,
                           Column('id', Integer, primary_key=True),
                           Column('limit', Float, nullable=False),
                           Column('coeffr', Float, nullable=False),
                           Column('coeffn', Float, nullable=False)
                           )
    # mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constants import Constants
    mapper(Constants, constants_table)

    # the session factory
    session_factory = sessionmaker()
    session_factory.configure(bind=engine)

    # a session
    session = session_factory()

    # We store some information and return it in a dictionary
    return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
            "constants_table": constants_table, "session": session}

The [config_layers] script configures the web server layers. We reuse a |script| we’ve seen before:


def configure(config: dict) -> dict:
    # instantiating the application layers
    
    # DAO
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    dao = ImpotsDaoWithAdminDataInDatabase(config)
    
    # business logic
    from TaxBusinessLogic import TaxBusinessLogic
    business = BusinessTaxes()

    # we put the layer instances in a dictionary that we return to the calling code
    return {
        "dao": dao,
        "business": business
    }
  • Line 6: The [dao] layer is implemented using a database;
  • [ImpotsDaoWithAdminDataInDatabase] has been defined |here|;
  • [BusinessTaxes] has been defined |here|;

The main script [server_01] is as follows:


# expects a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
    dbms = sys.argv[1].lower()
    error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
    print(f"syntax: {syntax}")
    sys.exit()

# configure the application
import config
config = config.configure({'db': db})

# dependencies
from TaxError import TaxError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status

# Retrieving data from the tax authority
try:
    # admindata will be read-only application-scope data
    admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as error:
    print(f"The following error occurred: {error}")
    sys.exit(1)

# Flask application
app = Flask(__name__)


# Home URL: /?married=xx&children=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    # initially no errors
    errors = []
    # The request must have three parameters in the URL
    if len(request.args) != 3:
        errors.append("GET method required with only the parameters [married, children, salary]")

    # retrieve marital status from the URL
    married = request.args.get('married')
    if married is None:
        errors.append("[married] parameter missing")
    else:
        married = married.strip().lower()
        error = married != "yes" and married != "no"
        if error:
            errors.append(f"invalid married parameter [{married}]")

    # retrieve the number of children from the URL
    children = request.args.get('children')
    if children is None:
        errors.append("[children] parameter missing")
    else:
        children = children.strip()
        match = re.match(r"^\d+", children)
        if not match:
            errors.append(f"invalid [{children}] parameter")
        else:
            children = int(children)

    # retrieve the salary from the URL
    salary = request.args.get('salary')
    if salary is None:
        errors.append("[salary] parameter missing")
    else:
        salary = salary.strip()
        match = re.match(r"^\d+", salary)
        if not match:
            errors.append(f"invalid salary parameter [{salary}]")
        else:
            salary = int(salary)

    # Invalid parameters in the URL?
    for key in request.args.keys():
        if key not in ['married', 'children', 'salary']:
            errors.append(f"Invalid parameter [{key}]")

    # any errors?
    if errors:
        # send an error response to the client
        results = {"response": {"errors": errors}}
        return json_response(results, status.HTTP_400_BAD_REQUEST)

    # no errors, we can proceed
    # calculate the tax
    taxpayer = TaxPayer().fromdict({'married': married, 'children': children, 'salary': salary})
    config["layers"]["occupation"].calculate_tax(taxpayer, admindata)
    # send the response to the client
    return json_response({"response": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)


# main only
if __name__ == '__main__':
    # start the Flask server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • lines 1–10: retrieve the parameter indicating which DBMS to use;
  • lines 12–14: With this information, we can configure the application. In particular, the Python Path is constructed;
  • lines 16–23: With the new Python Path, we import the necessary modules;
  • lines 25–31: retrieve data from the tax authority to calculate the tax;
  • lines 33–34: instantiation of the Flask application;
  • Line 38: The Flask application only serves the URL [/]. It expects a URL formatted as follows: [/ ?married=xx&children=yy&salary=zz], where:
    • xx: yes / no;
    • yy: number of children;
    • zz: annual salary;
  • lines 40–89: we check the validity of the URL parameters;
  • line 41: we will accumulate error messages in the [errors] list;
  • line 43: you may recall that the parameters of the URL are found in [request.args] (see |here|):
    • the [request] object is the Flask object imported on line 20;
    • the [request.args] object behaves like a dictionary;
  • lines 43–44: we verify that there are exactly three parameters (no fewer, no more);
  • lines 46–49: we check that the [married] parameter is present in the URL;
  • lines 50–54: if it is present, we check that its lowercase value, stripped of leading and trailing whitespace, is “yes” or “no”;
  • lines 56–59: check that the [children] parameter is in the URL;
  • lines 60–66: if present, check that its value is a positive integer;
  • line 66: remember that URL parameters and their values are strings. The value of the [children] parameter is converted to an ‘int’;
  • lines 68–78: For the [salary] parameter, we perform the same checks as for the [children] parameter;
  • lines 81–83: we check that there are no parameters other than [‘married’, ‘children’, ‘salary’] in the URL;
  • lines 85–89: if, after all these checks, the [errors] list is not empty, we send this list of errors to the client as a JSON string along with the status code [400 Bad Request];

Since we will often need to send a JSON string in response to the client later on, the few lines required for this have been factored into the [myutils.py] module that we have already used:

Image

The [myutils.py] script becomes the following:


# imports
import json
import os
import sys

from flask import make_response


def set_syspath(absolute_dependencies: list):
    # absolute_dependencies: a list of absolute folder names

    ….


# Generate a JSON HTTP response
def json_response(response: dict, status_code: int) -> tuple:
    # HTTP response body
    response = make_response(json.dumps(response, ensure_ascii=False))
    # HTTP response body is JSON
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    # send the HTTP response
    return response, status_code
  • Line 16: The [json_response] function expects two parameters:
    • [response]: the dictionary containing the JSON string to be sent to the web client;
    • [status_code]: the HTTP status code of the response;
  • line 18: we set the JSON body of the response;
  • line 20: we add the HTTP header that tells the web client it will receive JSON;
  • line 22: we send the HTTP response to the calling code. It is up to the calling code to send it to the web client;

The [__init__.py] file changes as follows:


from .myutils import set_syspath, json_response

The new version of [myutils] is installed among the machine-wide modules using the [pip install .] command in a PyCharm terminal:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install .
Processing c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\packages
Using legacy setup.py install for myutils, since the 'wheel' package is not installed.
Installing collected packages: myutils
  Attempting to uninstall: myutils
    Found existing installation: myutils 0.1
    Uninstalling myutils-0.1:
      Successfully uninstalled myutils-0.1
    Running setup.py install for myutils ... done
Successfully installed myutils-0.1
  • Line 1: You must be in the [packages] folder to enter this command;

The code for the [server_01] script continues as follows:


…    
    # Any errors?
    if errors:
        # send an error response to the client
        results = {"response": {"errors": errors}}
        return json_response(results, status.HTTP_400_BAD_REQUEST)

    # no errors, we can proceed
    # calculate the tax
    taxpayer = TaxPayer().fromdict({'id': 0, 'married': married, 'children': children, 'salary': salary})
    config["layers"]["job"].calculate_tax(taxpayer, admindata)
    # send the response to the client
    return json_response({"response": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
  • line 10: at this point, the expected parameters in the URL are present and correct;
  • line 10: we create the [TaxPayer] object that models the taxpayer;
  • line 11: we ask the [business] layer to calculate the tax. Note that the elements calculated by the [business] layer are inserted into the [taxpayer] object passed as a parameter;
  • line 13: the response is sent to the web client as a JSON string. This is the JSON string of a dictionary. Associated with the [result] key, we place the dictionary of the [taxpayer] object. We could not place the [taxpayer] object itself because it is not serializable in JSON;

We create two execution configurations, one for MySQL and the other for PostgreSQL:

Image

Here are some execution examples (you have launched the [server_01] application and the DBMS, then requested the URL http://localhost:5000/ using a browser):

Image

Image

Here is an example of the request in the Postman console:

Image


GET /?mari%C3%A9=xx&enfants=yy&salaire=zz HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e4c5df8c-4bd6-4250-b789-b7b164db4eff
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, Jul 17, 2020 6:15:44 AM GMT

{"response": {"errors": ["invalid married parameter [xx]", "invalid children parameter [yy]", "invalid salary parameter [zz]"]}}
  • line 1: an incorrect URL is requested;
  • line 10: the server responds with status 400 BAD REQUEST;

23.2.2. Version 2

Image

Version 2 of the server isolates URL processing in the [index_controller] module [5]:


# import dependencies
import re

from flask_api import status
from werkzeug.local import LocalProxy


# Parameterized URL: /?married=xx&children=yy&salary=zz
def execute(request: LocalProxy, config: dict) -> tuple:
    # dependencies
    from TaxPayer import TaxPayer

    # initially no errors
    errors = []
    # the request must have three parameters
    if len(request.args) != 3:
        errors.append("GET method required with only the parameters [married, children, salary]")

    # retrieve marital status from the URL
    married = request.args.get('married')
    if married is None:
        errors.append("[married] parameter missing")
    else:
        married = married.strip().lower()
        error = married != "yes" and married != "no"
        if error:
            errors.append(f"Invalid 'married' parameter [{married}]")

    # Get the number of children from the URL
    children = request.args.get('children')
    if children is None:
        errors.append("[children] parameter missing")
    else:
        children = children.strip()
        match = re.match(r"^\d+", children)
        if not match:
            errors.append(f"invalid children parameter {children}")
        else:
            children = int(children)

    # retrieve the salary from the URL
    salary = request.args.get('salary')
    if salary is None:
        errors.append("parameter [salary] missing")
    else:
        salary = salary.strip()
        match = re.match(r"^\d+", salary)
        if not match:
            errors.append(f"invalid {salary} parameter")
        else:
            salary = int(salary)

    # any other parameters in the URL?
    for key in request.args.keys():
        if not key in ['married', 'children', 'salary']:
            errors.append(f"Invalid parameter [{key}]")

    # any errors?
    if errors:
        # send an error response to the client
        results = {"response": {"errors": errors}}
        return results, status.HTTP_400_BAD_REQUEST

    # no errors, we can proceed
    # Calculating taxes
    taxpayer = TaxPayer().fromdict({'married': married, 'children': children, 'salary': salary})
    config["layers"]["occupation"].calculate_tax(taxpayer, config["admindata"])
    # send the response to the client
    return {"response": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
  • line 9: the [execute] function receives two parameters:
    • [request]: the client's HTTP request;
    • [config]: the application's configuration dictionary;

The [server_02] script is as follows:


# Expects a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
    dbms = sys.argv[1].lower()
    error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
    print(f"syntax: {syntax}")
    sys.exit()

# configure the application
import config
config = config.configure({'db': db})

# dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller

# Retrieving data from the tax authority
try:
    # admindata will be a read-only application-scoped variable
    config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as error:
    print(f"The following error occurred: {error}")
    sys.exit(1)

# Flask application
app = Flask(__name__)


# Home URL: /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    # execute the request
    result, statusCode = index_controller.execute(request, config)
    # send the response
    return json_response(result, statusCode)


# main only
if __name__ == '__main__':
    # start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • lines 36–41: handling the / route;
  • line 39: using the [IndexController.execute] function;

We will now use this technique: each route will be handled by its own module.

The execution results are the same as for version 1.

23.2.3. Version 3

Version 3 introduces the concept of authentication.

The [server_03] script becomes the following:


# expect a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
    dbms = sys.argv[1].lower()
    error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
    print(f"syntax: {syntax}")
    sys.exit()

# configure the application
import config
config = config.configure({'db': db})

# dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
from flask_httpauth import HTTPBasicAuth
import index_controller

# Retrieving data from the tax authority
try:
    # config['admindata'] will be a read-only application-scope variable
    config["admindata"] = config["layers"]["dao"].get_admindata()
except TaxError as error:
    print(f"The following error occurred: {error}")
    sys.exit(1)

# authentication handler
auth = HTTPBasicAuth()


# authentication method
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
    # list of users
    users = config['users']
    # iterate through this list
    for user in users:
        if user['login'] == login and user['password'] == password:
            return True
    # no match found
    return False


# Flask application
app = Flask(__name__)


# Home URL: /?married=xx&children=yy&salary=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
    # execute the request
    result, statusCode = index_controller.execute(request, config)
    # send the response
    return json_response(result, statusCode)


# main only
if __name__ == '__main__':
    # start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • line 21: import an authentication handler. There are various types of authentication for a web server. The one we’re using here is called [HTTP Basic]. Each type of authentication follows a specific client/server dialogue;
  • line 33: create an instance of the authentication handler;
  • line 37: the [@auth.verify_password] annotation marks the function to be executed when the authentication handler wants to verify the username and password sent by the client according to the [HTTP Basic] protocol;
  • line 55: the [@auth.login_required] annotation marks a route for which the web client must be authenticated. If the web client has not yet sent its credentials, the web server will automatically request them using the HTTP Basic protocol;

The [flask_httpauth] module must be installed:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install flask_httpauth
Collecting flask_httpauth
  Downloading Flask_HTTPAuth-4.1.0-py2.py3-none-any.whl (5.8 kB)
Requirement already satisfied: Flask in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from flask_httpauth) (1.1.2)
Requirement already satisfied: itsdangerous>=0.24 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.1.0)
Requirement already satisfied: click>=5.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (7.1.2)
Requirement already satisfied: Jinja2>=2.10.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (2.11.2)
Requirement already satisfied: Werkzeug>=0.15 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.0.1)
Requirement already satisfied: MarkupSafe>=0.23 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Jinja2>=2.10.1->Flask->flask_httpauth) (1.1.1
)
Installing collected packages: flask-httpauth
Successfully installed flask-httpauth-4.1.0

Let’s see what happens in the Postman console. You:

  • create a run configuration;
  • launch the web application;
  • launch the database of your choice;
  • request the URL [/] with Postman;

The client/server dialogue in the Postman console is as follows:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
GET / HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e65e2a28-4fe3-423b-88b3-b3e5a83092b1
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 19
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:05:37 GMT

Unauthorized Access
  • Line 10: The server responds that we are not authorized to access the URL [/];
  • Line 13: It tells us which authentication protocol to use, in this case the Basic Authentication protocol;

It is possible to configure Postman to send the user credentials according to the Basic Authentication protocol:

Image

  • in [6-7] we enter the credentials present in the [config] script: Image 

    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

The client/server dialogue in the Postman console becomes the following:


GET / HTTP/1.1
Authorization: Basic YWRtaW46YWRtaW4=
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 5ce20822-e87c-4eef-a2f4-b9eaec38d881
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 203
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:20:01 GMT

{"response": {"errors": ["GET method required with only the [married, children, salary] parameters", "missing [married] parameter", "missing [children] parameter", "missing [salary] parameter"]}}
  • Line 2: The Postman client sends the user credentials [admin / admin] in encrypted form;
  • line 17: the server responds correctly. It reports errors because the parameters [married, children, salary] were not sent (line 1), but it does not report an authentication error;

Now let’s request the URL / using a browser (Firefox below):

Image

  • As with Postman, Firefox received the HTTP response from the server with the following HTTP headers: 
1
2
3
4
HTTP/1.0 401 UNAUTHORIZED
…
WWW-Authenticate: Basic realm="Authentication Required"
…

Firefox, like other browsers, does not stop the dialog when it receives these headers. It asks the user for the credentials requested by the server. In the example above, simply typing admin / admin will receive the server’s response:

Image

23.3. The web client of the tax calculation server

23.3.1. Introduction

In the previous section, the web client for the tax calculation server was a browser. In this section, the web client will be a console script. The architecture becomes as follows:

Image

  • the web client consists of layers [1-2];
  • the web server consists of layers [3-9]. As mentioned in the previous section;

We therefore need to write layers [1-2].

Layer [dao] [2] must be able to communicate with the web server [3]. We now understand the HTTP protocol and could write, using the [pycurl] module we’ve already studied for example, a script that communicates with the web server [3]. However, there are modules specialized in HTTP client/server communication. We’ll use one of them, the [requests] module:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install requests
Collecting requests
  Downloading requests-2.24.0-py2.py3-none-any.whl (61 kB)
     || 61 kB 137 kB/s
Collecting idna<3,>=2.5
  Downloading idna-2.10-py2.py3-none-any.whl (58 kB)
     || 58 kB 692 kB/s
Collecting chardet<4,>=3.0.2
  Downloading chardet-3.0.4-py2.py3-none-any.whl (133 KB)
     || 133 kB 1.3 MB/s
Collecting urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1
  Downloading urllib3-1.25.9-py2.py3-none-any.whl (126 kB)
     || 126 kB 1.1 MB/s
Collecting certifi>=2017.4.17
  Downloading certifi-2020.6.20-py2.py3-none-any.whl (156 kB)
     || 156 kB 1.1 MB/s
Installing collected packages: idna, chardet, urllib3, certifi, requests
Successfully installed certifi-2020.6.20 chardet-3.0.4 idna-2.10 requests-2.24.0 urllib3-1.25.9

The directory structure for the web client scripts is as follows:

Image

The script will implement the batch-mode tax calculation application described in |version 1|. The latest version of this application is |version 5|. Here is a reminder of how it works:

  • the taxpayers for whom the tax will be calculated are listed in the text file [taxpayersdata.txt]: 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# valid data: id, married, children, salary
1,yes,2,55555
2,yes,2,50000
3,yes,3,50000
4,no,2,100000
5,no,3,100000
6,yes,3,100000
7,yes,5,100000
8,no,0,100000
9,yes,2,30000
10,no,0,200,000
11,yes,3,200000
# creating erroneous rows
# not enough values
11,12
# incorrect values
x,x,x,x
  • The results are saved in two files:
  • The text file [errors.txt] lists the errors detected in the taxpayer file: 

Analyzing the file C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt

Line 15, not enough values to unpack (expected 4, got 2)
Line 17, MyException[1, The ID of an entity <class 'TaxPayer.TaxPayer'> must be an integer >=0]
  • (continued)
    • The JSON file [results.json] contains the tax calculation results for the various taxpayers: 

[
  {
    "id": 0,
    "married": "yes",
    "children": 2,
    "salary": 55555,
    "tax": 2814,
    "surcharge": 0,
    "rate": 0.14,
    "discount": 0,
    "reduction": 0
  },
  {
    "id": 1,
    "married": "yes",
    "children": 2,
    "salary": 50000,
    "tax": 1384,
    "surcharge": 0,
    "rate": 0.14,
    "discount": 384,
    "reduction": 347
  },
  …
]

23.3.2. Web Client Configuration

Image

Configuration is performed using two scripts:

  • [config], which handles all configuration outside of the architecture layers;
  • [config_layers], which handles the configuration of the architecture layers;

The [config] script is as follows:


def configure(config: dict) -> dict:
    import os

    # Step 1 ------

    # directory of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))

    # root directory
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    # absolute dependencies
    absolute_dependencies = [
        # project directories
        # BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        # TaxDaoInterface, TaxBusinessInterface, TaxUiInterface
        f"{root_dir}/taxes/v04/interfaces",
        # AbstractTaxDao, TaxConsole, TaxBusiness
        f"{root_dir}/taxes/v04/services",
        # TaxDaoWithAdminDataInDatabase
        f"{root_dir}/taxes/v05/services",
        # AdminData, TaxError, Taxpayer
        f"{root_dir}/impots/v04/entities",
        # Constants, tax brackets
        f"{root_dir}/taxes/v05/entities",
        # TaxDaoWithHttpClient
        f"{script_dir}/../services",
        # configuration scripts
        script_dir,
    ]

    # Set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # step 2 ------
    # Configure the application with constants
    config.update({
        "taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
        "resultsFilename": f"{script_dir}/../data/output/results.json",
        "errorsFilename": f"{script_dir}/../data/output/errors.txt",
        "server": {
            "urlServer": "http://127.0.0.1:5000/",
            "authBasic": True,
            "user": {
                "login": "admin",
                "password": "admin"
            }
        }
    }
    )

    # Step 3 ------
    # instantiating layers
    import config_layers
    config['layers'] = config_layers.configure(config)

    # return the configuration
    return config
  • line 1: the [configure] function takes as a parameter the dictionary to be filled with configuration information. This dictionary may already be pre-filled or empty. Here, it will be empty;
  • lines 40–42: the absolute paths of the three text files managed by the [dao] layer;
  • lines 43-50: associated with the [server] key, the information the [dao] layer needs to know about the web server with which it must communicate:
    • line 44: the URL of the web service;
    • line 45: the [authBasic] key is set to True if access to the URL requires Basic authentication;
    • lines 46–49: the credentials of the user who will authenticate if authentication is required;
  • lines 56–57: we instantiate the layers—in this case, the single [dao] layer—and place the layer references in [config] under the [layers] key;

The [config_layers] script is as follows:


def configure(config: dict) -> dict:
    # instantiation of the application layers

    # dao layer
    from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
    dao = ImpôtsDaoWithHttpClient(config)

    # return the layer configuration
    return {
        "dao": dao
    }
  • Line 1: The [configure] function receives the dictionary that configures the application;
  • lines 4–6: the [dao] layer is instantiated. On line 6, we pass it the application configuration, where it will find the information it needs;
  • lines 8–11: a dictionary is returned containing the reference to the [dao] layer;

23.3.3. The main script [main]

The main script [main] is a variant of the one from |version 5|:


# configure the application
import config
config = config.configure({})

# dependencies
from ImpôtsError import ImpôtsError

# code
try:
    # retrieve the [dao] layer
    dao = config["layers"]["dao"]
    # reading taxpayer data
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    # Any taxpayers?
    if not taxpayers:
        raise TaxError(f"No valid taxpayers in the file {config['taxpayersFilename']}")
    # Calculate taxpayers' taxes
    for taxpayer in taxpayers:
        # taxpayer is both an input and output parameter
        # taxpayer will be modified
        dao.calculate_tax(taxpayer)
    # writing the results to a text file
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError   as error:
    # display the error
    print(f"The following error occurred: {error}")
finally:
    # done
    print("Job completed...")
  • lines 2-3: the application is configured;
  • line 13: the [dao] layer provides the list of taxpayers for whom taxes must be calculated;
  • line 21: the [dao] layer calculates the tax for each of them;
  • line 23: the results are saved to a JSON file;

23.3.4. Implementation of the [dao] layer

Image

Let’s revisit the client/server architecture used:

Image

  • in [2, 6], we see that the [dao] layer has two roles:
    • it accesses the file system both to read taxpayer data and to write the results of tax calculations. We already have an |AbstractImpôtsDao| class that can do this. It has been in use since |version 4|;
    • it communicates with the web server [3];

In |version 5|, the main script [main] [1] communicated directly with the [business] layer [4]. We would prefer not to change this script. To achieve this, we will ensure that the [DAO] layer [2] implements the interface of the [business] layer [4]. This way, the main script [main] will appear to communicate directly with the [business] layer [4] and can completely ignore the fact that it is located on another machine.

A definition of the class implementing the [DAO] layer [2] could be as follows:


class TaxDaoWithHttpClient(AbstractTaxDao, TaxBusinessInterface):
  • The [TaxDaoWithHttpClient] class:
    • inherits from the [AbstractTaxDao] class, which allows it to handle communication with the file system [6];
    • implements the [InterfaceImpôtsMétier] interface so as not to have to change the main script [main] of |version 5|;

The complete code for the [TaxDaoWithHttpClient] class is as follows:


# imports
import requests
from flask_api import status

from AbstractTaxDao import AbstractTaxDao
from AdminData import AdminData
from TaxError import TaxError
from BusinessTaxInterface import BusinessTaxInterface
from TaxPayer import TaxPayer


class TaxDaoWithHttpClient(AbstractTaxDao, BusinessTaxInterface):

    # constructor
    def __init__(self, config: dict):
        # Initialize parent
        AbstractTaxDao.__init__(self, config)
        # store parameters
        self.__config_server = config["server"]

    # unused method from [AbstractImpôtsDao]
    def get_admindata(self) -> AdminData:
        pass

    # tax calculation
    def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
        # let exceptions be raised
        # get parameters
        params = {"married": taxpayer.married, "children": taxpayer.children, "salary": taxpayer.salary}
        # connection with Basic Authentication?
        if self.__config_server['authBasic']:
            response = requests.get(
                # URL of the server being queried
                self.__config_server['urlServer'],
                # URL parameters
                params=params,
                # Basic authentication
                auth=(
                    self.__config_server["user"]["login"],
                    self.__config_server["user"]["password"]))
        else:
            # connection without Basic authentication
            response = requests.get(self.__config_server['urlServer'], params=params)
        # verification
        print(response.text)
        # HTTP response status code
        status_code = response.status_code
        # put the JSON response into a dictionary
        result = response.json()
        # error if status code is not 200 OK
        if status_code != status.HTTP_200_OK:
            # we know that errors have been associated with the [errors] key in the response
            raise ImpôtsError(87, result['response']['errors'])
        # we know that the result has been associated with the [result] key in the response
        # update the input parameter with this result
        taxpayer.fromdict(result["response"]["result"])
  • lines 21–23: The [AbstractTaxDao] class (line 12) has an abstract method [get_admindata]. We are required to implement it even if we don’t use it (admindata is managed by the server, not by the client);
  • line 26: the method [calculate_tax] belongs to the interface [InterfaceImpôtsMétier] (line 12). We must implement it;
  • line 15: the constructor receives the application configuration dictionary as its only parameter;
  • lines 16–17: the parent class [AbstractTaxDao] is initialized by passing it, here as well, the application configuration. It will find there the names of the three text files it needs to manage;
  • lines 18–19: information regarding the tax calculation web server is stored locally within the class;
  • line 26: the [calculate_tax] method receives an object of type |Taxpayer| as a parameter. To comply with the signature of the [InterfaceImpôtsMétier.calculate_tax] method, it also receives an [admindata] parameter, which is supposed to encapsulate the tax administration data. On the client side, we do not have this data. This parameter will always remain [None]. This workaround suggests that the [ImpôtsMétier] class was initially poorly designed:
  • the signature of [calculate_tax] should have simply been: 

def calculate_tax(self, taxpayer: TaxPayer)

and the [admindata: AdminData] parameter should have been passed to the class constructor;

  • line 27: the code for the [calculate_tax] method has not been encapsulated in a try / catch / finally block. This means that any exceptions will not be handled and will be propagated to the calling code, in this case the [main] script. This script does catch all exceptions propagated from the [dao] layer;
  • Line 28: The tax calculation is performed on the server side. We will therefore need to communicate with it. We do this using the [requests] module imported on line 2;
  • lines 31–43: to send a GET request to the web server, we use the [requests.get] method:
    • lines 33–34: the first parameter of the method is the URL to contact;
    • lines 35–40: the other two parameters are named parameters whose order does not matter;
    • lines 35-36: the value of the named parameter [params] must be a dictionary containing the information to be included in the URL in the form [/url?param1=value1&param2=value2&…];
    • line 29: the dictionary containing the three parameters [married, children, salary] that the web server expects. We don’t have to worry about the encoding (called urlencoded) that these parameters must undergo. [requests] handles this;
    • lines 37–40: the parameter named [auth] is a tuple of two elements (login, password). It represents the credentials for Basic authentication;
  • lines 44–45: these two lines are for educational purposes only (we’ll comment them out once debugging is complete):
    • [response] represents the server’s HTTP response;
    • [response.text] represents the text of the document contained in this response. During debugging, it is useful to verify what the server has sent us;
  • line 47: [response.status_code] is the HTTP status code of the received response. Our server sends only three:
    • 200 OK
    • 400 BAD REQUEST
    • 500 INTERNAL SERVER ERROR
  • line 49: our server always sends JSON, even in the event of an error. The [response.json()] function creates a dictionary from the received JSON string. Let’s review the two possible forms for the JSON string: 

{"response": {"errors": ["GET method required with only the [married, children, salary] parameters", "missing [married] parameter", "missing [children] parameter", "missing [salary] parameter"]}}

{"response": {"result": {"id": 0, "married": "yes", "children": 3, "salary": 200000, "tax": 42842, "surcharge": 17283, "rate": 0.41, "discount": 0, "reduction": 0}}}
  • lines 51–53: if the status code is not 200, then an exception is thrown with the error messages included in the response;
  • line 56: retrieve the dictionary produced by the tax calculation and use it to update the [taxpayer] input parameter;

23.3.5. Execution

To run the client:

  • start the server [server_03] with the DBMS of your choice;
  • run the client’s [main] script;

The results will be found in the [data/output] folder. They are the same as for version 5.

23.4. Tests of the [dao] layer

Let’s return to the client/server application architecture:

Image

  • in the client code, we have ensured that the [dao] layer [1] provides the same interface as the [business] layer [3]. We will therefore use the test class |TestDaoMétier|, which we previously studied, to test the [business] layer [3];

The test class will be executed in the following environment:

Image

  • Configuration [2] is identical to configuration [1], which we just examined;

The test class [TestHttpClientDao] is as follows:


import unittest


class TestHttpClientDao(unittest.TestCase):

    def test_1(self) -> None:
        from TaxPayer import TaxPayer

        # {'married': 'yes', 'children': 2, 'salary': 55555,
        # 'tax': 2814, 'surcharge': 0, 'discount': 0, 'reduction': 0, 'rate': 0.14}
        taxpayer = TaxPayer().fromdict({"married": "yes", "children": 2, "salary": 55555})
        dao.calculate_tax(taxpayer)
        # verification
        self.assertAlmostEqual(taxpayer.tax, 2815, delta=1)
        self.assertEqual(taxpayer.discount, 0)
        self.assertEqual(taxpayer.reduction, 0)
        self.assertAlmostEqual(taxpayer.rate, 0.14, delta=0.01)
        self.assertEqual(taxpayer.surcharge, 0)

    …

    def test_11(self) -> None:
        from TaxPayer import TaxPayer

        # {'married': 'yes', 'children': 3, 'salary': 200000,
        # 'tax': 42842, 'surcharge': 17283, 'discount': 0, 'reduction': 0, 'rate': 0.41}
        taxpayer = TaxPayer().fromdict({'married': 'yes', 'children': 3, 'salary': 200000})
        dao.calculate_tax(taxpayer)
        # checks
        self.assertAlmostEqual(taxpayer.tax, 42842, 1)
        self.assertEqual(taxpayer.discount, 0)
        self.assertEqual(taxpayer.reduction, 0)
        self.assertAlmostEqual(taxpayer.rate, 0.41, delta=0.01)
        self.assertAlmostEqual(taxpayer.surcharge, 17283, delta=1)


if __name__ == '__main__':

    # configure the application
    import config
    config = config.configure({})

    # DAO layer
    dao = config['layers']['dao']

    # Run the test methods
    print("Tests in progress...")
    unittest.main()

This class is similar to the one already studied in version 4 of the application.

  • lines 40-41: configure the test environment;
  • line 44: we retrieve a reference to the [DAO] layer;
  • lines 47-48: we run the tests;

To run the tests, we create a |run configuration|:

Image

  • We create a run configuration for a console script, not for a UnitTest;

When running this configuration, the following results are obtained:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\Scripts\python.exe C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020/impots/http-clients/01/tests/TestHttpClientDao.py
tests in progress...
{"response": {"result": {"married": "yes", "children": 2, "salary": 55555, "tax": 2814, "surcharge": 0, "rate": 0.14, "discount": 0, "reduction": 0}}}
....{"response": {"result": {"married": "no", "children": 0, "salary": 200000, "tax": 64210, "surcharge": 7498, "rate": 0.45, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "yes", "children": 3, "salary": 200000, "tax": 42,842, "surcharge": 17,283, "rate": 0.41, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "yes", "children": 2, "salary": 50000, "tax": 1384, "surcharge": 0, "rate": 0.14, "discount": 384, "reduction": 347}}}
{"response": {"result": {"married": "yes", "children": 3, "salary": 50000, "tax": 0, "surcharge": 0, "rate": 0.14, "discount": 720, "reduction": 0}}}
...{"response": {"result": {"married": "no", "children": 2, "salary": 100000, "tax": 19,884, "surcharge": 4,480, "rate": 0.41, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "no", "children": 3, "salary": 100000, "tax": 16,782, "surcharge": 7,176, "rate": 0.41, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "yes", "children": 3, "salary": 100000, "tax": 9,200, "surcharge": 2,180, "rate": 0.3, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "yes", "children": 5, "salary": 100000, "tax": 4,230, "surcharge": 0, "rate": 0.14, "discount": 0, "reduction": 0}}}
{"response": {"result": {"married": "no", "children": 0, "salary": 100000, "tax": 22,986, "surcharge": 0, "rate": 0.41, "discount": 0, "reduction": 0}}}
....
{"response": {"result": {"married": "yes", "children": 2, "salary": 30000, "tax": 0, "surcharge": 0, "rate": 0.0, "discount": 0, "reduction": 0}}}
----------------------------------------------------------------------
Ran 11 tests in 0.130s

OK

Process finished with exit code 0

All 11 tests passed.

https://tahe.developpez.com
This tutorial written by Serge Tahé is made available to the public under the terms of the Creative Commons Attribution – Non-Commercial – ShareAlike 3.0 Unported.