Skip to content

20. Exercício de aplicação: versão 5

Image

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 calculará os impostos em modo de lote;
  • A aplicação 3 calculará os impostos em modo interativo;

20.1. Aplicação 1: Inicialização da base de dados

A Aplicação 1 terá a seguinte arquitetura:

Image

Esta é uma evolução da arquitetura da Versão 4 (consulte a secção |Versão 4|): os dados fiscais serão armazenados numa base de dados em vez de num ficheiro JSON. A camada [DAO] será atualizada para implementar esta alteração.

20.1.1. O ficheiro [admindata.json]

Image

O ficheiro [admindata.json] é o mesmo que 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 as chaves deste dicionário como colunas na base de dados.

20.1.2. Criação das bases de dados

Conforme mostrado na secção |criar uma base de dados MySQL|, criamos uma base de dados MySQL chamada [dbimpots-2019], pertencente ao utilizador [admimpots] com a palavra-passe [mdpimpots]. No [phpMyAdmin], isto tem o seguinte aspeto:

Image

Da mesma forma, tal como mostrado na secção |Criação de uma base de dados PostgreSQL|, criamos uma base de dados PostgreSQL denominada [dbimpots-2019], pertencente ao utilizador [admimpots] e com a palavra-passe [mdpimpots]. No [pgAdmin], a situação apresenta-se da seguinte forma:

Image

As bases de dados foram criadas, mas, por enquanto, não têm tabelas. Estas serão criadas pelo ORM [sqlalchemy].

20.1.3. Entidades mapeadas pelo [sqlalchemy]

Iremos criar duas tabelas para encapsular os dados de [admindata.json]:

Definida pelo [sqlalchemy], a tabela [tbtranches] irá recolher dados das matrizes [limites, coeffr, coeffn] no dicionário [admindata.json]:


    # la table des tranches de l'impôt
    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 pelo [sqlalchemy], a tabela [tbconstantes] conterá as constantes do dicionário [admindata.json]:


    # la table des 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 para estas duas tabelas são as seguintes:

Image

A entidade [Constants] encapsula as constantes do dicionário [admindata.json]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from BaseEntity import BaseEntity


#  tax administration data container class
class Constantes(BaseEntity):
    #  keys excluded from class state
    excluded_keys = ["_sa_instance_state"]

    #  authorized keys
    @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 [Constants] estende a classe [BaseEntity];
  • linha 7: através do mapeamento [sqlalchemy], a classe [Constante] receberá a propriedade [_sa_instance_state]. Excluímo-la do dicionário [asdict] da entidade;
  • linhas 11–23: as propriedades da entidade. Reutilizá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 matrizes [limites, coeffr, coeffn] no dicionário [admindata.json]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from BaseEntity import BaseEntity


#  tax administration data container class
class Tranche(BaseEntity):
    #  keys excluded from class state
    excluded_keys = ["_sa_instance_state"]

    #  authorized keys
    @staticmethod
    def get_allowed_keys() -> list:
        return ["id", "limite", "coeffr", "coeffn"]
  • linha 5: a classe [Tranche] estende a classe [BaseEntity];
  • linha 7: a propriedade [_sa_instance_state] adicionada pelo [sqlalchemy] é excluída do dicionário [asdict] da entidade;
  • linhas 10–12: as propriedades da classe;

O mapeamento entre as entidades [Constants, Slice] e as tabelas [constants, slices] será o seguinte:

Image

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
…
    #  the constants table
    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)
                             )

    #  tax bracket table
    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)
                           )
    #  mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)
  • Os mapeamentos estão definidos nas linhas 24–29. Omitimos os mapeamentos entre as propriedades das entidades mapeadas e as tabelas da base de dados. Isto é possível quando os nomes das colunas da tabela são os mesmos que os das propriedades às quais devem estar associadas. Por este motivo, incluímos os nomes das propriedades das entidades mapeadas nas tabelas. Isto torna o código mais fácil de escrever e compreender;

20.1.4. O ficheiro de configuração [sqlalchemy]

Image

Acabámos de detalhar parte da configuração do [sqlalchemy]. O ficheiro [config_database] completo é o seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def configure(config: dict) -> dict:
    #  sqlalchemy configuration
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
    from sqlalchemy.orm import mapper, sessionmaker

    #  connection chains to the databases used
    connection_strings = {
        'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
        'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
    }
    #  connection chain to the database used
    engine = create_engine(connection_strings[config['sgbd']])

    #  metadata
    metadata = MetaData()

    #  the constants table
    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)
                             )

    #  tax bracket table
    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)
                           )
    #  mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

    #  the factory session
    session_factory = sessionmaker()
    session_factory.configure(bind=engine)

    #  a session
    session = session_factory()

    #  certain information is recorded
    config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
                          "constantes_table": constantes_table, "session": session}

    #  result
    return config
  • linha 1: a função [configure] recebe um dicionário como parâmetro, cuja chave [dbms] indica qual o SGBD a utilizar: MySQL (mysql) ou PostgreSQL (pgres);
  • linhas 6–12: o banco de dados especificado pela configuração é selecionado;
  • linhas 14–44: mapeamentos de entidade/tabela. Estes mapeamentos são simples porque não existe relação entre as tabelas [tranches] e [constantes]. São independentes. Por conseguinte, não há chaves estrangeiras entre elas para gerir;
  • linhas 46–51: criam a sessão de trabalho da aplicação [session];
  • linhas 53–58: as informações relevantes são colocadas no dicionário de configuração, que é então devolvido;

20.1.5. A camada [dao]

Voltemos à arquitetura da Aplicação 1 a ser construída:

Image

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];

Image

A camada [dao] fornece a interface [1] e é implementada pela classe [2].

A interface [InterfaceDao4TransferAdminData2Database] é a seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
#  imports
from abc import ABC, abstractmethod


#  interface InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
    #  transfer tax data to a database
    @abstractmethod
    def transfer_admindata_in_database(self:object):
        pass
  • linhas 8–10: a interface define apenas um método [transfer_admindata_in_database] sem parâmetros. Uma vez que este método requer parâmetros (qual ficheiro?, qual base de dados?), isto significa que esses parâmetros serão passados para o construtor das classes que implementam esta interface;

A classe [DaoTransferAdminDataFromJsonFile2Database] implementa a interface [InterfaceDao4TransferAdminData2Database] da seguinte forma:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
#  imports
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):

    #  manufacturer
    def __init__(self, config: dict):
        self.config = config

    #  transfer
    def transfer_admindata_in_database(self) -> None:
        #  initializations
        session = None
        config = self.config

        try:
            #  we retrieve data from the tax authorities
            with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
                #  transfer content to a dictionary
                admindata = json.load(fd)

            #  retrieve the database configuration
            database = config["database"]

            #  delete the two tables from the database
            #  checkfirst=True: first checks that the table exists
            database["tranches_table"].drop(database["engine"], checkfirst=True)
            database["constantes_table"].drop(database["engine"], checkfirst=True)

            #  recreate tables from mappings
            database["metadata"].create_all(database["engine"])

            #  the current [sqlalchemy] session
            session = database["session"]

            #  fill in the tax bracket table
            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]}))
            #  fill in the constants table
            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"]
            }))

            #  session validation [sqlalchemy]
            session.commit()
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            #  we relaunch the exception in another form
            raise ImpôtsError(17, f"{erreur}")
        finally:
            #  session resources are released
            if session:
                session.close()
  • linha 13: a classe [DaoTransferAdminDataFromJsonFile2Database] implementa a interface [InterfaceDao4TransferAdminData2Database];
  • linhas 15–17: o construtor da classe recebe o dicionário de configuração como parâmetro. Serão utilizadas as seguintes chaves:
    • [admindataFilename] (linha 27): o nome do ficheiro JSON que contém os dados da administração fiscal a serem transferidos para a base de dados;
    • [database] linha 32: a configuração [sqlalchemy] da aplicação;
  • linhas 34–37: eliminação das tabelas [constants] e [brackets], caso existam;
  • linhas 39–40: recriação das duas tabelas;
  • linha 43: recuperação da sessão [sqlalchemy] a partir da configuração;
  • linhas 45–51: as matrizes [limits, coeffr, coeffn] do dicionário [admindata] são adicionadas à sessão. Para tal, são adicionadas instâncias da entidade [Tranche] à sessão;
  • linhas 52–64: uma instância da entidade [Constantes] é adicionada à sessão;
  • linhas 66–67: a sessão é validada. Se os dados da sessão ainda não se encontravam na base de dados, são inseridos neste momento;
  • linhas 68–70: tratamento de erros;
  • 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

Image

A aplicação é configurada por três ficheiros [1]:

  • [config] é o ficheiro de configuração geral. Configura a aplicação [main]. É auxiliado pelos outros dois ficheiros:
    • [config_database], que já analisámos e que configura o ORM [sqlalchemy];
    • [config_layers], que configura as camadas da aplicação;

O ficheiro [config] é o seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def configure(config: dict) -> dict:
    #  [config] has the key [sgbd]:
    #  [mysql] to manage a MySQL database
    #  [pgres] to manage a PostgreSQL database

    import os

    #  step 1 ---
    #  establish the application's Python Path

    #  absolute path of this script's folder
    script_dir = os.path.dirname(os.path.abspath(__file__))

    #  root_dir (change if necessary)
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    #  absolute paths of dependencies
    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",
        #  local files
        f"{script_dir}",
        f"{script_dir}/../../interfaces",
        f"{script_dir}/../../services",
        f"{script_dir}/../../entities",
    ]

    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  complete application configuration
    config.update({
        #  absolute paths for data files
        "admindataFilename": f"{script_dir}/../../data/input/admindata.json"
    })

    #  step 3 ------
    #  database configuration
    import config_database
    config = config_database.configure(config)

    #  step 4 ------
    #  instantiation of application layers
    import config_layers
    config = config_layers.configure(config)

    #  return the config
    return config
  • linhas 8–36: Construir o Python Path da aplicação;
  • linhas 38–43: adicionar o caminho para o ficheiro [admindata.json] à configuração;
  • linhas 45–48: configuração do [SQLAlchemy];
  • linhas 50–53: instanciar as camadas da aplicação;
  • linha 56: devolve a configuração geral;

O ficheiro [config_layers] é o seguinte:

1
2
3
4
5
6
7
def configure(config: dict) -> dict:
    #  layer instantiation [dao]
    from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
    config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)

    #  return the config
    return config
  • linhas 3-4: instanciação da camada [dao]. Vimos que o construtor da classe [DaoTransferAdminDataFromJsonFile2Database] espera o dicionário de configuração geral da aplicação como parâmetro;
  • linha 4: a referência à camada [dao] é adicionada à configuração;
  • linha 7: retorna a configuração;

20.1.7. O script [main] da aplicação

Image

Image

O script principal [main] é o seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  the syspath is set up - imports can be made
from ImpôtsError import ImpôtsError

#  we recover the [dao] layer
dao = config["dao"]

#  code
try:
    #  data transfer to the database
    dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
    #  error is displayed
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    #  error is displayed
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    #  end
    print("Terminé...")
  • linhas 1–10: Aguardamos um parâmetro. Verificamos se está presente e se está correto;
  • linhas 12–14: Configuramos a aplicação (geral, SQLAlchemy, camadas) passando o tipo de SGBD escolhido como parâmetro;
  • linhas 19-20: vamos precisar da camada [dao]. Recuperamo-la;
  • linha 25: efetuamos a transferência para a base de dados. Toda a informação necessária ao método [transfer_admindata_in_database] está disponível nas propriedades da camada [dao] da linha 20. É daí que a irá recuperar;

Após executar o script com a base de dados MySQL, este contém os seguintes elementos (phpMyAdmin):

Image

Image

Image

A coluna [3] mostra os valores atribuídos pelo MySQL à chave primária [id]. A numeração começa em 1. A captura de ecrã acima foi tirada após a execução do script várias vezes.

Image

Image

Com a base de dados PostgreSQL, os resultados são os seguintes:

Image

  • Clique com o botão direito do rato em [1] e, em seguida, em [2-3];
  • Em [4], os dados relativos às faixas de imposto são apresentados de forma clara;

Fazemos o mesmo para a tabela de constantes [tbconstantes]:

Image

Image

Image

20.2. Aplicação 2: Cálculo de impostos em modo de lote

Image

20.2.1. Arquitetura

A aplicação de cálculo de impostos na versão 4 utilizava a seguinte arquitetura:

Image

A camada [dao] implementa uma interface [InterfaceImpôtsDao]. Criámos uma classe que implementa esta interface:

  • [TaxDaoWithAdminDataInJsonFile], que recuperava dados fiscais de um ficheiro JSON. Essa era a versão 3;

Iremos implementar a interface [InterfaceImpôtsDao] utilizando uma nova classe [ImpotsDaoWithTaxAdminDataInDatabase] que irá recuperar dados de administração fiscal de uma base de dados. A camada [dao], tal como anteriormente, irá gravar os resultados num ficheiro JSON e recuperar dados dos contribuintes a partir de um ficheiro de texto. Sabemos que, se continuarmos a aderir à interface [InterfaceImpôtsDao], a camada [business] não precisará de ser modificada.

A nova arquitetura será a seguinte:

Image

20.2.2. Configuração da Aplicação

Image

O ficheiro de configuração [config_database] permanece igual ao da Aplicação 1. A configuração [config] inclui novos elementos:


    # étape 2 ------
    # on complète la configuration de l'application
    config.update({
        # chemins absolus des fichiers de données
        "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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def configure(config: dict) -> dict:
    #  dao layer instantiation
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    #  instantiation layer [business]
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    #  return the config
    return config
  • linhas 3-4: a camada [dao] é agora implementada pela classe [TaxDaoWithAdminDataInDatabase]. Esta classe é nova, mas implementa a mesma interface [DaoInterface] que a versão 4 do exercício da aplicação;
  • linhas 7-8: a camada [business] é implementada pela classe [ImpôtsMétier]. Esta é a classe utilizada na versão 4 do exercício da aplicação;

20.2.3. A camada [DAO]

A classe de implementação [ImpotsDaoWithAdminDataInDatabase] para a interface [InterfaceImpôtsDao] será a seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
#  imports
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):
    #  manufacturer
    def __init__(self, config: dict):
        #  config["taxPayersFilename"]: name of the taxpayer text file
        #  config["taxPayersResultsFilename"]: name of the jSON results file
        #  config["errorsFilename"]: saves errors found in taxPayersFilename
        #  config["database"]: database configuration

        #  parent class initialization
        AbstractImpôtsDao.__init__(self, config)
        #  parameter memory
        self.__config = config
        #  admindata
        self.__admindata = None

    #  interface implementation
    def get_admindata(self):
        #  admindata memorized?
        if self.__admindata:
            return self.__admindata
        #  make a query in BD
        session = None
        config = self.__config
        try:
            #  a session
            database_config = config["database"]
            session = database_config["session"]

            #  read the table of tax brackets
            tranches = session.query(Tranche).all()

            #  read the constants table (1 line only)
            constantes = session.query(Constantes).first()

            #  create the admindata instance
            admindata = AdminData()
            #  we create limtes arrays, 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))
            #  we add the constants
            admindata.fromdict(constantes.asdict())
            #  admindata is memorized
            self.__admindata = admindata
            #  we return the value
            return self.__admindata
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            #  we relaunch the exception in another form
            raise ImpôtsError(27, f"{erreur}")
        finally:
            #  close session
            if session:
                session.close()

Notas

  • linha 11: a classe [ImpotsDaoWithAdminDataInDatabase] herda da classe [AbstractImpôtsDao] apresentada na versão 4. Sabemos que esta última implementa a interface [InterfaceDao] apresentada nessa mesma versão. É a conformidade com esta interface que nos permite manter a camada [business] inalterada;
  • linha 13: o construtor da classe recebe o dicionário de configuração da aplicação como parâmetro;
  • 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 [results.json];
    • [get_admindata] não está implementado;
  • linha 22: a configuração passada como parâmetros é armazenada;
  • linha 27: implementação do método [get_admindata] da interface [InterfaceDao]:
  • linhas 28–30: o método [get_admindata] recupera dados da administração fiscal para um objeto do tipo [AdminData] e armazena este objeto em [self.__admindata]. Se o método [get_admindata] for chamado várias vezes, a base de dados não é consultada várias vezes. É consultada apenas na primeira vez. Nas chamadas subsequentes, o objeto [self.__admindata] é devolvido;
  • linhas 36–37: recuperam a sessão [sqlalchemy] que foi criada durante a configuração da aplicação por [config_database];
  • linha 40: recuperamos as faixas de imposto numa lista;
  • linhas 43: recuperamos as constantes para o cálculo de impostos;
  • linha 46: criamos uma instância da classe [AdminData]. Recorde-se que esta deriva de [BaseEntity];
  • linhas 48–54: inicializamos as matrizes [limites, coeffr, coeffn] da instância [AdminData];
  • linhas 55–56: inicializamos as outras propriedades de [AdminData] com as constantes de cálculo de impostos. 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 durante chamadas subsequentes ao método [get_admindata];
  • linha 60: o valor solicitado pelo código de chamada é devolvido;
  • linhas 61–63: tratamento de erros;
  • linhas 64–67: a base de dados é consultada apenas uma vez. Podemos, portanto, encerrar a sessão [sqlalchemy];

20.2.4. Testar a camada [dao]

Na versão 4 desta aplicação, criámos uma classe de teste para a camada [business]. Mais especificamente, ela testava tanto a camada [business] como a camada [DAO]. Estamos a reutilizar este teste para verificar se a camada [DAO] está a funcionar conforme o esperado. A camada [business], no entanto, permanece inalterada.

Image

Image

O teste [TestDaoMétier] é o seguinte:


import unittest
 
 
class TestDaoMétier(unittest.TestCase):
 
    def test_1(self) -> None:
        from TaxPayer import TaxPayer
 
        # {'marié': 'oui', 'enfants': 2, 'salaire': 55555,
        # 'impôt': 2814, 'surcôte': 0, 'décôte': 0, 'réduction': 0, 'taux': 0.14}
        taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
        métier.calculate_tax(taxpayer, admindata)
        # vérification
        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
 
        # {'marié': 'oui', 'enfants': 3, 'salaire': 200000,
        # 'impôt': 42842, 'surcôte': 17283, 'décôte': 0, 'réduction': 0, 'taux': 0.41}
        taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
        métier.calculate_tax(taxpayer, admindata)
        # vérifications
        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__':
    # on attend un paramètre 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()
 
    # on configure l'application
    import config
    config = config.configure({'sgbd': sgbd})
    # couche métier
    métier = config['métier']
    try:
        # admindata
        admindata = config['dao'].get_admindata()
    except BaseException as ex:
        # affichage
        print((f"L'erreur suivante s'est produite : {ex}"))
        # fin
        sys.exit()
    # on enève le paramètre reçu par le script
    sys.argv.pop()
    # on exécute les méthodes de test
    print("tests en cours...")
   unittest.main()
  • Não revisitaremos os 11 testes descritos na secção |[business] layer test version 4|;
  • linhas 37–66: vamos executar o script de teste como uma aplicação normal, em vez de como um UnitTest. A linha 66 é o que irá acionar a estrutura UnitTest. Nos testes anteriores, utilizámos o método [setUp] para configurar a execução de cada teste. Estávamos a repetir a mesma configuração 11 vezes, uma vez que a função [setUp] é executada antes de cada teste. Aqui, realizamos a configuração uma única vez. Consiste em definir as variáveis globais [business] na linha 53 e [admindata] na linha 56, que serão depois utilizadas pelos métodos de [TestDaoBusiness], por exemplo na linha 12;
  • linhas 39–47: o script de teste espera um parâmetro [mysql / pgres] indicando se está a ser utilizada uma base de dados MySQL ou PostgreSQL;
  • linhas 50–51: o teste é configurado;
  • linha 53: a camada [business] é recuperada da configuração;
  • linha 56: fazemos o mesmo com a camada [dao]. Em seguida, recuperamos a instância [admindata], que encapsula os dados necessários para calcular o imposto;
  • Os testes revelaram que o método [unittest.main()] na linha 66 não ignorou o parâmetro [mysql / pgres] passado ao script, mas sim atribuiu-lhe um significado diferente. A linha 63 garante que este método já não tenha quaisquer parâmetros;

Criamos duas configurações de execução:

Image

Image

Se executarmos qualquer 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: todos os 11 testes foram aprovados;

Note-se que estes testes verificam apenas 11 casos de cálculo de impostos. O seu sucesso pode, no entanto, ser suficiente para nos dar confiança na camada [dao].

20.2.5. O script principal

Image

Image

O script principal [main] é o mesmo da versão 4:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  the syspath is set up - imports can be made
from ImpôtsError import ImpôtsError

#  retrieve application layers (already instantiated)
dao = config["dao"]
métier = config["métier"]

try:
    #  tax bracket recovery
    admindata = dao.get_admindata()
    #  reading taxpayer data
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    #  taxpayers?
    if not taxpayers:
        raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    #  tax calculation
    for taxPayer in taxpayers:
        #  taxPayer is both an input and output parameteri
        #  taxPayer will be modified
        métier.calculate_tax(taxPayer, admindata)
    #  writing results to a text file
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
    #  error display
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    #  completed
    print("Travail terminé...")

Notas

  • linhas 1-10: recuperamos o parâmetro [mysql / pgres], que especifica o SGBD a utilizar;
  • linhas 12–14: a aplicação é configurada;
  • linhas 16-17: a classe [ImpôtsError] é importada. Precisamos dela na linha 38;
  • linhas 19-21: recuperamos referências às camadas da aplicação;
  • linha 25: solicitamos os dados da administração fiscal à camada [dao]. A camada [business] precisa destes dados para calcular o imposto;
  • linha 27: recuperamos os dados dos contribuintes (id, casado, filhos, salário) para uma lista;
  • linhas 29–30: se esta lista estiver vazia, é lançada uma exceção;
  • linhas 32–35: calculamos o imposto para os itens da lista [taxpayers];
  • linha 37: gravamos os resultados no ficheiro JSON [results.json];
  • linhas 38–40: tratamos quaisquer erros;

Para executar o script, criamos duas |configurações de execução|:

Image

Os resultados obtidos no ficheiro [results.json] são os da versão 4.

Image

20.3. Aplicação 3: Cálculo de impostos em modo interativo

Apresentamos agora a aplicação que permite o cálculo interativo de impostos. Trata-se de uma adaptação da Aplicação 2 da Versão 4.

Image

Image

  • O script [main] inicia o diálogo com o utilizador utilizando o método [ui.run] da camada [ui];
  • A camada [ui]:
    • utiliza a camada [dao] para recuperar os dados necessários para calcular o imposto;
    • solicita ao utilizador informações relativas ao contribuinte para quem o imposto deve ser calculado;
    • utiliza a camada [business] para realizar este cálculo;

O ficheiro [config_layers] instancia uma camada adicional:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def configure(config: dict) -> dict:
    #  dao layer instantiation
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    #  instantiation layer [business]
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    #  ui
    from ImpôtsConsole import ImpôtsConsole
    config['ui'] = ImpôtsConsole(config)

    #  return the config
    return config

A classe [ImpôtsConsole], linhas 11–12, é a mesma da |versão 4|.

O script principal [main] é o seguinte:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  syspath is configured - imports can be made
from ImpôtsError import ImpôtsError

#  we recover the [ui] layer
ui = config["ui"]

#  code
try:
    #  execution of the [ui] layer
    ui.run()
except ImpôtsError as ex1:
    #  the error message is displayed
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    #  the error message is displayed
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    #  executed in all cases
    print("Travail terminé...")
  • linhas 1-10: o script espera um parâmetro [mysql / pgres] que especifique o SGBD a utilizar;
  • linhas 12-14: a aplicação é configurada;
  • linhas 19-20: a camada [ui] é recuperada da configuração;
  • linha 25: é 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 preservadas na versão 5.

https://stahe.github.io
Este tutorial, escrito por Serge Tahé, é disponibilizado ao público sob os termos da Creative Commons Atribuição – Não comercial – Partilha igual 3.0 Unported.