33. Application exercise: Version 13
Version 13 modifies version 12 in the following ways:
- Certain parts of the code have been restructured (refactoring);
- session management is handled differently using the [flask_session] module;
- encrypted passwords are used in the configuration file;
Version 13 is initially created by copying version 12:
- in [1], the configuration will be refactored. In particular, it will be moved out of the [flask] folder;
- in [2], the main script will be refactored. It is also moved out of the [flask] folder;
- in [3], the configuration will be split across multiple files;
- in [4]: we will simplify the main script [main] by moving code to other files;
- in [5], the authentication controller will be modified since user passwords will now be encrypted;
- in [6], the main controller will incorporate code previously present in the main script [main];
33.1. Refactoring the application configuration
Three new configuration files are created:
- [mvc]: to configure the application’s MVC architecture;
- [parameters]: which will contain all the application’s constants;
- [syspath]: which configures the application’s Python path;
The [syspath.py] file is as follows:
def configure(config: dict) -> dict:
import os
# 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"
# 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",
# Logger, SendAdminMail
f"{root_dir}/taxes/http-servers/02/utilities",
# directory of the main script
script_dir,
# configs [database, layers, parameters, controllers, views]
f"{script_dir}/../configs",
# controllers
f"{script_dir}/../controllers",
# HTTP responses
f"{script_dir}/../responses",
# view models
f"{script_dir}/../models_for_views",
]
# Set the syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# return the configuration
return {
"root_dir": root_dir,
"script_dir": script_dir
}
- The [syspath] script is used to configure the application's Python Path (lines 40–41);
- it provides two pieces of information useful to the other configuration scripts (lines 45–46);
The [mvc] script configures the MVC architecture of the JSON/XML/HTML web application:
def configure(config: dict) -> dict:
# MVC application configuration
# controllers
from DisplayTaxCalculationController import DisplayTaxCalculationController
from AuthenticateUserController import AuthenticateUserController
from CalculateTaxesController import CalculateTaxesController
from CalculateTaxesController import CalculateTaxesController
from EndSessionController import EndSessionController
from GetAdminDataController import GetAdminDataController
from InitSessionController import InitSessionController
from ListSimulationsController import ListSimulationsController
from MainController import MainController
from DeleteSimulationController import DeleteSimulationController
# HTTP responses
from HtmlResponse import HtmlResponse
from JsonResponse import JsonResponse
from XmlResponse import XmlResponse
# view models
from ModelForAuthentificationView import ModelForAuthentificationView
from ModelForTaxCalculationView import ModelForTaxCalculationView
from ModelForErrorView import ModelForErrorView
from ModelForSimulationListView import ModelForSimulationListView
# Authorized actions and their controllers
controllers = {
# Initialization of a calculation session
"init-session": InitSessionController(),
# user authentication
"authenticate-user": AuthenticateUserController(),
# calculate tax in individual mode
"calculate-tax": CalculateTaxController(),
# calculate tax in batch mode
"calculate-taxes": CalculateTaxesController(),
# list of simulations
"list-simulations": ListSimulationsController(),
# Delete a simulation
"delete-simulation": DeleteSimulationController(),
# end of calculation session
"end-session": EndSessionController(),
# display the tax calculation view
"display-tax-calculation": DisplayTaxCalculationController(),
# retrieve data from the tax administration
"get-admindata": GetAdminDataController(),
# main controller
"main-controller": MainController()
}
# different response types (json, xml, html)
responses = {
"json": JsonResponse(),
"html": HtmlResponse(),
"xml": XmlResponse()
}
# HTML views and their models depend on the state returned by the controller
views = [
{
# authentication view
"statuses": [
700, # /init-session - success
201, # /authenticate-user failure
],
"view_name": "views/authentication-view.html",
"model_for_view": ModelForAuthentificationView()
},
{
# tax calculation view
"statements": [
200, # /authenticate-user success
300, # /calculate-tax success
301, # /calculate-tax failure
800, # /display-tax-calculation success
],
"view_name": "views/tax-calculation-view.html",
"model_for_view": ModelForCalculImpotView()
},
{
# view of the list of simulations
"statements": [
500, # /list-simulations success
600, # /delete-simulation success
],
"view_name": "views/simulation-list-view.html",
"model_for_view": ModelForListeSimulationsView()
},
]
# unexpected errors view
view_errors = {
"view_name": "views/error-view.html",
"model_for_view": ModelForErrorsView()
}
# redirects
redirects = [
{
"statuses": [
400, # /end-session success
],
# redirect to the URL
"to": "/init-session/html",
}
]
# return the MVC configuration
return {
# controllers
"controllers": controllers,
# HTTP responses
"responses": responses,
# views and models
"views": views,
# list of redirects
"redirects": redirects,
# unexpected error view
"error_view": error_view
}
- lines 1-101: this code is known;
- Lines 105–116: We render the application’s MVC configuration;
The [parameters] script collects the application's constants:
def configure(config: dict) -> dict:
# application configuration
# script_dir
script_dir = config['syspath']['script_dir']
# application configuration
parameters = {
# users authorized to use the application
"users": [
{
"login": "admin",
"password": "$pbkdf2-sha256$29000$mPM.h3COkTIGYOzde68VIg$7LH5Q7rN/1hW.Xa.6rcmR6h52PntvVqd5.na7EtgQNw"
}
],
# log file
"logsFilename": f"{script_dir}/../data/logs/logs.txt",
# SMTP server configuration
"adminMail": {
# SMTP server
"smtp-server": "localhost",
# SMTP server port
"smtp-port": "25",
# administrator
"from": "guest@localhost.com",
"to": "guest@localhost.com",
# email subject
"subject": "Tax calculation server crash",
# Set TLS to True if the SMTP server requires authentication, False otherwise
"tls": False
},
# thread pause duration in seconds
"sleep_time": 0,
# Redis server
"redis": {
"host": "127.0.0.1",
"port": 6379
},
}
# Return the application configuration
return parameters
- line 13: user passwords will now be encrypted;
- lines 34–38: configuration of a [Redis] server, which we will return to later;
With these new configuration files, the [config] script becomes the following:
def configure(config: dict) -> dict:
# syspath configuration
import syspath
config['syspath'] = syspath.configure(config)
# Application configuration
import parameters
config['parameters'] = parameters.configure(config)
# database configuration
import database
config["database"] = database.configure(config)
# instantiate the application layers
import layers
config['layers'] = layers.configure(config)
# MVC configuration of the [web] layer
import mvc
config['mvc'] = mvc.configure(config)
# return the configuration
return config
33.2. Refactoring the main script [main]
The main script [main.py] simply starts the server:
# 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 SendAdminMail import SendAdminMail
from Logger import Logger
from ImpôtsError import ImpôtsError
import redis
# Send an email to the administrator
def send_adminmail(config: dict, message: str):
# Send an email to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# Check the log file
logger = None
error = False
error_message = None
try:
# Logger
logger = Logger(config['parameters']['logsFilename'])
except BaseException as exception:
# console log
print(f"The following error occurred: {exception}")
# log the error
error = True
error_message = f"{exception}"
# store the logger in the config
config['logger'] = logger
# error handling
if error:
# email to the administrator
send_adminmail(config, error_message)
# end of application
sys.exit(1)
# startup log
log = "[server] server startup"
logger.write(f"{log}\n")
# Checking Redis server availability
redis_client = redis.Redis(host=config["parameters"]["redis"]["host"],
port=config["parameters"]["redis"]["port"])
# ping the Redis server
try:
redis_client.ping()
except BaseException as exception:
# Redis is not available
log = f"[server] The Redis server is not available: {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# Store the [redis] client in the config
config['redis_client'] = redis_client
# retrieve data from the tax authority
error = False
try:
# admindata will be a read-only application-scope data
config["admindata"] = config["layers"]["dao"].get_admindata()
# success log
logger.write("[server] database connection successful\n")
except ImpôtsError as ex:
# log the error
error = True
# error log
log = f"The following error occurred: {ex}"
# console
print(log)
# log file
logger.write(f"{log}\n")
# email to the administrator
send_adminmail(config, log)
# the main thread no longer needs the logger
logger.close()
# if there was an error, stop
if error:
sys.exit(2)
# Import the web application routes
import routes
routes.config = config
routes.execute(__name__)
- lines 56–73: a Redis server and its client are introduced;
- lines 102–104: the application’s routes have been externalized into the [routes] script;
33.2.1. The [flask_session] and [redis] modules
The [Redis] server will be used to store user sessions. We will use the [flask_session] module to manage these sessions. This module can store user sessions in several locations. Redis is one of them, and we will use it.
The [flask_session] module must be installed in a PyCharm terminal:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install flask-session
Collecting flask-session
…
To communicate with the Redis server, we need a Redis client. This will be provided by the [redis] module, which we will also install:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install redis
Collecting redis
…
33.2.2. The Redis server
The Redis server will be used to store user sessions. The [flask_session] module works as follows:
- Each user has a session ID, and that is what is sent to the client—and only that. The client receives a session cookie only once, after its first request. This cookie contains the user’s session ID, which will not change as the client makes subsequent requests. This is why the server does not need to send a new session cookie;
- Previously, the session content was sent to the client. This will no longer be the case. The user’s session content will be stored on the Redis server;
Laragon comes with a Redis server that is not enabled by default. You must therefore start by enabling it:
- in [3], enable the [Redis] server;
- in [4], leave port [6379], which Redis clients use by default;
Laragon services are automatically restarted after Redis is enabled:
The Redis server can be queried in command mode. Open a Laragon terminal [6]:
- in [1], the [redis-cli] command launches the client in command mode for the Redis server;
As of July 2019, the Redis client can use 172 commands to interact with the server [https://redis.io/commands#list]. One of them [command count] [2] displays this number [3].
Writing to [Redis] is done using the Redis command [set attribute value] [4]. The value can then be retrieved using the command [get attribute] [5].
It may be necessary to clear the Redis database. This is done using the [flushdb] command [6]. Then, if you query the value of the [title] attribute [7], you get a [nil] reference [8] indicating that the attribute was not found. You can also use the [exists] command [9-10] to check if an attribute exists.
To exit the Redis client, type the [quit] command [11].
You can also use a web interface to manage the keys on the Redis server. To do this, the Laragon Apache server must be running:
The following interface appears:
- in [1-4], one of the sessions stored on the Redis server;
33.2.3. Managing the Redis server in the main script [main]
The [main] script checks for the presence of the Redis server as follows:
import redis
…
# Check the availability of the Redis server
redis_client = redis.Redis(host=config["parameters"]["redis"]["host"],
port=config["parameters"]["redis"]["port"])
# ping the Redis server
try:
redis_client.ping()
except BaseException as exception:
# Redis is not available
log = f"[server] The Redis server is unavailable: {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# Store the [redis] client in the config
config['redis_client'] = redis_client
- line 4: the class constructor [redis.Redis] creates a Redis server client. Its properties (address, port) are found in the [parameters] script;
- line 8: the [ping] method checks for the presence of the Redis server;
- lines 9–17: if the ping fails, the error is logged and the server is stopped;
- line 20: the Redis client reference is stored in the configuration;
33.2.4. Route management in the main script [main]
Route handling in [main] is limited to the following lines:
# Import the web application routes
import routes
routes.config = config
routes.execute(__name__)
- line 1: the routes have been externalized into the [routes] module;
- line 3: the routes need to know the execution configuration;
- line 4: we launch the Flask application by passing it the name of the executed script (__main__);
The routes script is as follows:
# dependencies
import os
from flask import Flask, redirect, request, session, url_for
from flask_api import status
from flask_session import Session
# Flask application
app = Flask(__name__, template_folder="../flask/templates", static_folder="../flask/static")
# application configuration
config = {}
# the front controller
def front_controller() -> tuple:
# forward the request to the main controller
main_controller = config['mvc']['controllers']['main-controller']
return main_controller.execute(request, session, config)
@app.route('/', methods=['GET'])
def index() -> tuple:
# Redirect to /init-session/html
return redirect(url_for("init_session", type_response="html"), status.HTTP_302_FOUND)
# init-session
@app.route('/init-session/<string:type_response>', methods=['GET'])
def init_session(type_response: str) -> tuple:
# execute the controller associated with the action
return front_controller()
# authenticate-user
@app.route('/authenticate-user', methods=['POST'])
def authenticate_user() -> tuple:
# execute the controller associated with the action
return front_controller()
# calculate-tax
@app.route('/calculate-tax', methods=['POST'])
def calculate_tax() -> tuple:
# execute the controller associated with the action
return front_controller()
# Calculate taxes in batches
@app.route('/calculate-taxes', methods=['POST'])
def calculate_taxes():
# Execute the controller associated with the action
return front_controller()
# list-simulations
@app.route('/list-simulations', methods=['GET'])
def list_simulations() -> tuple:
# execute the controller associated with the action
return front_controller()
# delete-simulation
@app.route('/delete-simulation/<int:number>', methods=['GET'])
def delete_simulation(number: int) -> tuple:
# execute the controller associated with the action
return front_controller()
# end-session
@app.route('/end-session', methods=['GET'])
def end_session() -> tuple:
# execute the controller associated with the action
return front_controller()
# display-tax-calculation
@app.route('/display-tax-calculation', methods=['GET'])
def display_tax_calculation() -> tuple:
# execute the controller associated with the action
return front_controller()
# get-admindata
@app.route('/get-admindata', methods=['GET'])
def get_admindata() -> tuple:
# execute the controller associated with the action
return front_controller()
def execute(name: str):
# session secret key
app.secret_key = os.urandom(12).hex()
# Flask-Session
app.config.update(SESSION_TYPE='redis',
SESSION_REDIS=config['redis_client'])
Session(app)
# When running the Flask application via a console script
if name == '__main__':
app.config.update(ENV="development", DEBUG=True)
app.run(threaded=True)
- line 9: the Flask application is instantiated;
- line 12: the application configuration is not yet known when the script is written. It is only known at the time of execution;
- lines 20–77: the application routes as defined in the previous version. This remains unchanged;
- lines 14–18: all routes simply call the [front_controller] function. We have stripped this function of its original code. It now simply calls the web application’s main controller;
- lines 79–89: [execute] is the function called by the [main] script to launch the web application;
- line 81: the [flask_session] module uses Flask’s secret key;
- lines 82–84: configuration of the [flask_session] module. This involves adding the keys [SESSION_TYPE, SESSION_REDIS] to the [app.config] configuration of the Flask application [app]:
- [SESSION_TYPE]: the session type. There are several types. The [redis] type indicates that [flask_session] uses a [redis] server to store user sessions. Because of this, we must define the [SESSION_REDIS] key, which must be the reference of a Redis client;
- line 85: the [Flask-Session] is associated with the Flask application;
- lines 86–89: if the [name] parameter on line 79 is the string [__main__], then the Flask application is launched;
33.3. Refactoring the main controller
The code that was previously in the [front_controller] function of the [main] script has been moved to the main controller:
# import dependencies
import threading
import time
from random import randint
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
from Logger import Logger
from SendAdminMail import SendAdminMail
def send_adminmail(config: dict, message: str):
# Send an email to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# Main controller of the application
class MainController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# process the request
logger = None
action = None
response_type1 = None
try:
# retrieve the path elements
params = request.path.split('/')
# the action is the first element
action = params[1]
# no errors initially
error = False
# the session type must be known before certain actions
response_type1 = session.get('response_type')
if type_response1 is None and action != "init-session":
# log the error
result = {"action": action, "status": 101,
"response": ["No session in progress. Start with action [init-session]"]}
error = True
# logger
logger = Logger(config['parameters']['logsFilename'])
# store it in a configuration associated with the thread
thread_config = {"logger": logger}
thread_name = threading.current_thread().name
config[thread_name] = {"config": thread_config}
# log the request
logger.write(f"[MainController] request: {request}\n")
# We terminate the thread if requested
sleep_time = config['parameters']['sleep_time']
if sleep_time != 0:
# The pause is random so that some threads are interrupted and others are not
random = randint(0, 1)
if random == 1:
# log before pause
logger.write(f"[front_controller] paused the thread for {sleep_time} second(s)\n")
# pause
time.sleep(sleep_time)
# For certain actions, you must be authenticated
user = session.get('user')
if not error and user is None and action not in ["init-session", "authenticate-user"]:
# log the error
result = {"action": action, "status": 101,
"response": [f"action [{action}] requested by unauthenticated user"]}
error = True
# Are there any errors?
if error:
# The request is invalid
status_code = status.HTTP_400_BAD_REQUEST
else:
# execute the controller associated with the action
controller = config['mvc']['controllers'][action]
result, status_code = controller.execute(request, session, config)
except BaseException as exception:
# other (unexpected) exceptions
result = {"action": action, "status": 131, "response": [f"{exception}"]}
status_code = status.HTTP_400_BAD_REQUEST
finally:
pass
# Log the result sent to the client
log = f"[MainController] {result}\n"
logger.write(log)
# Was there a fatal error?
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# Send an email to the application administrator
send_adminmail(config, log)
# Determine the desired response type
type_response2 = session.get('typeResponse')
if type_response2 is None and type_response1 is None:
# The session type has not yet been set—it will be JSON
type_response = 'json'
elif type_response2 is not None:
# the response type is known and in the session
type_response = type_response2
else:
# otherwise, continue using type_response1
type_response = type_response1
# build the response to send
response_builder = config['mvc']['responses'][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, result)
# close the log file if it was opened
if logger:
logger.close()
# send the HTTP response
return response, status_code
All of this code has been covered at one time or another.
33.4. Handling encrypted passwords
To handle encrypted passwords, we’ll use the [passlib] module, which we install from a PyCharm terminal:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install passlib
Collecting passlib
…
Here is an example script that encrypts the password passed to it as a parameter:
The [create_hashed_password] script is as follows (https://passlib.readthedocs.io/en/stable/):
import sys
# encryption function
from passlib.hash import pbkdf2_sha256
# wait for the password to be hashed
syntax = f"{sys.argv[0]} password"
error = len(sys.argv) != 2
if error:
print(f"syntax: {syntax}")
sys.exit()
else:
password = sys.argv[1]
# encrypt the password
hash = pbkdf2_sha256.hash(password)
print(f"Encrypted version of [{password}] = {hash}")
# verification
correct = pbkdf2_sha256.verify(password, hash)
print(correct)
- line 16: the password passed as a parameter is encrypted;
- line 20: we compare the password [password] passed as a parameter to its encrypted version [hash]. The [verify] function encrypts the password [password] and compares the resulting encrypted string to [hash]. Returns True if the two strings are equal;
The script above allows us to obtain the encrypted version of the password [admin]:
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-servers/08/passlib/create_hashed_password.py admin
encrypted version of [admin] = $pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek
True
Line 2, the value we put in the [parameters] script:
"users": [
{
"login": "admin",
"password": "$pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek"
}
],
The authentication controller [AuthentifierUtilisateurController] evolves as follows:
from passlib.handlers.pbkdf2 import pbkdf2_sha256
…
# we verify the validity of the (user, password) pair
users = config['parameters']['users']
i = 0
nbusers = len(users)
found = False
while not found and i < nbusers:
found = user == users[i]["login"] and pbkdf2_sha256.verify(password, users[i]["password"])
i += 1
# Found?
if not found:
…
33.5. Tests
In addition to testing with a browser, you can also use the clients in the [http-servers/07] folder written for version 12. They should also work for version 13:
- in [1], all three clients should work;
- in [2], both tests should work;