23. Exercício prático: versão 6
23.1. Introdução
Voltamos agora à nossa aplicação de cálculo de impostos. Vamos construir, em torno dela, diferentes aplicações web.
Na versão 5 do nosso exercício prático, os dados da administração fiscal estavam armazenados numa base de dados. Esta versão 5 incluía duas aplicações distintas, mas com camadas em comum:
- uma aplicação que calculava o imposto em modo |batch| para contribuintes registados num ficheiro de texto;
- uma aplicação que calculava o imposto em modo |interativo| para contribuintes cujas informações eram introduzidas através do teclado;
A versão 5 da aplicação de cálculo do imposto em lote (modo batch) tinha a seguinte arquitetura:

Por fim, a versão web desta aplicação terá a seguinte arquitetura:

- o cliente web [1] comunica com o servidor web [2], que, por sua vez, comunica com os servidores SGBD e [3];
- o servidor web [2] mantém as camadas [métier], [8] e [dao], [9] da aplicação inicial;
- A aplicação inicial mantém o seu script principal [4] e as suas camadas [métier] e [15]. As camadas [métier], [8] e [15] são idênticas;
- a comunicação cliente/servidor requer duas camadas adicionais:
- a camada [web] [7], que implementa a aplicação web;
- a camada [dao] [5], cliente da aplicação web [7];
Na versão final, o cálculo do imposto em lotes poderá ser efetuado de duas formas:
- o cálculo funcional do imposto é efetuado pela camada [métier] do servidor. O script [main] utilizará este método;
- o cálculo do imposto é efetuado pela camada [métier] do cliente. O script [main2] utilizará este método;
A partir de agora, iremos desenvolver várias aplicações cliente/servidor do tipo acima referido, cada uma delas ilustrando uma ou mais novas tecnologias de desenvolvimento web.
23.2. O servidor web de cálculo do imposto
23.2.1. Versão 1

O script [server_01] é a seguinte aplicação web:

- no [1], utiliza-se um URL configurado, no qual se passam três valores:
- [marié] (sim/não) para indicar se o contribuinte é casado;
- [enfants]: o número de filhos do contribuinte;
- [salaire]: salário anual do contribuinte;
- em [2], o servidor web devolve uma cadeia jSON que indica o montante do imposto a pagar com as suas diferentes componentes;
A arquitetura da aplicação é a seguinte:

- o navegador [1] consulta o servidor [2]. O script [server_01] implementa a camada [web] [2] do servidor;
- as camadas [3-8] são as já utilizadas na |versão 5| da aplicação de cálculo de impostos. Retomamo-las tal como estão;
- a camada [métier] [3] está definida |aqui|;
- a camada [dao] [4] está definida |aqui|;
A aplicação web [server_01] é configurada através de três scripts:
- [config], que configura toda a aplicação;
- [config_database], que configura o acesso à base de dados. Iremos trabalhar com os SGBD, MySQL e PostgreSQL;
- [config_layers], que configura as camadas da aplicação;
O script [config] é o seguinte:
def configure(config: dict) -> dict:
import os
# passo 1 ------
# pasta deste ficheiro
script_dir = os.path.dirname(os.path.abspath(__file__))
# caminho raiz
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# dependências absolutas
absolute_dependencies = [
# pastas do projeto
# 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",
# Constantes, intervalos
f"{root_dir}/impots/v05/entities",
# IndexController
f"{script_dir}/../controllers",
# scripts [config_database, config_layers]
script_dir,
]
# definimos o syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# passo 2 ------
# configuração da aplicação
# lista de utilizadores autorizados a utilizar a aplicação
config['users'] = [
{
"login": "admin",
"password": "admin"
}
]
# passo 3 ------
# configuração da base de dados
import config_database
config["database"] = config_database.configure(config)
# etapa 4 ------
# instanciação das camadas da aplicação
import config_layers
config['layers'] = config_layers.configure(config)
# aplicamos a configuração
return config
- A função [configure] recebe um dicionário [config] como parâmetro (linha 1) e devolve-o como resultado (linha 54) após ter enriquecido o seu conteúdo. Já há muito que se poderia ter dito que não era necessário devolver o resultado [config]. Com efeito, [config] é uma referência ao dicionário que o código chamador partilha com o código chamado. O código chamador já possui, portanto, essa referência (linha 1) e é desnecessário voltar a fornecê-la (linha 54). Assim, escrever:
config=[module].configure(config) (1)
é redundante. Basta escrever:
[module].configure(config) (2)
No entanto, mantive o tipo (1) de escrita porque achei que talvez mostrasse melhor que o código chamado alterava o dicionário [config].
- linha 1: o dicionário [config] recebido pela função [configure] tem uma chave «sgbd» cujo valor é obtido da lista [‘mysql’, ‘pgres’]. [mysql] significa que a base de dados utilizada é gerida por MySQL, enquanto «pgres» significa que a base de dados utilizada é gerida por PostgreSQL;
- linhas 4-27: enumeram-se todas as pastas que contêm os elementos necessários para a aplicação web. Estas farão parte do Python Path da aplicação (linhas 30-31);
- linhas 33-40: só serão autorizados determinados utilizadores a aceder à aplicação. Aqui, temos uma lista com um único utilizador;
- linhas 43-46: é o script [config_database] que cria a configuração da base de dados utilizada;
- linha 46: a configuração criada pelo script [config_database] é um dicionário que é armazenado na configuração geral, associado à chave «database»;
- linhas 48-51: o script [config_layers] instancia as camadas da aplicação web. Este devolve um dicionário que é armazenado na configuração geral associada à chave «layers»;
O script [config_database] é o mesmo já utilizado na |versão 5|. Reproduzimo-lo aqui para referência:
def configure(config: dict) -> dict:
# configuração do SQLAlchemy
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# cadeias de ligação às bases de dados utilizadas
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# cadeia de ligação à base de dados utilizada
engine = create_engine(connection_strings[config['sgbd']])
# metadados
metadata = MetaData()
# a tabela de constantes
constantes_table = Table("tbconstantes", metadata,
Column('id', Integer, primary_key=True),
Column('plafond_qf_demi_part', Float, nullable=False),
Column('plafond_revenus_celibataire_pour_reduction', Float, nullable=False),
Column('plafond_revenus_couple_pour_reduction', Float, nullable=False),
Column('valeur_reduc_demi_part', Float, nullable=False),
Column('plafond_decote_celibataire', Float, nullable=False),
Column('plafond_decote_couple', Float, nullable=False),
Column('plafond_impot_celibataire_pour_decote', Float, nullable=False),
Column('plafond_impot_couple_pour_decote', Float, nullable=False),
Column('abattement_dixpourcent_max', Float, nullable=False),
Column('abattement_dixpourcent_min', Float, nullable=False)
)
# a tabela das faixas de imposto
tranches_table = Table("tbtranches", metadata,
Column('id', Integer, primary_key=True),
Column('limite', Float, nullable=False),
Column('coeffr', Float, nullable=False),
Column('coeffn', Float, nullable=False)
)
# os mapeamentos
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
# a fábrica de sessões
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# uma sessão
session = session_factory()
# registam-se determinadas informações e estas são devolvidas num dicionário
return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
O script [config_layers] configura as camadas do servidor web. Retomamos um |script| já mencionado:
def configure(config: dict) -> dict:
# instanciação das camadas da aplicação
# DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
dao = ImpotsDaoWithAdminDataInDatabase(config)
# logística
from ImpôtsMétier import ImpôtsMétier
métier = ImpôtsMétier()
# colocam-se as instâncias das camadas num dicionário que é devolvido ao código chamador
return {
"dao": dao,
"métier": métier
}
- linha 6: a camada [dao] é implementada com uma base de dados;
- [ImpotsDaoWithAdminDataInDatabase] foi definida |aqui|;
- [ImpôtsMétier] foi definida |aqui|;
O script principal [server_01] é o seguinte:
# espera-se um parâmetro mysql ou pgres
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()
# configura-se a aplicação
import config
config = config.configure({'sgbd': sgbd})
# dependências
from ImpôtsError import ImpôtsError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status
# recuperação dos dados da administração fiscal
try:
# «admindata» será um conjunto de dados de âmbito da aplicação, apenas para leitura
admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# aplicação Flask
app = Flask(__name__)
# Página inicial URL: /?casado=xx&filhos=yy&salário=zz
@app.route('/', methods=['GET'])
def index():
# Inicialmente, sem erros
erreurs = []
# a solicitação deve ter três parâmetros no URL
if len(request.args) != 3:
erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")
# recupera-se o estado civil no URL
marié = request.args.get('marié')
if marié is None:
erreurs.append("paramètre [marié] manquant")
else:
marié = marié.strip().lower()
erreur = marié != "oui" and marié != "non"
if erreur:
erreurs.append(f"paramétre marié [{marié}] invalide")
# o número de filhos é recuperado no URL
enfants = request.args.get('enfants')
if enfants is None:
erreurs.append("paramètre [enfants] manquant")
else:
enfants = enfants.strip()
match = re.match(r"^\d+", enfants)
if not match:
erreurs.append(f"paramétre enfants [{enfants}] invalide")
else:
enfants = int(enfants)
# recupera-se o salário no URL
salaire = request.args.get('salaire')
if salaire is None:
erreurs.append("paramètre [salaire] manquant")
else:
salaire = salaire.strip()
match = re.match(r"^\d+", salaire)
if not match:
erreurs.append(f"paramétre salaire [{salaire}] invalide")
else:
salaire = int(salaire)
# parâmetros inválidos no URL?
for key in request.args.keys():
if key not in ['marié', 'enfants', 'salaire']:
erreurs.append(f"paramètre [{key}] invalide")
# Erros?
if erreurs:
# envia-se uma resposta de erro ao cliente
résultats = {"réponse": {"erreurs": erreurs}}
return json_response(résultats, status.HTTP_400_BAD_REQUEST)
# sem erros, é possível continuar
# cálculo do imposto
taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, admindata)
# envia-se a resposta ao cliente
return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
# apenas em modo manual
if __name__ == '__main__':
# iniciamos o servidor Flask
app.config.update(ENV="development", DEBUG=True)
app.run()
- linhas 1-10: recupera-se o parâmetro que indica qual o SGBD a utilizar;
- linhas 12-14: com esta informação, é possível configurar a aplicação. Em particular, é criado o Python Path;
- linhas 16-23: com o novo Python Path, importam-se os elementos necessários;
- linhas 25-31: recuperam-se os dados da administração fiscal que permitem calcular o imposto;
- linhas 33-34: instanciamento da aplicação Flask;
- linha 38: a aplicação Flask apenas serve o URL [/]. Espera um URL configurado da seguinte forma [/ ?marié=xx&enfants=yy&salaire=zz] com:
- xx: sim / não;
- yy: número de filhos;
- zz: salário anual;
- linhas 40-89: verifica-se a validade dos parâmetros do URL;
- linha 41: acumulam-se as mensagens de erro na lista [erreurs];
- linha 43: talvez nos lembremos de que os parâmetros do URL configurado se encontram no [request.args] (ver |aqui|):
- o objeto [request] é o objeto Flask importado na linha 20;
- o objeto [request.args] comporta-se como um dicionário;
- linhas 43-44: verifica-se se existem exatamente três parâmetros (nem menos, nem mais);
- linhas 46-49: verifica-se se o parâmetro [marié] está presente no URL;
- linhas 50-54: se estiver presente, verifica-se se o seu valor em minúsculas, sem os «espaços» no início e no fim, é «sim» ou «não»;
- linhas 56-59: verifica-se se o parâmetro [enfants] está presente no URL;
- linhas 60-66: se estiver presente, verifica-se se o seu valor é um número inteiro positivo;
- linha 66: não se deve esquecer que os parâmetros do URL e os seus valores são cadeias de caracteres. O valor do parâmetro [enfants] é convertido para «int»;
- linhas 68-78: para o parâmetro [salaire], realizam-se os mesmos testes que para o parâmetro [enfants];
- linhas 81-83: verifica-se se não existem outros parâmetros além de [‘marié, ‘enfants’, ‘salaire’] no URL;
- linhas 85-89: se, após todas estas verificações, a lista [erreurs] não estiver vazia, enviamos essa lista de erros ao cliente sob a forma de uma cadeia jSON e do código de estado [400 Bad Request];
Como, posteriormente, teremos frequentemente a oportunidade de enviar uma cadeia jSON em resposta ao cliente, as poucas linhas necessárias para esse envio foram agrupadas no módulo [myutils.py] que já utilizámos:

O script [myutils.py] passa a ser o seguinte:
# importações
import json
import os
import sys
from flask import make_response
def set_syspath(absolute_dependencies: list):
# absolute_dependencies: uma lista de nomes absolutos de pastas
….
# geração de uma resposta HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
# corpo da resposta HTTP
response = make_response(json.dumps(réponse, ensure_ascii=False))
# corpo da resposta HTTP é do jSON
response.headers['Content-Type'] = 'application/json; charset=utf-8'
# envia-se a resposta HTTP
return response, status_code
- linha 16: a função [json_response] espera dois parâmetros:
- [réponse]: o dicionário cuja cadeia jSON deve ser enviada ao cliente web;
- [status_code]: o código de estado HTTP da resposta;
- linha 18: define-se o corpo jSON da resposta;
- linha 20: adiciona-se o cabeçalho HTTP, que indica ao cliente web que irá receber o jSON;
- linha 22: envia-se a resposta HTTP ao código chamador. Cabe a este enviá-la ao cliente web;
O ficheiro [__init__.py] evolui da seguinte forma:
from .myutils import set_syspath, json_response
A nova versão do [myutils] é instalada entre os módulos de âmbito da máquina com o comando [pip install .] num terminal do Pycharm:
(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 package 'wheel' is not installed.
Installing collected packages: myutils
Attempting 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
- linha 1: é necessário estar na pasta [packages] para introduzir esta instrução;
O código do script [server_01] continua da seguinte forma:
…
# erros?
if erreurs:
# envia-se uma resposta de erro ao cliente
résultats = {"réponse": {"erreurs": erreurs}}
return json_response(résultats, status.HTTP_400_BAD_REQUEST)
# sem erros, é possível prosseguir
# cálculo do imposto
taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, admindata)
# envia-se a resposta ao cliente
return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
- linha 10: quando se está aqui, os parâmetros esperados no URL estão presentes e estão corretos;
- linha 10: cria-se o objeto [TaxPayer] que modela o contribuinte;
- linha 11: solicita-se à camada [métier] que calcule o imposto. Recorde-se que os elementos calculados pela camada [métier] são inseridos no objeto [taxpayer] passado como parâmetro;
- linha 13: a resposta é enviada ao cliente web sob a forma de uma cadeia jSON. Trata-se da cadeia jSON de um dicionário. Associado à chave [result], insere-se aí o dicionário do objeto [taxpayer]. Não foi possível inserir o próprio objeto [taxpayer], uma vez que este não é serializável em jSON;
Criam-se duas configurações de execução, uma para o MySQL e outra para o PostgreSQL:

Eis alguns exemplos de execução (iniciou a aplicação [server_01] e utilizou o SGBD; em seguida, acedeu ao URL http://localhost:5000/ através de um navegador):


Eis um exemplo de execução na consola do Postman:

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, 17 Jul 2020 06:15:44 GMT
{"réponse": {"erreurs": ["paramètre marié [xx] invalide", "paramètre enfants [yy] invalide", "paramètre salaire [zz] invalide"]}}
- linha 1: é solicitada uma URL incorreta;
- linha 10: o servidor responde com o estado 400 BAD REQUEST;
23.2.2. Versão 2

A versão 2 do servidor isola o processamento do URL no módulo [index_controller] [5]:
# importação das dependências
import re
from flask_api import status
from werkzeug.local import LocalProxy
# URL com os seguintes parâmetros: /?casado=xx&filhos=yy&salário=zz
def execute(request: LocalProxy, config: dict) -> tuple:
# dependentes
from TaxPayer import TaxPayer
# inicialmente sem erros
erreurs = []
# a consulta deve ter três parâmetros
if len(request.args) != 3:
erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")
# recupera-se o estado civil do URL
marié = request.args.get('marié')
if marié is None:
erreurs.append("paramètre [marié] manquant")
else:
marié = marié.strip().lower()
erreur = marié != "oui" and marié != "non"
if erreur:
erreurs.append(f"paramétre marié [{marié}] invalide")
# recupera-se o número de filhos do URL
enfants = request.args.get('enfants')
if enfants is None:
erreurs.append("paramètre [enfants] manquant")
else:
enfants = enfants.strip()
match = re.match(r"^\d+", enfants)
if not match:
erreurs.append(f"paramétre enfants {enfants} invalide")
else:
enfants = int(enfants)
# recupera-se o salário do URL
salaire = request.args.get('salaire')
if salaire is None:
erreurs.append("paramètre [salaire] manquant")
else:
salaire = salaire.strip()
match = re.match(r"^\d+", salaire)
if not match:
erreurs.append(f"paramétre salaire {salaire} invalide")
else:
salaire = int(salaire)
# Existem outros parâmetros no URL?
for key in request.args.keys():
if not key in ['marié', 'enfants', 'salaire']:
erreurs.append(f"paramètre [{key}] invalide")
# erros?
if erreurs:
# envia-se uma resposta de erro ao cliente
résultats = {"réponse": {"erreurs": erreurs}}
return résultats, status.HTTP_400_BAD_REQUEST
# sem erros, é possível prosseguir
# cálculo do imposto
taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
# envia-se a resposta ao cliente
return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
- linha 9: a função [execute] recebe dois parâmetros:
- [request]: a solicitação HTTP do cliente;
- [config]: o dicionário de configuração da aplicação;
O script [server_02] é o seguinte:
# aguarda-se um parâmetro mysql ou pgres
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()
# configurar a aplicação
import config
config = config.configure({'sgbd': sgbd})
# dependências
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller
# recuperação de dados da administração fiscal
try:
# «admindata» será um dado de âmbito da aplicação, apenas para leitura
config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# aplicação Flask
app = Flask(__name__)
# Página inicial URL: /?casado=xx&filho=yy&salário=zz
@app.route('/', methods=['GET'])
def index():
# executa-se a consulta
résultat, statusCode = index_controller.execute(request, config)
# envia-se a resposta
return json_response(résultat, statusCode)
# apenas main
if __name__ == '__main__':
# inicia-se o servidor
app.config.update(ENV="development", DEBUG=True)
app.run()
- linhas 36-41: processamento da rota /;
- linha 39: utilização da função [IndexController.execute];
Passaremos a utilizar esta técnica: cada rota será processada por um módulo específico.
Os resultados da execução são os mesmos que na versão 1.
23.2.3. Versão 3

A versão 3 introduz o conceito de autenticação.
O script [server_03] passa a ser o seguinte:
# aguarda um parâmetro mysql ou pgres
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()
# configura-se a aplicação
import config
config = config.configure({'sgbd': sgbd})
# dependências
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
# recuperação de dados da administração fiscal
try:
# config[‘admindata’] será um dado de âmbito da aplicação, apenas para leitura
config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# gestor de autenticação
auth = HTTPBasicAuth()
# método de autenticação
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
# lista de utilizadores
users = config['users']
# percorre-se esta lista
for user in users:
if user['login'] == login and user['password'] == password:
return True
# não foi encontrado
return False
# aplicação Flask
app = Flask(__name__)
# Página inicial URL: /?casado=xx&filho=yy&salário=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
# executa-se a consulta
résultat, statusCode = index_controller.execute(request, config)
# envia-se a resposta
return json_response(résultat, statusCode)
# apenas main
if __name__ == '__main__':
# inicia-se o servidor
app.config.update(ENV="development", DEBUG=True)
app.run()
- linha 21: importa-se um gestor de autenticação. Existem vários tipos de autenticação num servidor web. O que utilizamos aqui chama-se [HTTP Basic]. Cada tipo de autenticação segue um diálogo cliente/servidor específico;
- linha 33: cria-se uma instância do gestor de autenticação;
- linha 37: a anotação [@auth.verify_password] identifica a função a executar quando o gestor de autenticação pretende verificar o nome de utilizador e a palavra-passe enviados pelo cliente, de acordo com o protocolo [HTTP Basic];
- linha 55: a anotação [@auth.login_required] identifica uma rota para a qual o cliente web deve ser autenticado. Se o cliente web ainda não tiver enviado as suas credenciais, o servidor web irá solicitá-las automaticamente de acordo com o protocolo HTTP basic;
O módulo [flask_httpauth] deve estar instalado:
(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
Vamos ver o que acontece com a consola do Postman. Deve:
- crie uma configuração de execução;
- inicie a aplicação web;
- inicia o SGBD da sua escolha;
- solicite o URL [/] com o Postman;
O diálogo cliente/servidor na consola do Postman é o seguinte:
- linha 10: o servidor responde que não estamos autorizados a aceder ao URL [/];
- linha 13: indica-nos o protocolo de autenticação a utilizar, neste caso o protocolo denominado «Autenticação Básica»;
É possível configurar o Postman para que envie as credenciais do utilizador de acordo com o protocolo Auth Basic:

- em [6-7], inserimos as credenciais presentes no script [config]:
config['users'] = [
{
"login": "admin",
"password": "admin"
}
]
O diálogo cliente/servidor na consola do Postman passa a ser o seguinte:
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
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
- linha 2: o cliente Postman envia, de forma codificada, os identificadores do utilizador [admin / admin];
- linha 17: o servidor responde corretamente. Indica erros porque não foram enviados os parâmetros [marié, enfants, salaire] (linha 1), mas não indica qualquer erro de autenticação;
Agora, vamos solicitar o URL / com um navegador (Firefox, abaixo):

- tal como no Postman, o Firefox recebeu a resposta HTTP do servidor com os cabeçalhos HTTP:
O Firefox, tal como outros navegadores, não interrompe a sessão quando recebe estes cabeçalhos. Solicita ao utilizador as credenciais exigidas pelo servidor. Basta, no exemplo acima, digitar admin / admin para receber a resposta do servidor:

23.3. O cliente web do servidor de cálculo de impostos
23.3.1. Introdução
No parágrafo anterior, o cliente web do servidor de cálculo de impostos era um navegador. Nesta parte, o cliente web será um script de consola. A arquitetura passa a ser a seguinte:

- o cliente web é composto pelas camadas [1-2];
- o servidor web é composto pelas camadas [3-9]. Como foi referido no parágrafo anterior;
Portanto, temos de escrever as camadas [1-2].
A camada [dao] [2] tem de ser capaz de comunicar com o servidor web [3]. Já conhecemos o protocolo HTTP e poderíamos escrever, utilizando, por exemplo, o módulo [pycurl] já estudado, um script que comunique com o servidor web [3]. No entanto, existem módulos especializados em diálogos cliente/servidor HTTP. Vamos utilizar um deles, o módulo [requests]:
(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
A estrutura dos scripts do cliente web é a seguinte:

O script irá implementar a aplicação de cálculo do imposto em modo batch descrita desde a |versão 1|. A última versão desta aplicação é a |versão 5|. Recorde-se o seu funcionamento:
- os contribuintes cujo imposto será calculado estão reunidos no ficheiro de texto [taxpayersdata.txt]:
- os resultados são guardados em dois ficheiros:
- o ficheiro de texto [errors.txt] reúne os erros detetados no ficheiro dos contribuintes:
Analyse du fichier C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt
Ligne 15, not enough values to unpack (expected 4, got 2)
Ligne 17, MyException[1, L'identifiant d'une entité <class 'TaxPayer.TaxPayer'> doit être un entier >=0]
- (continuação)
- O ficheiro jSON [résultats.json] reúne os resultados dos cálculos do imposto dos diferentes contribuintes:
[
{
"id": 0,
"marié": "oui",
"enfants": 2,
"salaire": 55555,
"impôt": 2814,
"surcôte": 0,
"taux": 0.14,
"décôte": 0,
"réduction": 0
},
{
"id": 1,
"marié": "oui",
"enfants": 2,
"salaire": 50000,
"impôt": 1384,
"surcôte": 0,
"taux": 0.14,
"décôte": 384,
"réduction": 347
},
…
]
23.3.2. Configuração do cliente web

A configuração é efetuada através de dois scripts:
- [config], que assegura toda a configuração fora das camadas da arquitetura;
- [config_layers], que assegura a configuração das camadas da arquitetura;
O script [config] é o seguinte:
def configure(config: dict) -> dict:
import os
# etapa 1 ------
# pasta deste ficheiro
script_dir = os.path.dirname(os.path.abspath(__file__))
# caminho raiz
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# dependências absolutas
absolute_dependencies = [
# pastas do projeto
# 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",
# Constantes, intervalos
f"{root_dir}/impots/v05/entities",
# ImpôtsDaoWithHttpClient
f"{script_dir}/../services",
# scripts de configuração
script_dir,
]
# definir o syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# etapa 2 ------
# configuração da aplicação com constantes
config.update({
"taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
"resultsFilename": f"{script_dir}/../data/output/résultats.json",
"errorsFilename": f"{script_dir}/../data/output/errors.txt",
"server": {
"urlServer": "http://127.0.0.1:5000/",
"authBasic": True,
"user": {
"login": "admin",
"password": "admin"
}
}
}
)
# etapa 3 ------
# instanciação das camadas
import config_layers
config['layers'] = config_layers.configure(config)
# aplicamos a configuração
return config
- linha 1: a função [configure] recebe como parâmetro o dicionário a preencher com as informações de configuração. Este pode já estar pré-preenchido ou vazio. Neste caso, estará vazio;
- linhas 40-42: os nomes absolutos dos três ficheiros de texto geridos pela camada [dao];
- linhas 43-50: associadas à chave [server], as informações que a camada [dao] deve conhecer sobre o servidor web com o qual deve comunicar:
- linha 44: o URL do serviço web;
- linha 45: a chave [authBasic] tem o valor True se o acesso ao URL exigir uma autenticação do tipo Basic;
- linhas 46-49: os identificadores do utilizador que se irá autenticar caso a autenticação seja solicitada;
- linhas 56-57: instanciamos as camadas, neste caso a única camada [dao], e colocamos as referências das camadas em [config] associadas à chave [layers];
O script [config_layers] é o seguinte:
def configure(config: dict) -> dict:
# instanciação das camadas da aplicação
# camada DAO
from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
dao = ImpôtsDaoWithHttpClient(config)
# efetua-se a configuração das camadas
return {
"dao": dao
}
- linha 1: a função [configure] recebe o dicionário que configura a aplicação;
- linhas 4-6: a camada [dao] é instanciada. Na linha 6, é-lhe passada a configuração da aplicação, na qual encontrará as informações de que necessita;
- linhas 8-11: é devolvido um dicionário no qual foi colocada a referência da camada [dao];
23.3.3. O script principal [main]
O script principal [main] é uma variante do da |versão 5|:
# configura-se a aplicação
import config
config = config.configure({})
# dependências
from ImpôtsError import ImpôtsError
# código
try:
# recuperação da camada [dao]
dao = config["layers"]["dao"]
# leitura dos dados dos contribuintes
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# dos contribuintes?
if not taxpayers:
raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# cálculo do imposto dos contribuintes
for taxpayer in taxpayers:
# o contribuinte é simultaneamente um parâmetro de entrada e de saída
# o «contribuinte» vai ser alterado
dao.calculate_tax(taxpayer)
# gravação dos resultados num ficheiro de texto
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# exibição do erro
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# concluído
print("Travail terminé...")
- linhas 2-3: a aplicação está configurada;
- linha 13: a camada [dao] fornece a lista dos contribuintes cujo imposto deve ser calculado;
- linha 21: a camada [dao] calcula o imposto de cada um deles;
- linha 23: os resultados são gravados num ficheiro jSON;
23.3.4. Implementação da camada [dao]

Voltemos à arquitetura cliente/servidor utilizada:

- em [2, 6], vemos que a camada [dao] tem duas funções:
- acede ao sistema de ficheiros tanto para ler os dados dos contribuintes como para gravar os resultados dos cálculos do imposto. Já dispomos de uma classe |AbstractImpôtsDao| capaz de realizar essas tarefas. Esta classe tem vindo a ser utilizada desde a |versão 4|;
- ela comunica com o servidor web [3];
Na |versão 5|, o script principal [main] [1] comunicava diretamente com a camada [métier] [4]. Gostaríamos de não alterar este script. Para tal, vamos garantir que a camada [dao] [2] implemente a interface da camada [métier] [4]. Desta forma, o script principal [main] terá a impressão de comunicar diretamente com a camada [métier] [4] e poderá ignorar completamente o facto de esta se encontrar noutro computador.
Uma definição da classe que implementa a camada [dao] [2] poderia ser a seguinte:
class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
- a classe [ImpôtsDaoWithHttpClient]:
- herda da classe [AbstractImpôtsDao], o que lhe permitirá gerir a interação com o sistema de ficheiros [6];
- implementa a interface [InterfaceImpôtsMétier] para não ter de alterar o script principal [main] da |versão 5|;
O código completo da classe [ImpôtsDaoWithHttpClient] é o seguinte:
# importações
import requests
from flask_api import status
from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from ImpôtsError import ImpôtsError
from InterfaceImpôtsMétier import InterfaceImpôtsMétier
from TaxPayer import TaxPayer
class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
# construtor
def __init__(self, config: dict):
# inicialização do pai
AbstractImpôtsDao.__init__(self, config)
# armazenamento de parâmetros
self.__config_server = config["server"]
# método não utilizado de [AbstractImpôtsDao]
def get_admindata(self) -> AdminData:
pass
# cálculo do imposto
def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
# deixa-se que as exceções sejam propagadas
# parâmetros do get
params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
# ligação com autenticação Auth Basic?
if self.__config_server['authBasic']:
response = requests.get(
# URL do servidor consultado
self.__config_server['urlServer'],
# parâmetros do URL
params=params,
# autenticação Basic
auth=(
self.__config_server["user"]["login"],
self.__config_server["user"]["password"]))
else:
# ligação sem autenticação Auth Basic
response = requests.get(self.__config_server['urlServer'], params=params)
# verificação
print(response.text)
# código de estado da resposta HTTP
status_code = response.status_code
# a resposta jSON é inserida num dicionário
résultat = response.json()
# erro se o código de estado for diferente de 200 OK
if status_code != status.HTTP_200_OK:
# sabe-se que os erros foram associados à chave [erreurs] da resposta
raise ImpôtsError(87, résultat['réponse']['erreurs'])
# sabe-se que o resultado foi associado à chave [result] da resposta
# altera-se o parâmetro de entrada com este resultado
taxpayer.fromdict(résultat["réponse"]["result"])
- linhas 21-23: a classe [AbstractImpôtsDao] (linha 12) possui um método abstrato [get_admindata]. Somos obrigados a implementá-lo, mesmo que não o utilizemos (o admindata é gerido pelo servidor e não pelo cliente);
- linha 26: o método [calculate_tax] pertence à interface [InterfaceImpôtsMétier] (linha 12). Temos de o implementar;
- linha 15: o construtor recebe como único parâmetro o dicionário da configuração da aplicação;
- linhas 16-17: a classe pai [AbstractImpôtsDao] é inicializada passando-lhe, também aqui, a configuração da aplicação. Nela encontrará os nomes dos três ficheiros de texto que tem de gerir;
- linhas 18-19: as informações relativas ao servidor web de cálculo do imposto são armazenadas localmente na classe;
- linha 26: o método [calculate_tax] recebe como parâmetro um objeto do tipo |Taxpayer|. Para respeitar a assinatura do método [InterfaceImpôtsMétier.calculate_tax], recebe também um parâmetro [admindata], que se destina a encapsular os dados da administração fiscal. Do lado do cliente, não dispomos desses dados. Este parâmetro permanecerá sempre como [None]. Esta contorção leva a concluir que a classe [ImpôtsMétier] foi inicialmente mal escrita:
- a assinatura de [calculate_tax] deveria ter sido simplesmente:
def calculate_tax(self, taxpayer: TaxPayer)
e o parâmetro [admindata : AdminData] deveria ter sido passado ao construtor da classe;
- linha 27: o código do método [calculate_tax] não foi encapsulado num try / catch / finally. Isto significa que eventuais exceções não serão tratadas e serão propagadas para o código chamador, neste caso o script [main]. Este último intercepta, de facto, todas as exceções que sobem da camada [dao];
- linha 28: o cálculo do imposto é feito no lado do servidor. Por isso, será necessário comunicar com ele. Faz-se isso através do módulo [requests] importado na linha 2;
- linhas 31-43: para enviar um pedido GET ao servidor web, utiliza-se o método [requests.get]:
- linhas 33-34: o primeiro parâmetro do método é o URL a contactar;
- linhas 35-40: os outros dois parâmetros são parâmetros nomeados, cuja ordem não importa;
- linhas 35-36: o valor do parâmetro denominado [params] deve ser um dicionário contendo as informações a incluir no URL na forma [/url ?param1=valeur1¶m2=valeur2&…];
- linha 29: o dicionário que contém os três parâmetros [marié, enfants, salaire] que o servidor web espera. Não é necessário preocupar-se com a codificação (denominada urlencoded) a que estes parâmetros devem ser submetidos. O [requests] encarrega-se disso;
- linhas 37-40: o parâmetro denominado [auth] é uma tupla de dois elementos (login, password). Representa os dados de identificação de uma autenticação do tipo Basic;
- linhas 44-45: estas duas linhas têm apenas um objetivo didático (serão colocadas em comentários quando a depuração estiver concluída):
- [response] representa a resposta HTTP do servidor;
- [response.text] representa o texto do documento encapsulado nesta resposta. Durante a fase de depuração, é útil verificar o que o servidor nos enviou;
- linha 47: [response.status_code] é o código de estado HTTP da resposta recebida. O nosso servidor envia apenas três:
- 200 OK
- 400 BAD REQUEST
- 500 INTERNAL SERVER ERROR
- linha 49: o nosso servidor envia sempre jSON, mesmo em caso de erro. A função [response.json()] cria um dicionário a partir da cadeia jSON recebida. Recorde-se que existem duas formas possíveis para a cadeia jSON:
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
{"réponse": {"result": {"id": 0, "marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
- linhas 51-53: se o código de estado não for 200, é lançada uma exceção com as mensagens de erro encapsuladas na resposta;
- linha 56: recupera-se o dicionário gerado pelo cálculo do imposto e utiliza-se para atualizar o parâmetro de entrada [taxpayer];
23.3.5. Execução
Para executar o cliente:
- inicie o servidor [server_03] com o SGBD da sua escolha;
- execute o script [main] do cliente;
Os resultados estarão na pasta [data/output]. São os mesmos que para a versão 5.
23.4. Testes da camada [dao]
Voltemos à arquitetura da aplicação cliente/servidor:
- No cliente escrito, conseguimos fazer com que a camada [dao] [1] ofereça a mesma interface que a camada [métier] [3]. Por isso, vamos utilizar, na camada [4], a classe de testes |TestDaoMétier| já analisada para testar a camada [métier] [3];
A classe de teste será executada no seguinte ambiente:

- a configuração [2] é idêntica à configuração [1] que acabámos de analisar;
A classe de teste [TestHttpClientDao] é a seguinte:
import unittest
class TestHttpClientDao(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'casado': 'sim', 'filhos': 2, 'salário': 55555,
# 'imposto': 2814, 'majoração': 0, 'redução': 0, 'abatimento': 0, 'taxa': 0,14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
dao.calculate_tax(taxpayer)
# verificação
self.assertAlmostEqual(taxpayer.impôt, 2815, delta=1)
self.assertEqual(taxpayer.décôte, 0)
self.assertEqual(taxpayer.réduction, 0)
self.assertAlmostEqual(taxpayer.taux, 0.14, delta=0.01)
self.assertEqual(taxpayer.surcôte, 0)
…
def test_11(self) -> None:
from TaxPayer import TaxPayer
# {'casado': 'sim', 'filhos': 3, 'salário': 200000,
# 'imposto': 42842, 'sobretaxa': 17283, 'desconto': 0, 'redução': 0, 'taxa': 0,41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
dao.calculate_tax(taxpayer)
# verificações
self.assertAlmostEqual(taxpayer.impôt, 42842, 1)
self.assertEqual(taxpayer.décôte, 0)
self.assertEqual(taxpayer.réduction, 0)
self.assertAlmostEqual(taxpayer.taux, 0.41, delta=0.01)
self.assertAlmostEqual(taxpayer.surcôte, 17283, delta=1)
if __name__ == '__main__':
# configura-se a aplicação
import config
config = config.configure({})
# camada DAO
dao = config['layers']['dao']
# executam-se os métodos de teste
print("tests en cours...")
unittest.main()
Esta classe é análoga à |aquela| já analisada na versão 4 da aplicação.
- linhas 40-41: configura-se o ambiente de testes;
- linha 44: obtém-se uma referência à camada [dao];
- linhas 47-48: executam-se os testes;
Para executar os testes, cria-se uma |configuração de execução|:

- cria-se uma configuração de execução para um script de consola, e não para um teste UnitTest;
Ao executar esta configuração, obtêm-se os seguintes resultados:
Os 11 testes foram bem-sucedidos.