33. Exercício prático: Versão 13
A versão 13 modifica a versão 12 das seguintes formas:
- Algumas partes do código foram reestruturadas (refatoração);
- a gestão de sessões é tratada de forma diferente utilizando o módulo [flask_session];
- são utilizadas palavras-passe encriptadas no ficheiro de configuração;
A versão 13 é inicialmente criada através da cópia da versão 12:


- em [1], a configuração será refatorada. Em particular, será removida da pasta [flask];
- em [2], o script principal será refatorado. Também será movido para fora da pasta [flask];
- em [3], a configuração será dividida em vários ficheiros;
- em [4]: simplificaremos o script principal [main] movendo código para outros ficheiros;
- em [5], o controlador de autenticação será modificado, uma vez que as palavras-passe dos utilizadores passarão a ser encriptadas;
- em [6], o controlador principal incorporará código anteriormente presente no script principal [main];
33.1. Reestruturação da configuração da aplicação

São criados três novos ficheiros de configuração:
- [mvc]: para configurar a arquitetura MVC da aplicação;
- [parameters]: que conterá todas as constantes da aplicação;
- [syspath]: que configura o caminho Python da aplicação;
O ficheiro [syspath.py] é o seguinte:
| def configure(config: dict) -> dict:
import os
# folder 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 files
# BaseEntity, MyException
f"{root_dir}/classes/02/entities",
# InterfaceImpôtsDao, InterfaceImpôtsMétier, InterfaceImpôtsUi
f"{root_dir}/impots/v04/interfaces",
# AbstractImpôtsdao, ImpôtsConsole, ImpôtsMétier
f"{root_dir}/impots/v04/services",
# ImpotsDaoWithAdminDataInDatabase
f"{root_dir}/impots/v05/services",
# AdminData, ImpôtsError, TaxPayer
f"{root_dir}/impots/v04/entities",
# Constants, slices
f"{root_dir}/impots/v05/entities",
# Logger, SendAdminMail
f"{root_dir}/impots/http-servers/02/utilities",
# main script folder
script_dir,
# configs [database, layers, parameters, controllers, views]
f"{script_dir}/../configs",
# controllers
f"{script_dir}/../controllers",
# answers HTTP
f"{script_dir}/../responses",
# view models
f"{script_dir}/../models_for_views",
]
# set the syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# we return the configuration
return {
"root_dir": root_dir,
"script_dir": script_dir
}
|
- O script [syspath] é utilizado para configurar o Python Path da aplicação (linhas 40–41);
- fornece duas informações úteis para os outros scripts de configuração (linhas 45–46);
O script [mvc] configura a arquitetura MVC da aplicação web JSON/XML/HTML:
| def configure(config: dict) -> dict:
# application configuration MVC
# controllers
from AfficherCalculImpotController import AfficherCalculImpotController
from AuthentifierUtilisateurController import AuthentifierUtilisateurController
from CalculerImpotController import CalculerImpotController
from CalculerImpotsController import CalculerImpotsController
from FinSessionController import FinSessionController
from GetAdminDataController import GetAdminDataController
from InitSessionController import InitSessionController
from ListerSimulationsController import ListerSimulationsController
from MainController import MainController
from SupprimerSimulationController import SupprimerSimulationController
# answers HTTP
from HtmlResponse import HtmlResponse
from JsonResponse import JsonResponse
from XmlResponse import XmlResponse
# view models
from ModelForAuthentificationView import ModelForAuthentificationView
from ModelForCalculImpotView import ModelForCalculImpotView
from ModelForErreursView import ModelForErreursView
from ModelForListeSimulationsView import ModelForListeSimulationsView
# authorized shares and their controllers
controllers = {
# initialization of a calculation session
"init-session": InitSessionController(),
# user authentication
"authentifier-utilisateur": AuthentifierUtilisateurController(),
# tax calculation in individual mode
"calculer-impot": CalculerImpotController(),
# batch mode tax calculation
"calculer-impots": CalculerImpotsController(),
# list of simulations
"lister-simulations": ListerSimulationsController(),
# deleting a simulation
"supprimer-simulation": SupprimerSimulationController(),
# end of calculation session
"fin-session": FinSessionController(),
# display tax calculation view
"afficher-calcul-impot": AfficherCalculImpotController(),
# obtaining data from tax authorities
"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 rendered by the controller
views = [
{
# authentication view
"états": [
700, # /init-session - success
201, # /authentifier-user failure
],
"view_name": "views/vue-authentification.html",
"model_for_view": ModelForAuthentificationView()
},
{
# tax calculation
"états": [
200, # /authentifier-user success
300, # /calculate-tax-success
301, # /calculate-tax failure
800, # /show-tax-calculation-success
],
"view_name": "views/vue-calcul-impot.html",
"model_for_view": ModelForCalculImpotView()
},
{
# view of simulation list
"états": [
500, # /lister-simulations success
600, # /suppress-simulation success
],
"view_name": "views/vue-liste-simulations.html",
"model_for_view": ModelForListeSimulationsView()
},
]
# view of unexpected errors
view_erreurs = {
"view_name": "views/vue-erreurs.html",
"model_for_view": ModelForErreursView()
}
# redirections
redirections = [
{
"états": [
400, # /end-session success
],
# redirection to URL
"to": "/init-session/html",
}
]
# return the MVC configuration
return {
# controllers
"controllers": controllers,
# answers HTTP
"responses": responses,
# views and models
"views": views,
# list of redirections
"redirections": redirections,
# view of unexpected errors
"view_erreurs": view_erreurs
}
|
- linhas 1-101: este código é conhecido;
- Linhas 105–116: renderizamos a configuração MVC da aplicação;
O script [parameters] recolhe as constantes da aplicação:
| def configure(config: dict) -> dict:
# application setup
# 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",
# server config SMTP
"adminMail": {
# server SMTP
"smtp-server": "localhost",
# server port SMTP
"smtp-port": "25",
# director
"from": "guest@localhost.com",
"to": "guest@localhost.com",
# mail subject
"subject": "plantage du serveur de calcul d'impôts",
# tls to True if server SMTP requires authorization, False otherwise
"tls": False
},
# thread pause time in seconds
"sleep_time": 0,
# redis server
"redis": {
"host": "127.0.0.1",
"port": 6379
},
}
# we return the application settings
return parameters
|
- linha 13: as senhas dos utilizadores serão agora encriptadas;
- linhas 34–38: configuração de um servidor [Redis], ao qual voltaremos mais tarde;
Com estes novos ficheiros de configuração, o script [config] fica da seguinte forma:
| def configure(config: dict) -> dict:
# syspath configuration
import syspath
config['syspath'] = syspath.configure(config)
# application setup
import parameters
config['parameters'] = parameters.configure(config)
# database configuration
import database
config["database"] = database.configure(config)
# instantiation of application layers
import layers
config['layers'] = layers.configure(config)
# configuration MVC of the [web] layer
import mvc
config['mvc'] = mvc.configure(config)
# we return the configuration
return config
|
33.2. Reestruturação do script principal [main]

O script principal [main.py] simplesmente inicia o servidor:
| # a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
sgbd = sys.argv[1].lower()
erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
# configure the application
import config
config = config.configure({'sgbd': sgbd})
# dependencies
from SendAdminMail import SendAdminMail
from Logger import Logger
from ImpôtsError import ImpôtsError
import redis
# send an e-mail to the administrator
def send_adminmail(config: dict, message: str):
# send an e-mail to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# check log file
logger = None
erreur = False
message_erreur = None
try:
# logger
logger = Logger(config['parameters']['logsFilename'])
except BaseException as exception:
# log console
print(f"L'erreur suivante s'est produite : {exception}")
# we note the error
erreur = True
message_erreur = f"{exception}"
# store the logger in the config
config['logger'] = logger
# error handling
if erreur:
# mail to administrator
send_adminmail(config, message_erreur)
# end of application
sys.exit(1)
# start-up log
log = "[serveur] démarrage du serveur"
logger.write(f"{log}\n")
# check 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 not available
log = f"[serveur] Le serveur Redis n'est pas disponible : {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# save client [redis] in config
config['redis_client'] = redis_client
# data recovery from tax authorities
erreur = False
try:
# admindata will be read-only application data
config["admindata"] = config["layers"]["dao"].get_admindata()
# success log
logger.write("[serveur] connexion à la base de données réussie\n")
except ImpôtsError as ex:
# we note the error
erreur = True
# error log
log = f"L'erreur suivante s'est produite : {ex}"
# console
print(log)
# log file
logger.write(f"{log}\n")
# mail to administrator
send_adminmail(config, log)
# the main thread no longer needs the logger
logger.close()
# if there has been an error, we stop
if erreur:
sys.exit(2)
# import routes from web application
import routes
routes.config=config
routes.execute(__name__)
|
- linhas 56–73: são apresentados um servidor Redis e o seu cliente;
- linhas 102–104: as rotas da aplicação foram externalizadas para o script [routes];
33.2.1. Os módulos [flask_session] e [redis]
O servidor [Redis] será utilizado para armazenar as sessões dos utilizadores. Utilizaremos o módulo [flask_session] para gerir estas sessões. Este módulo pode armazenar as sessões dos utilizadores em vários locais. O Redis é um deles, e é esse que iremos utilizar.
O módulo [flask_session] deve ser instalado num terminal do PyCharm:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install flask-session
Collecting flask-session
…
Para comunicar com o servidor Redis, precisamos de um cliente Redis. Este será fornecido pelo módulo [redis], que também iremos instalar:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install redis
Collecting redis
…
33.2.2. O servidor Redis
O servidor Redis será utilizado para armazenar as sessões dos utilizadores. O módulo [flask_session] funciona da seguinte forma:
- Cada utilizador tem um ID de sessão, e é isso que é enviado ao cliente — e apenas isso. O cliente recebe um cookie de sessão apenas uma vez, após o seu primeiro pedido. Este cookie contém o ID de sessão do utilizador, que não se alterará à medida que o cliente efetua pedidos subsequentes. É por isso que o servidor não precisa de enviar um novo cookie de sessão;
- Anteriormente, o conteúdo da sessão era enviado para o cliente. Isso já não acontecerá. O conteúdo da sessão do utilizador será armazenado no servidor Redis;
O Laragon vem com um servidor Redis que não está ativado por predefinição. Por isso, deve começar por ativá-lo:

- em [3], ative o servidor [Redis];
- em [4], mantenha a porta [6379], que os clientes Redis utilizam por predefinição;
Os serviços do Laragon são reiniciados automaticamente após o Redis ser ativado:

É possível consultar o servidor Redis no modo de comando. Abra um terminal do Laragon [6]:

- em [1], o comando [redis-cli] inicia o cliente no modo de comando para o servidor Redis;
Desde julho de 2019, o cliente Redis pode utilizar 172 comandos para interagir com o servidor [https://redis.io/commands#list]. Um deles [contagem de comandos] [2] exibe este número [3].
A gravação no [Redis] é feita utilizando o comando Redis [set atributo valor] [4]. O valor pode então ser recuperado utilizando o comando [get atributo] [5].
Pode ser necessário limpar a base de dados Redis. Isto é feito utilizando o comando [flushdb] [6]. Em seguida, se consultar o valor do atributo [title] [7], obtém uma referência [nil] [8] indicando que o atributo não foi encontrado. Também pode utilizar o comando [exists] [9-10] para verificar se um atributo existe.
Para sair do cliente Redis, digite o comando [quit] [11].
Também pode utilizar uma interface web para gerir as chaves no servidor Redis. Para tal, o servidor Apache do Laragon deve estar em execução:

Aparece a seguinte interface:

- em [1-4], uma das sessões armazenadas no servidor Redis;
33.2.3. Gerir o servidor Redis no script principal [main]
O script [main] verifica a presença do servidor Redis da seguinte forma:
| import redis
…
# check 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 not available
log = f"[serveur] Le serveur Redis n'est pas disponible : {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# save client [redis] in config
config['redis_client'] = redis_client
|
- linha 4: o construtor da classe [redis.Redis] cria um cliente do servidor Redis. As suas propriedades (endereço, porta) encontram-se no script [parameters];
- linha 8: o método [ping] verifica a presença do servidor Redis;
- linhas 9–17: se o ping falhar, o erro é registado e o servidor é parado;
- linha 20: a referência do cliente Redis é armazenada na configuração;
33.2.4. Gestão de rotas no script principal [main]
O tratamento de rotas em [main] limita-se às seguintes linhas:
| # import routes from web application
import routes
routes.config=config
routes.execute(__name__)
|
- linha 1: as rotas foram externalizadas para o módulo [routes];
- linha 3: as rotas precisam de conhecer a configuração de execução;
- linha 4: iniciamos a aplicação Flask passando-lhe o nome do script executado (__main__);
O script de rotas é o seguinte:
| # 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('/authentifier-utilisateur', methods=['POST'])
def authentifier_utilisateur() -> tuple:
# execute the controller associated with the action
return front_controller()
# calculate-tax
@app.route('/calculer-impot', methods=['POST'])
def calculer_impot() -> tuple:
# execute the controller associated with the action
return front_controller()
# batch tax calculation
@app.route('/calculer-impots', methods=['POST'])
def calculer_impots():
# execute the controller associated with the action
return front_controller()
# lister-simulations
@app.route('/lister-simulations', methods=['GET'])
def lister_simulations() -> tuple:
# execute the controller associated with the action
return front_controller()
# delete-simulation
@app.route('/supprimer-simulation/<int:numero>', methods=['GET'])
def supprimer_simulation(numero: int) -> tuple:
# execute the controller associated with the action
return front_controller()
# end of session
@app.route('/fin-session', methods=['GET'])
def fin_session() -> tuple:
# execute the controller associated with the action
return front_controller()
# display-calculation-tax
@app.route('/afficher-calcul-impot', methods=['GET'])
def afficher_calcul_impot() -> 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)
# if you launch the Flask application via a console script
if name == '__main__':
app.config.update(ENV="development", DEBUG=True)
app.run(threaded=True)
|
- linha 9: a aplicação Flask é instanciada;
- linha 12: a configuração da aplicação ainda não é conhecida quando o script é escrito. Só é conhecida no momento da execução;
- linhas 20–77: as rotas da aplicação, tal como definidas na versão anterior. Isto permanece inalterado;
- linhas 14–18: todas as rotas chamam simplesmente a função [front_controller]. Retirámos o código original desta função. Agora, ela chama simplesmente o controlador principal da aplicação web;
- linhas 79–89: [execute] é a função chamada pelo script [main] para iniciar a aplicação web;
- linha 81: o módulo [flask_session] utiliza a chave secreta do Flask;
- linhas 82–84: configuração do módulo [flask_session]. Isto envolve adicionar as chaves [SESSION_TYPE, SESSION_REDIS] à configuração [app.config] da aplicação Flask [app]:
- [SESSION_TYPE]: o tipo de sessão. Existem vários tipos. O tipo [redis] indica que o [flask_session] utiliza um servidor [redis] para armazenar as sessões dos utilizadores. Por isso, temos de definir a chave [SESSION_REDIS], que deve ser a referência de um cliente Redis;
- linha 85: o [Flask-Session] está associado à aplicação Flask;
- linhas 86–89: se o parâmetro [name] na linha 79 for a string [__main__], então a aplicação Flask é iniciada;
33.3. Reestruturação do controlador principal
O código que anteriormente se encontrava na função [front_controller] do script [main] foi movido para o controlador principal:

| # 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 e-mail to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# main application controller
class MainController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# we process the request
logger = None
action = None
type_response1 = None
try:
# path elements are retrieved
params = request.path.split('/')
# action is the 1st element
action = params[1]
# no errors at the start
erreur = False
# session type must be known prior to certain actions
type_response1 = session.get('typeResponse')
if type_response1 is None and action != "init-session":
# we note the error
résultat = {"action": action, "état": 101,
"réponse": ["pas de session en cours. Commencer par action [init-session]"]}
erreur = True
# logger
logger = Logger(config['parameters']['logsFilename'])
# we store it in a config 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] requête : {request}\n")
# the thread is interrupted if requested
sleep_time = config['parameters']['sleep_time']
if sleep_time != 0:
# pause is randomized so that some threads are interrupted and others not
aléa = randint(0, 1)
if aléa == 1:
# log before break
logger.write(f"[front_controller] mis en pause du thread pendant {sleep_time} seconde(s)\n")
# break
time.sleep(sleep_time)
# some actions require authentication
user = session.get('user')
if not erreur and user is None and action not in ["init-session", "authentifier-utilisateur"]:
# we note the error
résultat = {"action": action, "état": 101,
"réponse": [f"action [{action}] demandée par utilisateur non authentifié"]}
erreur = True
# are there any mistakes?
if erreur:
# the request is invalid
status_code = status.HTTP_400_BAD_REQUEST
else:
# execute the controller associated with the action
controller = config['mvc']['controllers'][action]
résultat, status_code = controller.execute(request, session, config)
except BaseException as exception:
# other (unexpected) exceptions
résultat = {"action": action, "état": 131, "réponse": [f"{exception}"]}
status_code = status.HTTP_400_BAD_REQUEST
finally:
pass
# we log the result sent to the customer
log = f"[MainController] {résultat}\n"
logger.write(log)
# was there a fatal error?
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# send an e-mail to the application administrator
send_adminmail(config, log)
# determine the desired type of response
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 type of response is known and in the session
type_response = type_response2
else:
# otherwise continue to use type_response1
type_response = type_response1
# build the response to be sent
response_builder = config['mvc']['responses'][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, résultat)
# close the log file if it has been opened
if logger:
logger.close()
# we send the HTTP response
return response, status_code
|
Todo este código já foi abordado em algum momento.
33.4. Tratamento de senhas encriptadas
Para lidar com senhas criptografadas, usaremos o módulo [passlib], que instalamos a partir de um terminal do PyCharm:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install passlib
Collecting passlib
…
Aqui está um exemplo de script que encripta a palavra-passe que lhe é passada como parâmetro:

O script [create_hashed_password] é o seguinte (https://passlib.readthedocs.io/en/stable/):
| import sys
# encryption function
from passlib.hash import pbkdf2_sha256
# wait for the password to be encrypted
syntaxe = f"{sys.argv[0]} password"
erreur = len(sys.argv) != 2
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
else:
password = sys.argv[1]
# password encryption
hash = pbkdf2_sha256.hash(password)
print(f"version cryptée de [{password}] = {hash}")
# check
correct = pbkdf2_sha256.verify(password, hash)
print(correct)
|
- linha 16: a palavra-passe passada como parâmetro é encriptada;
- linha 20: comparamos a palavra-passe [password] passada como parâmetro com a sua versão encriptada [hash]. A função [verify] encripta a palavra-passe [password] e compara a cadeia de caracteres encriptada resultante com [hash]. Retorna True se as duas cadeias de caracteres forem iguais;
O script acima permite-nos obter a versão encriptada da palavra-passe [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
version cryptée de [admin] = $pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek
True
Linha 2, o valor que colocamos no script [parameters]:
"users": [
{
"login": "admin",
"password": "$pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek"
}
],
O controlador de autenticação [AuthentifierUtilisateurController] evolui da seguinte forma:
| from passlib.handlers.pbkdf2 import pbkdf2_sha256
…
# check the validity of the (user, password) pair
users = config['parameters']['users']
i = 0
nbusers = len(users)
trouvé = False
while not trouvé and i < nbusers:
trouvé = user == users[i]["login"] and pbkdf2_sha256.verify(password, users[i]["password"])
i += 1
# found?
if not trouvé:
…
|
33.5. Testes
Para além de testar com um navegador, também pode utilizar os clientes na pasta [http-servers/07] criados para a versão 12. Estes também deverão funcionar na versão 13:

- em [1], os três clientes devem funcionar;
- em [2], ambos os testes devem funcionar;