20. Exercício prático: versão 5
Vamos desenvolver três aplicações:
- a aplicação 1 irá inicializar a base de dados que substituirá o ficheiro [admindata.json] da versão 4;
- a aplicação 2 fará o cálculo dos impostos em modo batch;
- a aplicação 3 fará o cálculo dos impostos em modo interativo;
20.1. Aplicação 1: inicialização da base de dados
A aplicação 1 terá a seguinte arquitetura:
Trata-se de uma evolução da arquitetura da versão 4 (parágrafo |Versão 4|): os dados fiscais serão encontrados numa base de dados, em vez de se encontrarem num ficheiro jSON. A camada [dao] será atualizada para implementar esta alteração.
20.1.1. O ficheiro [admindata.json]
O ficheiro [admindata.json] é o mesmo que existia na versão 4:
{
"limites": [9964, 27519, 73779, 156244, 0],
"coeffr": [0, 0.14, 0.3, 0.41, 0.45],
"coeffn": [0, 1394.96, 5798, 13913.69, 20163.45],
"plafond_qf_demi_part": 1551,
"plafond_revenus_celibataire_pour_reduction": 21037,
"plafond_revenus_couple_pour_reduction": 42074,
"valeur_reduc_demi_part": 3797,
"plafond_decote_celibataire": 1196,
"plafond_decote_couple": 1970,
"plafond_impot_couple_pour_decote": 2627,
"plafond_impot_celibataire_pour_decote": 1595,
"abattement_dixpourcent_max": 12502,
"abattement_dixpourcent_min": 437
}
Vamos utilizar como colunas da base de dados as chaves deste dicionário.
20.1.2. Criação das bases de dados
Tal como foi demonstrado no parágrafo |criação de uma base de dados MySQL|, criamos uma base de dados MySQL denominada [dbimpots-2019], propriedade do utilizador [admimpots], com a palavra-passe [mdpimpots]. Em [phpMyAdmin], o resultado é o seguinte:
Da mesma forma, tal como foi demonstrado no parágrafo |criação de uma base de dados PostgreSQL|, criamos uma base de dados PostgreSQL denominada [dbimpots-2019], propriedade do utilizador [admimpots], com a palavra-passe [mdpimpots]. Em [pgAdmin], o resultado é o seguinte:
As bases de dados são criadas, mas, por enquanto, não têm nenhuma tabela. Estas serão criadas pelo ORM e pelo [sqlalchemy].
20.1.3. As entidades mapeadas pelo [sqlalchemy]
Vamos criar duas tabelas para encapsular os dados do [admindata.json]:
Definida por [sqlalchemy], a tabela [tbtranches] reunirá os dados das tabelas [limites, coeffr, coeffn] do dicionário [admindata.json]:
# a tabela de escalões 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)
)
Definida pela tabela [sqlalchemy], a tabela [tbconstantes] irá reunir as constantes do dicionário [admindata.json]:
# 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)
)
As entidades que serão mapeadas com estas duas tabelas serão as seguintes:
A entidade [Constantes] encapsula as constantes do dicionário [admindata.json]:
from BaseEntity import BaseEntity
# classe-recipiente dos dados da administração fiscal
class Constantes(BaseEntity):
# chaves excluídas do estado da classe
excluded_keys = ["_sa_instance_state"]
# chaves autorizadas
@staticmethod
def get_allowed_keys() -> list:
return ["id",
"plafond_qf_demi_part",
"plafond_revenus_celibataire_pour_reduction",
"plafond_revenus_couple_pour_reduction",
"valeur_reduc_demi_part",
"plafond_decote_celibataire",
"plafond_decote_couple",
"plafond_decote_couple",
"plafond_impot_celibataire_pour_decote",
"plafond_impot_couple_pour_decote",
"abattement_dixpourcent_max",
"abattement_dixpourcent_min"]
- linha 5: a classe [Constantes] estende a classe [BaseEntity];
- linha 7: através do mapeamento [sqlalchemy], a classe [Constante] irá receber a propriedade [_sa_instance_state]. Excluímos-a do dicionário [asdict] da entidade;
- linhas 11-23: as propriedades da entidade. Retomámos os nomes utilizados no dicionário [admindata.json] para facilitar a escrita do código;
A entidade [Tranche] encapsula uma linha das três tabelas [limites, coeffr, coeffn] do dicionário [admindata.json]:
from BaseEntity import BaseEntity
# classe-recipiente dos dados da administração fiscal
class Tranche(BaseEntity):
# chaves excluídas do estado da classe
excluded_keys = ["_sa_instance_state"]
# chaves autorizadas
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
- linha 5: a classe [Tranche] estende a classe [BaseEntity];
- linha 7: exclui-se das propriedades do dicionário [asdict] da entidade a propriedade [_sa_instance_state] adicionada por [sqlalchemy];
- linhas 10-12: as propriedades da classe;
O mapeamento entre as entidades [Constantes, Tranche] e as tabelas [constantes, tranches] será o seguinte:
…
# 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)
- os mapeamentos são definidos nas linhas 24-29. Nestas, omitiu-se estabelecer as correspondências entre as propriedades das entidades mapeadas e as tabelas da base de dados. Tal é possível quando os nomes das colunas das tabelas são idênticos aos das propriedades às quais devem ser associadas. Por este motivo, incluímos nas tabelas os nomes das propriedades das entidades mapeadas. Isto facilita a escrita do código e a sua compreensão;
20.1.4. O ficheiro de configuração do [sqlalchemy]
Acabámos de detalhar uma parte da configuração do [sqlalchemy]. O ficheiro [config_database] na sua totalidade é o seguinte:
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
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# resultado
return config
- linha 1: a função [configure] recebe como parâmetro um dicionário cuja chave [sgbd] indica-lhe qual o SGBD a utilizar: MySQL (mysql) ou PostgreSQL (pgres);
- linhas 6-12: seleciona-se a base de dados solicitada pela configuração;
- linhas 14-44: mapeamentos de entidades/tabelas. Estes mapeamentos são simples, pois não existe qualquer ligação entre as tabelas [tranches] e [constantes]. São independentes. Não há, portanto, nenhuma chave estrangeira de uma para a outra para gerir;
- linhas 46-51: cria-se a sessão de trabalho [session] da aplicação;
- linhas 53-58: as informações úteis são inseridas no dicionário de configuração, que é devolvido;
20.1.5. A camada [dao]
Voltemos à arquitetura da aplicação 1 a construir:
A camada [dao] [1] deve ler o ficheiro [admindata.json] [2] e transferir o seu conteúdo para uma das bases de dados [3, 4];
A camada [dao] apresenta a interface [1] e é implementada pela classe [2].
A interface [InterfaceDao4TransferAdminData2Database] é a seguinte:
# importações
from abc import ABC, abstractmethod
# interface InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# transferência de dados fiscais para uma base de dados
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
- linhas 8-10: a interface apresenta apenas um método [transfer_admindata_in_database] sem parâmetros. Como este método necessita de parâmetros (que ficheiro?, que base de dados?), isso significa que estes serão passados para o construtor das classes que implementam esta interface;
A classe [DaoTransferAdminDataFromJsonFile2Database] implementa a interface [InterfaceDao4TransferAdminData2Database] da seguinte forma:
# importações
import codecs
import json
from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError
from Constantes import Constantes
from ImpôtsError import ImpôtsError
from InterfaceDao4TransferAdminData2Database import InterfaceDao4TransferAdminData2Database
from Tranche import Tranche
class DaoTransferAdminDataFromJsonFile2Database(InterfaceDao4TransferAdminData2Database):
# fabricante
def __init__(self, config: dict):
self.config = config
# transferência
def transfer_admindata_in_database(self) -> None:
# inicializações
session = None
config = self.config
try:
# recuperação dos dados da administração fiscal
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# transferência do conteúdo para um dicionário
admindata = json.load(fd)
# recuperação da configuração da base de dados
database = config["database"]
# eliminação das duas tabelas da base de dados
# checkfirst=True: verifica primeiro se a tabela existe
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# recriação das tabelas a partir dos mapeamentos
database["metadata"].create_all(database["engine"])
# a sessão [sqlalchemy] atual
session = database["session"]
# preenche-se a tabela das faixas de imposto
limites = admindata["limites"]
coeffr = admindata["coeffr"]
coeffn = admindata["coeffn"]
for i in range(len(limites)):
session.add(Tranche().fromdict(
{"limite": limites[i], "coeffr": coeffr[i], "coeffn": coeffn[i]}))
# preenchimento da tabela de constantes
session.add(Constantes().fromdict({
'plafond_qf_demi_part': admindata["plafond_qf_demi_part"],
'plafond_revenus_celibataire_pour_reduction': admindata["plafond_revenus_celibataire_pour_reduction"],
'plafond_revenus_couple_pour_reduction': admindata["plafond_revenus_couple_pour_reduction"],
'valeur_reduc_demi_part': admindata["valeur_reduc_demi_part"],
'plafond_decote_celibataire': admindata["plafond_decote_celibataire"],
'plafond_decote_couple': admindata["plafond_decote_couple"],
'plafond_impot_celibataire_pour_decote': admindata["plafond_impot_celibataire_pour_decote"],
'plafond_impot_couple_pour_decote': admindata["plafond_impot_couple_pour_decote"],
'abattement_dixpourcent_max': admindata["abattement_dixpourcent_max"],
'abattement_dixpourcent_min': admindata["abattement_dixpourcent_min"]
}))
# validação da sessão [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# a exceção é relançada sob outra forma
raise ImpôtsError(17, f"{erreur}")
finally:
# liberam-se os recursos da sessão
if session:
session.close()
- linha 13: a classe [DaoTransferAdminDataFromJsonFile2Database] implementa a interface [InterfaceDao4TransferAdminData2Database];
- linhas 15-17: o construtor da classe recebe como parâmetro o dicionário de configuração. Serão utilizadas as seguintes chaves:
- [admindataFilename] (linha 27): o nome do ficheiro jSON que contém os dados da administração fiscal a transferir para a base de dados;
- [database], linha 32: a configuração [sqlalchemy] da aplicação;
- linhas 34-37: eliminação das tabelas [constantes] e [tranches], caso existam;
- linhas 39-40: recriação das duas tabelas;
- linha 43: recupera-se a sessão [sqlalchemy] presente na configuração;
- linhas 45-51: as tabelas [limites, coeffr, coeffn] do dicionário [admindata] são inseridas na sessão. Para tal, são inseridas na sessão instâncias da entidade [Tranche];
- linhas 52-64: uma instância da entidade [Constantes] é colocada na sessão;
- linhas 66-67: a sessão é validada. Se os dados da sessão ainda não se encontrassem na base de dados, são inseridos nesse momento;
- linhas 68-70: gestão de um eventual erro;
- linhas 71-74: a sessão é encerrada. Isto é possível porque a camada [dao] é utilizada apenas uma vez;
20.1.6. Configuração da aplicação
A aplicação é configurada por três ficheiros [1]:
- [config] é o ficheiro de configuração geral. É este ficheiro que configura a aplicação [main]. Para tal, conta com a ajuda dos outros dois ficheiros:
- [config_database], que já analisámos e que configura o ORM e o [sqlalchemy];
- [config_layers], que configura as camadas da aplicação;
O ficheiro [config] é o seguinte:
def configure(config: dict) -> dict:
# [config] com a chave [sgbd], cujo valor é:
# [mysql] para gerir uma base de dados MySQL
# [pgres] para gerir uma base de dados PostgreSQL
import os
# passo 1 ---
# define-se o Python Path da aplicação
# caminho absoluto da pasta deste script
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (a alterar, se necessário)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# caminhos absolutos das dependências
absolute_dependencies = [
# 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",
# AdminData, ImpôtsError, TaxPayer
f"{root_dir}/impots/v04/entities",
# BaseEntity, MyException
f"{root_dir}/classes/02/entities",
# pastas locais
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# definir o syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# etapa 2 ------
# conclui-se a configuração da aplicação
config.update({
# caminhos absolutos dos ficheiros de dados
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# passo 3 ------
# configuração da base de dados
import config_database
config = config_database.configure(config)
# etapa 4 ------
# instanciação das camadas da aplicação
import config_layers
config = config_layers.configure(config)
# aplicamos a configuração
return config
- linhas 8-36: constrói-se o Python Path da aplicação;
- linhas 38-43: insere-se na configuração o caminho do ficheiro [admindata.json];
- linhas 45-48: configuração do ficheiro [sqlalchemy];
- linhas 50-53: instanciamento das camadas da aplicação;
- linha 56: devolve-se a configuração geral;
O ficheiro [config_layers] é o seguinte:
def configure(config: dict) -> dict:
# instanciação da camada [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# retornamos a configuração
return config
- linhas 3-4: instanciação da camada [dao]. Vimos que o construtor da classe [DaoTransferAdminDataFromJsonFile2Database] esperava, como parâmetro, o dicionário da configuração geral da aplicação;
- linha 4: a referência à camada [dao] é inserida na configuração;
- linha 7: a configuração é devolvida;
20.1.7. O script [main] da aplicação
O script principal [main] é 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()
# configuramos a aplicação
import config
config = config.configure({'sgbd': sgbd})
# o syspath está definido — já é possível efetuar as importações
from ImpôtsError import ImpôtsError
# recuperamos a camada [dao]
dao = config["dao"]
# código
try:
# transferência de dados para a base de dados
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# é apresentado o erro
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# exibe o erro
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# fim
print("Terminé...")
- linhas 1-10: aguarda-se um parâmetro. Verifica-se se este existe e se está correto;
- linhas 12-14: configura-se a aplicação (geral, SQLAlchemy, camadas) passando como parâmetro o tipo de SGBD escolhido;
- linhas 19-20: vamos precisar da camada [dao]. Recuperamo-la;
- linha 25: efetuamos a transferência para a base de dados. Todas as informações necessárias para o método [transfer_admindata_in_database] estão disponíveis nas propriedades da camada [dao] da linha 20. É aí que o método irá buscá-las;
Após a execução com a base MySQL, esta contém os seguintes elementos (phpMyAdmin):
Na coluna [3], podem ver-se os valores atribuídos por MySQL à chave primária [id]. A numeração começa em 1. A captura de ecrã acima foi obtida após várias execuções do script.
Com a base de dados PostgreSQL, os resultados são os seguintes:
- clica-se com o botão direito do rato em [1] e, em seguida, em [2-3];
- em [4], os dados das faixas de imposto aparecem corretamente;
Repetimos o mesmo procedimento para a tabela de constantes [tbconstantes]:
20.2. Aplicação 2: cálculo do imposto em modo batch
20.2.1. Arquitetura
A aplicação de cálculo do imposto da versão 4 utilizava a seguinte arquitetura:
A camada [dao] implementa uma interface [InterfaceImpôtsDao]. Criámos uma classe que implementa essa interface:
- [ImpôtsDaoWithAdminDataInJsonFile], que iria buscar os dados fiscais a um ficheiro jSON. Esta era a versão 3;
Vamos implementar a interface [InterfaceImpôtsDao] através de uma nova classe, [ImpotsDaoWithTaxAdminDataInDatabase], que irá buscar os dados da administração fiscal numa base de dados. A camada [dao], tal como anteriormente, gravará os resultados num ficheiro jSON e irá buscar os dados dos contribuintes num ficheiro de texto. Sabemos que, se continuarmos a respeitar a interface [InterfaceImpôtsDao], a camada [métier] não terá de ser alterada.
A nova arquitetura será a seguinte:
20.2.2. Configuração da aplicação
O ficheiro de configuração [config_database] mantém-se tal como estava na aplicação 1. A configuração [config] inclui novos elementos:
# etapa 2 ------
# conclui-se a configuração da aplicação
config.update({
# caminhos absolutos dos ficheiros de dados
"admindataFilename": f"{script_dir}/../../data/input/admindata.json",
"taxpayersFilename": f"{script_dir}/../../data/input/taxpayersdata.txt",
"errorsFilename": f"{script_dir}/../../data/output/errors.txt",
"resultsFilename": f"{script_dir}/../../data/output/résultats.json"
})
- linhas 6-8: os caminhos absolutos dos ficheiros de texto utilizados pela aplicação 2;
A configuração das camadas [config_layers] evolui da seguinte forma:
def configure(config: dict) -> dict:
# instanciação da camada DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instanciação da camada [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# a configuração é guardada
return config
- linhas 3-4: a camada [dao] é agora implementada pela classe [ImpotsDaoWithAdminDataInDatabase]. Esta classe é nova, mas implementa a mesma interface [InterfaceDao] que a versão 4 do exercício de aplicação;
- linhas 7-8: a camada [métier] é implementada pela classe [ImpôtsMétier]. Esta é a classe utilizada na versão 4 do exercício prático;
20.2.3. A camada [dao]
A classe de implementação [ImpotsDaoWithAdminDataInDatabase] da interface [InterfaceImpôtsDao] será a seguinte:
# importações
from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError
from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from Constantes import Constantes
from ImpôtsError import ImpôtsError
from Tranche import Tranche
class ImpotsDaoWithAdminDataInDatabase(AbstractImpôtsDao):
# construtor
def __init__(self, config: dict):
# config["taxPayersFilename"]: o nome do ficheiro de texto dos contribuintes
# config["taxPayersResultsFilename"]: o nome do ficheiro jSON dos resultados
# config["errorsFilename"]: regista os erros encontrados em taxPayersFilename
# config["database"]: configuração da base de dados
# inicialização da classe Parent
AbstractImpôtsDao.__init__(self, config)
# armazenamento de parâmetros
self.__config = config
# dados de administração
self.__admindata = None
# implementação da interface
def get_admindata(self):
# O admindata foi memorizado?
if self.__admindata:
return self.__admindata
# é efetuada uma consulta em BD
session = None
config = self.__config
try:
# uma sessão
database_config = config["database"]
session = database_config["session"]
# está a ser lida a tabela de escalões de imposto
tranches = session.query(Tranche).all()
# é lida a tabela de constantes (apenas uma linha)
constantes = session.query(Constantes).first()
# cria-se a instância admindata
admindata = AdminData()
# criam-se as tabelas de limites, coeffR, coeffN
limites = admindata.limites = []
coeffr = admindata.coeffr = []
coeffn = admindata.coeffn = []
for tranche in tranches:
limites.append(float(tranche.limite))
coeffr.append(float(tranche.coeffr))
coeffn.append(float(tranche.coeffn))
# adicionam-se as constantes
admindata.fromdict(constantes.asdict())
# guarda-se a instância «admindata»
self.__admindata = admindata
# retorna-se o valor
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# lança-se novamente a exceção sob outra forma
raise ImpôtsError(27, f"{erreur}")
finally:
# encerra-se a sessão
if session:
session.close()
Notas
- linha 11: a classe [ImpotsDaoWithAdminDataInDatabase] herda da classe [AbstractImpôtsDao] apresentada na versão 4. Sabe-se que esta última implementa a interface [InterfaceDao] apresentada nessa mesma versão. É o cumprimento desta interface que nos permite não alterar a camada [métier];
- linha 13: o construtor da classe recebe como parâmetro o dicionário da configuração da aplicação;
- linha 20: a classe pai [] é inicializada. Ela implementa parcialmente a interface [InterfaceDao]:
- [get_taxpayers_data] lê o ficheiro [taxpayersdata.txt], que contém os dados dos contribuintes;
- [write_taxpayers_results] grava os resultados no ficheiro jSON [résultats.json];
- [get_admindata] não está implementada;
- linha 22: guarda-se a configuração passada como parâmetros;
- linha 27: implementação do método [get_admindata] da interface [InterfaceDao]:
- linhas 28-30: o método [get_admindata] recupera os dados da administração fiscal num objeto do tipo [AdminData] e armazena esse objeto em [self.__admindata]. Se o método [get_admindata] for chamado várias vezes, a base de dados não é consultada repetidamente. É consultada apenas na primeira vez. Nas vezes seguintes, é devolvido o objeto [self.__admindata];
- linhas 36-37: recupera-se a sessão [sqlalchemy], que foi criada durante a configuração da aplicação pelo método [config_database];
- linhas 40: recuperam-se as faixas de imposto numa lista;
- linhas 43: recuperam-se as constantes do cálculo do imposto;
- linha 46: cria-se uma instância da classe [AdminData]. Recorde-se que esta deriva de [BaseEntity];
- linhas 48-54: inicializam-se os tabuletos [limites, coeffr, coeffn] da instância [AdminData];
- linhas 55-56: inicializam-se as restantes propriedades de [AdminData] com as constantes do cálculo do imposto. Tivemos o cuidado de atribuir os mesmos nomes às propriedades das classes [AdminData] e [Constantes], o que simplifica o código;
- linhas 57-58: a instância [AdminData] é armazenada na camada [dao] para ser devolvida nas próximas chamadas ao método [get_admindata];
- linha 60: devolve-se o valor solicitado pelo código chamador;
- linhas 61-63: gestão de um eventual erro;
- linhas 64-67: a base de dados é alvo de apenas uma única consulta. Por isso, é possível encerrar a sessão [sqlalchemy];
20.2.4. Teste da camada [dao]
Na versão 4 desta aplicação, criámos uma classe de teste da camada [métier]. Mais precisamente, esta testava simultaneamente as camadas [métier] e [dao]. Retomamos este teste para verificar se a camada [dao] funciona conforme o esperado. Com efeito, a camada [métier] não sofre alterações.
O teste [TestDaoMétier] é o seguinte:
import unittest
class TestDaoMétier(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})
métier.calculate_tax(taxpayer, admindata)
# 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})
métier.calculate_tax(taxpayer, admindata)
# 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__':
# é esperado 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()
# a aplicação está a ser configurada
import config
config = config.configure({'sgbd': sgbd})
# camada de negócio
métier = config['métier']
try:
# admindata
admindata = config['dao'].get_admindata()
except BaseException as ex:
# visualização
print((f"L'erreur suivante s'est produite : {ex}"))
# fim
sys.exit()
# envia-se o parâmetro recebido pelo script
sys.argv.pop()
# executam-se os métodos de teste
print("tests en cours...")
unittest.main()
- Não vamos voltar aos 11 testes descritos no parágrafo |teste da camada [métier] versão 4|;
- linhas 37-66: vamos executar o script dos testes como uma aplicação normal e não como um teste UnitTest. É a linha 66 que irá acionar o framework UnitTest. Nos testes anteriores, utilizávamos o método [setUp] para configurar a execução de cada teste. Repetíamos 11 vezes a mesma configuração, uma vez que a função [setUp] é executada antes de cada teste. Aqui, fazemos a configuração uma única vez. Consiste em definir as variáveis globais [métier], na linha 53, e [admindata], na linha 56, que serão posteriormente utilizadas pelos métodos de [TestDaoMétier], na linha 12, por exemplo;
- linhas 39-47: o script de teste aguarda um parâmetro [mysql / pgres] que indica se se utiliza uma base MySQL ou PostgreSQL;
- linhas 50-51: o teste é configurado;
- linha 53: recupera-se a camada [métier] na configuração;
- linha 56: faz-se o mesmo com a camada [dao]. Recupera-se então a instância [admindata], que encapsula os dados necessários ao cálculo do imposto;
- os testes revelaram que o método [unittest.main()] da linha 66 não ignorava o parâmetro [mysql / pgres] recebido pelo script, mas atribuía-lhe um significado diferente. A linha 63 faz com que este método deixe de ter qualquer parâmetro;
Criamos duas configurações de execução:
Se executarmos uma destas duas configurações, obtemos os seguintes resultados:
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/v05/tests/TestDaoMétier.py mysql
tests en cours...
...........
----------------------------------------------------------------------
Ran 11 tests in 0.001s
OK
Process finished with exit code 0
- linhas 5 e 7: os 11 testes foram bem-sucedidos;
Recorde-se que estes testes verificam apenas 11 casos de cálculo do imposto. O seu sucesso pode, no entanto, ser suficiente para nos dar confiança na camada [dao].
20.2.5. O script principal
O script principal [main] é o mesmo da versão 4:
# 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()
# configura-se a aplicação
import config
config = config.configure({'sgbd': sgbd})
# o syspath está definido — já é possível efetuar as importações
from ImpôtsError import ImpôtsError
# recuperam-se as camadas da aplicação (já estão instanciadas)
dao = config["dao"]
métier = config["métier"]
try:
# recuperação dos escalões de imposto
admindata = dao.get_admindata()
# leitura dos dados dos contribuintes
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# dos contribuintes?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# cálculo do imposto dos contribuintes
for taxPayer in taxpayers:
# taxPayer é simultaneamente um parâmetro de entrada e de saída
# O taxPayer vai ser alterado
métier.calculate_tax(taxPayer, admindata)
# 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é...")
Notas
- linhas 1-10: recupera-se o parâmetro [mysql / pgres], que indica o SGBD a utilizar;
- linhas 12-14: a aplicação é configurada;
- linhas 16-17: a classe [ImpôtsError] é importada. É necessária na linha 38;
- linhas 19-21: recuperam-se referências às camadas da aplicação;
- linha 25: solicitam-se à camada [dao] os dados da administração fiscal. A camada [métier] necessita deles para o cálculo do imposto;
- linha 27: recuperam-se, numa lista, os dados (ID, estado civil, filhos, salário) dos contribuintes;
- linhas 29-30: se esta lista estiver vazia, é lançada uma exceção;
- linhas 32-35: cálculo do imposto dos elementos da lista [taxpayers];
- linha 37: gravação dos resultados no ficheiro jSON[résultats.json];
- linhas 38-40: gestão de eventuais erros;
Para a execução do script, criam-se duas |configurações de execução|:
Os resultados obtidos no ficheiro [résultats.json] são os da versão 4.
20.3. Aplicação 3: cálculo do imposto em modo interativo
Apresentamos agora a aplicação que permite calcular o imposto de forma interativa. Trata-se de uma adaptação da aplicação 2 da versão 4.
- o script [main] inicia o diálogo com o utilizador através do método [ui.run] da camada [ui];
- a camada [ui]:
- utiliza a camada [dao] para obter os dados necessários ao cálculo do imposto;
- solicita ao utilizador as informações relativas ao contribuinte cujo imposto se pretende calcular;
- utiliza a camada [métier] para efetuar esse cálculo;
O ficheiro [config_layers] instancia uma camada adicional:
def configure(config: dict) -> dict:
# instanciação da camada DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instanciação da camada [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# interface do utilizador
from ImpôtsConsole import ImpôtsConsole
config['ui'] = ImpôtsConsole(config)
# enviando a configuração
return config
A classe [ImpôtsConsole], linhas 11-12, é a mesma que na |versão 4|.
O script principal [main] é 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()
# configuramos a aplicação
import config
config = config.configure({'sgbd': sgbd})
# o syspath está configurado — já é possível efetuar as importações
from ImpôtsError import ImpôtsError
# recuperar a camada [ui]
ui = config["ui"]
# código
try:
# execução da camada [ui]
ui.run()
except ImpôtsError as ex1:
# é apresentada a mensagem de erro
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# é apresentada a mensagem de erro
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# executado em todos os casos
print("Travail terminé...")
- nas linhas 1-10, o script aguarda um parâmetro [mysql / pgres] que indica o SGBD a utilizar;
- linhas 12-14: a aplicação é configurada;
- linhas 19-20: recupera-se a camada [ui] da configuração;
- linha 25: executa-se a camada;
Os resultados são idênticos aos da |versão 4|. Não poderia ser de outra forma, uma vez que todas as interfaces da versão 4 foram respeitadas na versão 5.