20. Exercício prático: versão 5
Vamos desenvolver três aplicativos:
- a aplicação 1 inicializará o banco de dados que substituirá o arquivo [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 no modo interativo;
20.1. Aplicativo 1: inicialização do banco de dados
O aplicativo 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 armazenados em um banco de dados, em vez de estarem em um arquivo jSON. A camada [dao] será adaptada para implementar essa mudança.
20.1.1. O arquivo [admindata.json]
O arquivo [admindata.json] é o mesmo da 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
}
Usaremos como colunas do banco de dados as chaves desse dicionário.
20.1.2. Criação dos bancos de dados
Conforme demonstrado no parágrafo |criação de um banco de dados MySQL|, criamos um banco de dados MySQL denominado [dbimpots-2019], de propriedade do usuário [admimpots], com a senha [mdpimpots]. Em [phpMyAdmin], isso resulta no seguinte:
Da mesma forma, conforme demonstrado no parágrafo |criação de um banco de dados PostgreSQL|, criamos um banco de dados PostgreSQL chamado [dbimpots-2019], de propriedade do usuário [admimpots], com a senha [mdpimpots]. Em [pgAdmin], isso resulta no seguinte:
Os bancos de dados foram criados, mas, por enquanto, não possuem nenhuma tabela. Estas serão criadas pelo ORM e [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 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)
)
Definida por [sqlalchemy], a tabela [tbconstantes] 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 essas duas tabelas serão as seguintes:
A entidade [Constantes] encapsula as constantes do dicionário [admindata.json]:
from BaseEntity import BaseEntity
# classe contêiner dos dados da administração fiscal
class Constantes(BaseEntity):
# chaves excluídas da declaração 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: por meio do mapeamento [sqlalchemy], a classe [Constante] receberá a propriedade [_sa_instance_state]. Nós a excluímos do dicionário [asdict] da entidade;
- linhas 11-23: as propriedades da entidade. Retomamos 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 contêiner dos dados da administração fiscal
class Tranche(BaseEntity):
# chaves excluídas do estado da classe
excluded_keys = ["_sa_instance_state"]
# chaves permitidas
@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 de 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. Nelas, omitimos estabelecer as correspondências entre as propriedades das entidades mapeadas e as tabelas do banco de dados. Isso é possível quando os nomes das colunas das tabelas são os mesmos das propriedades às quais devem ser associadas. Por esse motivo, incluímos nas tabelas os nomes das propriedades das entidades mapeadas. Isso facilita a escrita do código e sua compreensão;
20.1.4. O arquivo de configuração do [sqlalchemy]
Acabamos de detalhar parte da configuração do [sqlalchemy]. O arquivo [config_database] na íntegra é 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 conexão aos bancos de dados utilizados
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# cadeia de conexão com o banco de dados em uso
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 de 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()
# registramos algumas 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 qual SGBD deve ser utilizado: MySQL (mysql) ou PostgreSQL (pgres);
- linhas 6-12: seleciona-se o banco de dados solicitado pela configuração;
- linhas 14-44: mapeamentos de entidades/tabelas. Esses mapeamentos são simples, pois não há nenhuma relação entre as tabelas [tranches] e [constantes]. Elas são independentes. Portanto, não há chave estrangeira de uma na outra para ser gerenciada;
- linhas 46-51: cria-se a sessão de trabalho [session] do aplicativo;
- linhas 53-58: as informações necessárias são inseridas no dicionário de configuração, e este é retornado;
20.1.5. A camada [dao]
Voltemos à arquitetura do aplicativo 1 a ser construído:
A camada [dao] [1] deve ler o arquivo [admindata.json] [2] e transferir seu conteúdo para um dos bancos 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 um banco 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 esse método precisa de parâmetros (qual arquivo?, qual banco de dados?), isso significa que esses parâmetros serão passados para o construtor das classes que implementam essa interface;
A classe [DaoTransferAdminDataFromJsonFile2Database] implementa a interface [InterfaceDao4TransferAdminData2Database] da seguinte maneira:
# 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 do banco de dados
database = config["database"]
# exclusão das duas tabelas do banco 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 atual [sqlalchemy]
session = database["session"]
# preenche-se a tabela de 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 de outra forma
raise ImpôtsError(17, f"{erreur}")
finally:
# os recursos da sessão são liberados
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 arquivo jSON que contém os dados da administração fiscal a serem transferidos para o banco de dados;
- [database], linha 32: a configuração [sqlalchemy] do aplicativo;
- linhas 34-37: exclusã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 isso, são inseridas na sessão instâncias da entidade [Tranche];
- linhas 52-64: uma instância da entidade [Constantes] é inserida na sessão;
- linhas 66-67: a sessão é validada. Se os dados da sessão ainda não estiverem no banco de dados, eles são inseridos nesse momento;
- linhas 68-70: tratamento de um possível erro;
- linhas 71-74: a sessão é encerrada. Isso é possível porque a camada [dao] é utilizada apenas uma vez;
20.1.6. Configuração do aplicativo
A aplicação é configurada por três arquivos [1]:
- [config] é o arquivo de configuração geral. É ele que configura a aplicação [main]. Ele conta com a ajuda dos outros dois arquivos:
- [config_database], que já analisamos e que configura o ORM e o [sqlalchemy];
- [config_layers], que configura as camadas do aplicativo;
O arquivo [config] é o seguinte:
def configure(config: dict) -> dict:
# [config] com a chave [sgbd], cujo valor é:
# [mysql] para gerenciar um banco de dados MySQL
# [pgres] para gerenciar um banco de dados PostgreSQL
import os
# etapa 1 ---
# define-se o Python Path do aplicativo
# caminho absoluto da pasta deste script
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (a ser alterado, 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",
]
# definimos o syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# etapa 2 ------
# concluindo a configuração do aplicativo
config.update({
# caminhos absolutos dos arquivos de dados
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# etapa 3 ------
# configuração do banco de dados
import config_database
config = config_database.configure(config)
# etapa 4 ------
# instanciação das camadas do aplicativo
import config_layers
config = config_layers.configure(config)
# aplicação da configuração
return config
- linhas 8-36: constrói-se o Python Path do aplicativo;
- linhas 38-43: insere-se na configuração o caminho do arquivo [admindata.json];
- linhas 45-48: configuração do [sqlalchemy];
- linhas 50-53: instanciamento das camadas do aplicativo;
- linha 56: retorna-se a configuração geral;
O arquivo [config_layers] é o seguinte:
def configure(config: dict) -> dict:
# instanciação da camada [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# retornando 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 do aplicativo;
- linha 4: a referência à camada [dao] é inserida na configuração;
- linha 7: a configuração é devolvida;
20.1.7. O script [main] do aplicativo
O script principal [main] é 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()
# configurando o aplicativo
import config
config = config.configure({'sgbd': sgbd})
# o syspath está definido — podemos fazer as importações
from ImpôtsError import ImpôtsError
# recuperando a camada [dao]
dao = config["dao"]
# código
try:
# transferência de dados para o banco de dados
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# é exibido 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 ele está presente 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]. A recuperamos;
- linha 25: fazemos a transferência para o banco 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. É daí que ele irá buscá-las;
Após a execução com a base MySQL, esta contém os seguintes elementos (phpMyAdmin):
Na coluna [3], vemos os valores atribuídos por MySQL à chave primária [id]. A numeração começa em 1. A captura de tela acima foi obtida após várias execuções do script.
Com a base PostgreSQL, os resultados são os seguintes:
- clica-se com o botão direito do mouse em [1] e, em seguida, em [2-3];
- em [4], os dados das faixas de imposto estão corretos;
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 de impostos da versão 4 utilizava a seguinte arquitetura:
A camada [dao] implementa uma interface [InterfaceImpôtsDao]. Criamos uma classe que implementa essa interface:
- [ImpôtsDaoWithAdminDataInJsonFile], que buscava os dados fiscais em um arquivo jSON. Essa era a versão 3;
Vamos implementar a interface [InterfaceImpôtsDao] por meio de uma nova classe, [ImpotsDaoWithTaxAdminDataInDatabase], que irá buscar os dados da administração fiscal em um banco de dados. A camada [dao], assim como antes, gravará os resultados em um arquivo jSON e buscará os dados dos contribuintes em um arquivo de texto. Sabemos que, se continuarmos a respeitar a interface [InterfaceImpôtsDao], a camada [métier] não precisará ser alterada.
A nova arquitetura será a seguinte:
20.2.2. Configuração do aplicativo
O arquivo de configuração [config_database] permanece igual ao da aplicação 1. A configuração [config] inclui novos elementos:
# etapa 2 ------
# conclui-se a configuração do aplicativo
config.update({
# caminhos absolutos dos arquivos 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 arquivos de texto utilizados pelo aplicativo 2;
A configuração das camadas [config_layers] sofre as seguintes alterações:
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()
# estamos retornando a configuração
return config
- linhas 3-4: a camada [dao] agora é implementada pela classe [ImpotsDaoWithAdminDataInDatabase]. Essa classe é nova, mas implementa a mesma interface [InterfaceDao] da versão 4 do exercício de aplicação;
- linhas 7-8: a camada [métier] é implementada pela classe [ImpôtsMétier]. Essa é 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 arquivo de texto dos contribuintes
# config["taxPayersResultsFilename"]: o nome do arquivo jSON dos resultados
# config["errorsFilename"]: registra os erros encontrados em taxPayersFilename
# config["database"]: configuração do banco 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 armazenado?
if self.__admindata:
return self.__admindata
# é feita uma consulta em BD
session = None
config = self.__config
try:
# uma sessão
database_config = config["database"]
session = database_config["session"]
# está sendo lida a tabela de faixas de imposto
tranches = session.query(Tranche).all()
# está-se lendo a tabela de constantes (apenas uma linha)
constantes = session.query(Constantes).first()
# cria-se a instância admindata
admindata = AdminData()
# criam-se nela 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))
# adiciona-se as constantes a ela
admindata.fromdict(constantes.asdict())
# armazenamos admindata
self.__admindata = admindata
# retorna-se o valor
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# relança-se a exceção de 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 dessa interface que nos permite não alterar a camada [métier];
- linha 13: o construtor da classe recebe como parâmetro o dicionário de configuração do aplicativo;
- linha 20: a classe pai [] é inicializada. Ela implementa parcialmente a interface [InterfaceDao]:
- [get_taxpayers_data] lê o arquivo [taxpayersdata.txt], que contém os dados dos contribuintes;
- [write_taxpayers_results] grava os resultados no arquivo jSON [résultats.json];
- [get_admindata] não está implementado;
- linha 22: armazena-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 em um objeto do tipo [AdminData] e armazena esse objeto em [self.__admindata]. Se o método [get_admindata] for chamado várias vezes, o banco de dados não é consultado repetidamente. Ele é consultado apenas na primeira vez. Nas vezes seguintes, retorna-se o objeto [self.__admindata];
- linhas 36-37: recupera-se a sessão [sqlalchemy], que foi criada durante a configuração do aplicativo pelo método [config_database];
- linhas 40: recuperam-se as faixas de imposto em uma lista;
- linhas 43: recuperam-se as constantes do cálculo do imposto;
- linha 46: cria-se uma instância da classe [AdminData]. Vale lembrar que ela deriva de [BaseEntity];
- linhas 48-54: inicializam-se os tabuletos [limites, coeffr, coeffn] da instância [AdminData];
- linhas 55-56: inicializam-se as demais 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: retorna-se o valor solicitado pelo código chamador;
- linhas 61-63: tratamento de um possível erro;
- linhas 64-67: o banco de dados é alvo de apenas uma única consulta. Portanto, é possível encerrar a sessão [sqlalchemy];
20.2.4. Teste da camada [dao]
Na versão 4 desta aplicação, criamos uma classe de teste para a camada [métier]. Mais precisamente, ela testava tanto a camada [métier] quanto a camada [dao]. Retomamos esse teste para verificar se a camada [dao] funciona conforme o esperado. De fato, a camada [métier] não sofreu 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, 'sobretaxa': 0, 'abatimento': 0, 'redução': 0, 'alíquota': 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, 'abatimento': 0, 'redução': 0, 'alíquota': 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__':
# 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()
# configurando o aplicativo
import config
config = config.configure({'sgbd': sgbd})
# camada de negócios
métier = config['métier']
try:
# dados de administração
admindata = config['dao'].get_admindata()
except BaseException as ex:
# exibição
print((f"L'erreur suivante s'est produite : {ex}"))
# fim
sys.exit()
# envia-se o parâmetro recebido pelo script
sys.argv.pop()
# executa-se os métodos de teste
print("tests en cours...")
unittest.main()
- Não vamos nos deter nos 11 testes descritos no parágrafo |teste da camada [métier] versão 4|;
- linhas 37-66: vamos executar o script de testes como um aplicativo normal e não como um teste UnitTest. É a linha 66 que acionará o framework UnitTest. Nos testes anteriores, utilizávamos o método [setUp] para configurar a execução de cada teste. Repetíamos a mesma configuração 11 vezes, já que a função [setUp] é executada antes de cada teste. Aqui, fazemos a configuração uma única vez. Ela 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 será utilizada 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 para o cálculo do imposto;
- os testes mostraram que o método [unittest.main()] da linha 66 não ignorava o parâmetro [mysql / pgres] recebido pelo script, mas atribuía a ele um significado diferente. A linha 63 faz com que esse método não tenha mais nenhum parâmetro;
Criamos duas configurações de execução:
Se executarmos uma dessas 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;
Vale lembrar que esses testes verificam apenas 11 casos de cálculo do imposto. No entanto, o fato de terem sido bem-sucedidos pode 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 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 o aplicativo
import config
config = config.configure({'sgbd': sgbd})
# o syspath está definido — é possível realizar as importações
from ImpôtsError import ImpôtsError
# recuperam-se as camadas da aplicação (elas já estão instanciadas)
dao = config["dao"]
métier = config["métier"]
try:
# recuperação das faixas 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 é tanto um parâmetro de entrada quanto de saída
# taxPayer será modificado
métier.calculate_tax(taxPayer, admindata)
# gravação dos resultados em um arquivo 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 ser utilizado;
- linhas 12-14: a aplicação é configurada;
- linhas 16-17: a classe [ImpôtsError] é importada. Ela é 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] precisa deles para o cálculo do imposto;
- linha 27: recuperamos, em uma lista, os dados (ID, estado civil, filhos, salário) dos contribuintes;
- linhas 29-30: se essa 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 arquivo jSON[résultats.json];
- linhas 38-40: tratamento de eventuais erros;
Para a execução do script, criam-se duas |configurações de execução|:
Os resultados obtidos no arquivo [résultats.json] são os da versão 4.
20.3. Aplicativo 3: cálculo do imposto no 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 a interação com o usuário por meio 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 usuário as informações relativas ao contribuinte cujo imposto se deseja calcular;
- utiliza a camada [métier] para realizar esse cálculo;
O arquivo [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 usuário
from ImpôtsConsole import ImpôtsConsole
config['ui'] = ImpôtsConsole(config)
# retornando a configuração
return config
A classe [ImpôtsConsole], linhas 11-12, é a mesma da |versão 4|.
O script principal [main] é o seguinte:
# aguardando 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 o aplicativo
import config
config = config.configure({'sgbd': sgbd})
# o syspath está configurado — já é possível fazer as importações
from ImpôtsError import ImpôtsError
# recuperando a camada [ui]
ui = config["ui"]
# código
try:
# execução da camada [ui]
ui.run()
except ImpôtsError as ex1:
# exibe-se a mensagem de erro
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# é exibida a mensagem de erro
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# executado em todos os casos
print("Travail terminé...")
- linhas 1-10: o script aguarda um parâmetro [mysql / pgres] que indica o SGBD a ser utilizado;
- linhas 12-14: o aplicativo é configurado;
- linhas 19-20: recupera-se a camada [ui] da configuração;
- linha 25: ela é executada;
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 mantidas na versão 5.