20. Ejercicio práctico: version 5
Vamos a desarrollar tres aplicaciones:
- la aplicación 1 inicializará la base de datos que sustituirá al archivo [admindata.json] del version 4;
- la aplicación 2 realizará el cálculo de los impuestos en modo batch;
- la aplicación 3 realizará el cálculo de los impuestos en modo interactivo;
20.1. Aplicación 1: inicialización de la base de datos
La aplicación 1 tendrá la siguiente arquitectura:
Se trata de una evolución de la arquitectura de version 4 (apartado |Version 4|): los datos fiscales se encontrarán en una base de datos en lugar de estar en un archivo jSON. La capa [dao] se actualizará para implementar este cambio.
20.1.1. El archivo [admindata.json]
El archivo [admindata.json] es el mismo que en version 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 a utilizar como columnas de la base de datos las claves de este diccionario.
20.1.2. Creación de las bases de datos
Tal y como se ha mostrado en el apartado |creación de una base de datos MySQL|, creamos una base de datos MySQL denominada [dbimpots-2019], propiedad del usuario [admimpots] con contraseña [mdpimpots]. En [phpMyAdmin] esto da como resultado lo siguiente:
Del mismo modo, tal y como se ha mostrado en el apartado |creación de una base de datos PostgreSQL|, creamos una base de datos PostgreSQL denominada [dbimpots-2019], propiedad del usuario [admimpots] con la contraseña [mdpimpots]. En [pgAdmin] esto da como resultado lo siguiente:
Las bases de datos se han creado, pero por el momento no tienen ninguna tabla. Estas serán creadas por ORM y [sqlalchemy].
20.1.3. Las entidades mapeadas por [sqlalchemy]
Vamos a crear dos tablas para encapsular los datos de [admindata.json]:
Definida por [sqlalchemy], la tabla [tbtranches] recopilará los datos de las tablas [limites, coeffr, coeffn] del diccionario [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 por [sqlalchemy], la tabla [tbconstantes] recopilará las constantes del diccionario [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)
)
Las entidades que se asignarán a estas dos tablas serán las siguientes:
La entidad [Constantes] encapsula las constantes del diccionario [admindata.json]:
from BaseEntity import BaseEntity
# clase contenedora de los datos de la administración tributaria
class Constantes(BaseEntity):
# claves excluidas del estado de la clase
excluded_keys = ["_sa_instance_state"]
# claves 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"]
- línea 5: la clase [Constantes] extiende la clase [BaseEntity];
- línea 7: mediante la asignación [sqlalchemy], la clase [Constante] recibirá la propiedad [_sa_instance_state]. La excluimos del diccionario [asdict] de la entidad;
- líneas 11-23: las propiedades de la entidad. Se han retomado los nombres utilizados en el diccionario [admindata.json] para facilitar la escritura del código;
La entidad [Tranche] encapsula una línea de las tres tablas [limites, coeffr, coeffn] del diccionario [admindata.json]:
from BaseEntity import BaseEntity
# clase contenedora de los datos de la administración tributaria
class Tranche(BaseEntity):
# claves excluidas del informe de la clase
excluded_keys = ["_sa_instance_state"]
# claves autorizadas
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
- línea 5: la clase [Tranche] extiende la clase [BaseEntity];
- línea 7: se excluye de las propiedades del diccionario [asdict] de la entidad la propiedad [_sa_instance_state] añadida por [sqlalchemy];
- líneas 10-12: las propiedades de la clase;
La correspondencia entre las entidades [Constantes, Tranche] y las tablas [constantes, tranches] será la siguiente:
…
# la tabla 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)
)
# tabla de tramos del impuesto
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)
)
# las asignaciones
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
- Las asignaciones se realizan en las líneas 24-29. Se ha omitido establecer las correspondencias entre las propiedades de las entidades asignadas y las tablas de la base de datos. Esto es posible cuando los nombres de las columnas de las tablas son los mismos que los de las propiedades a las que deben asociarse. Por este motivo, hemos incluido en las tablas los nombres de las propiedades de las entidades asignadas. Esto facilita la escritura del código y su comprensión;
20.1.4. El archivo de configuración de [sqlalchemy]
Acabamos de detallar una parte de la configuración de [sqlalchemy]. El archivo [config_database] en su totalidad es el siguiente:
def configure(config: dict) -> dict:
# configuración de SQLAlchemy
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# cadenas de conexión a las bases de datos utilizadas
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# cadena de conexión a la base de datos en uso
engine = create_engine(connection_strings[config['sgbd']])
# metadatos
metadata = MetaData()
# la tabla 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)
)
# tabla de tramos impositivos
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)
)
# las asignaciones
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
# la fábrica de sesiones
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# una sesión
session = session_factory()
# se registra cierta información
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# resultado
return config
- línea 1: la función [configure] recibe como parámetro un diccionario cuya clave [sgbd] le indica qué SGBD debe utilizar: MySQL (mysql) o PostgreSQL (pgres);
- líneas 6-12: se selecciona la base de datos solicitada por la configuración;
- líneas 14-44: asignaciones de entidades/tablas. Estas asignaciones son sencillas, ya que no existe ningún vínculo entre las tablas [tranches] y [constantes]. Son independientes. Por lo tanto, no hay que gestionar ninguna clave externa de una sobre la otra;
- líneas 46-51: se crea la sesión de trabajo [session] de la aplicación;
- líneas 53-58: la información útil se introduce en el diccionario de configuración y este se devuelve;
20.1.5. La capa [dao]
Volvamos a la arquitectura de la aplicación 1 que hay que construir:
La capa [dao] [1] debe leer el archivo [admindata.json] [2] y transferir su contenido a una de las bases [3, 4];
La capa [dao] presenta la interfaz [1] y está implementada por la clase [2].
La interfaz [InterfaceDao4TransferAdminData2Database] es la siguiente:
# importaciones
from abc import ABC, abstractmethod
# interfaz InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# transferencia de datos fiscales a una base de datos
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
- líneas 8-10: la interfaz solo presenta un método [transfer_admindata_in_database] sin parámetros. Dado que este método necesita parámetros (¿qué archivo?, ¿qué base de datos?), esto significa que estos se pasarán al constructor de las clases que implementan esta interfaz;
La clase [DaoTransferAdminDataFromJsonFile2Database] implementa la interfaz [InterfaceDao4TransferAdminData2Database] de la siguiente manera:
# importaciones
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
# transferencia
def transfer_admindata_in_database(self) -> None:
# inicializaciones
session = None
config = self.config
try:
# se recuperan los datos de la administración tributaria
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# transferencia del contenido a un diccionario
admindata = json.load(fd)
# se recupera la configuración de la base de datos
database = config["database"]
# eliminación de las dos tablas de la base de datos
# checkfirst=True: comprueba primero si la tabla existe
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# recreación de las tablas a partir de las asignaciones
database["metadata"].create_all(database["engine"])
# la sesión [sqlalchemy] actual
session = database["session"]
# se rellena la tabla de tramos del impuesto
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]}))
# se rellena la tabla 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"]
}))
# validación de la sesión [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# Se vuelve a lanzar la excepción de otra forma
raise ImpôtsError(17, f"{erreur}")
finally:
# se liberan los recursos de la sesión
if session:
session.close()
- línea 13: la clase [DaoTransferAdminDataFromJsonFile2Database] implementa la interfaz [InterfaceDao4TransferAdminData2Database];
- líneas 15-17: el constructor de la clase recibe como parámetro el diccionario de configuración. Se utilizarán las siguientes claves:
- [admindataFilename] (línea 27): el nombre del archivo jSON que contiene los datos de la administración tributaria que se van a transferir a la base de datos;
- [database] línea 32: la configuración [sqlalchemy] de la aplicación;
- líneas 34-37: eliminación de las tablas [constantes] y [tranches] si existen;
- líneas 39-40: recreación de las dos tablas;
- línea 43: se recupera la sesión [sqlalchemy] presente en la configuración;
- líneas 45-51: las tablas [limites, coeffr, coeffn] del diccionario [admindata] se colocan en la sesión. Para ello, se colocan en la sesión instancias de la entidad [Tranche];
- líneas 52-64: se introduce en la sesión una instancia de la entidad [Constantes];
- líneas 66-67: se valida la sesión. Si los datos de la sesión aún no estaban en la base de datos, se introducen en ese momento;
- líneas 68-70: gestión de un posible error;
- líneas 71-74: se cierra la sesión. Esto es posible porque la capa [dao] solo se utiliza una vez;
20.1.6. Configuración de la aplicación
La aplicación se configura mediante tres archivos [1]:
- [config] es el archivo de configuración general. Es el que configura la aplicación [main]. Para ello, se ayuda de los otros dos archivos:
- [config_database], que ya hemos analizado y que configura ORM y [sqlalchemy];
- [config_layers], que configura las capas de la aplicación;
El archivo [config] es el siguiente:
def configure(config: dict) -> dict:
# [config] tiene la clave [sgbd] que vale:
# [mysql] para gestionar una base MySQL
# [pgres] para gestionar una base PostgreSQL
import os
# paso 1 ---
# se configura el Python Path de la aplicación
# ruta absoluta de la carpeta de este script
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (a cambiar si es necesario)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# rutas absolutas de las dependencias
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",
# carpetas locales
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# se establece la ruta del sistema
from myutils import set_syspath
set_syspath(absolute_dependencies)
# paso 2 ------
# se completa la configuración de la aplicación
config.update({
# rutas absolutas de los archivos de datos
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# paso 3 ------
# Configuración de la base de datos
import config_database
config = config_database.configure(config)
# paso 4 ------
# instanciación de las capas de la aplicación
import config_layers
config = config_layers.configure(config)
# se convierte en config
return config
- líneas 8-36: se compila el Python Path de la aplicación;
- líneas 38-43: se introduce en la configuración la ruta del archivo [admindata.json];
- líneas 45-48: configuración de [sqlalchemy];
- líneas 50-53: instanciación de las capas de la aplicación;
- línea 56: se devuelve la configuración general;
El archivo [config_layers] es el siguiente:
def configure(config: dict) -> dict:
# instanciación de la capa [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# se devuelve config
return config
- líneas 3-4: instanciación de la capa [dao]. Hemos visto que el constructor de la clase [DaoTransferAdminDataFromJsonFile2Database] esperaba como parámetro el diccionario de la configuración general de la aplicación;
- línea 4: la referencia a la capa [dao] se incluye en la configuración;
- línea 7: se devuelve la configuración;
20.1.7. El script [main] de la aplicación
El script principal [main] es el siguiente:
# se espera un parámetro mysql o 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()
# se configura la aplicación
import config
config = config.configure({'sgbd': sgbd})
# se establece la ruta del sistema (syspath); ya se pueden realizar las importaciones
from ImpôtsError import ImpôtsError
# se recupera la capa [dao]
dao = config["dao"]
# código
try:
# se transfieren los datos a la base
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# se muestra el error
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# se muestra el error
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# fin
print("Terminé...")
- líneas 1-10: se espera un parámetro. Se comprueba que esté presente y sea correcto;
- líneas 12-14: se configura la aplicación (general, SQLAlchemy, capas) pasando como parámetro el tipo de SGBD elegido;
- líneas 19-20: vamos a necesitar la capa [dao]. La recuperamos;
- línea 25: se realiza la transferencia a la base de datos. Toda la información necesaria para el método [transfer_admindata_in_database] está disponible en las propiedades de la capa [dao] de la línea 20. Es ahí donde la buscará;
Tras la ejecución con la base MySQL, esta contiene los siguientes elementos (phpMyAdmin):
En la columna [3] se ven los valores asignados por MySQL a la clave primaria [id]. La numeración comienza en 1. La captura de pantalla anterior se obtuvo tras varias ejecuciones del script.
Con la base PostgreSQL, los resultados son los siguientes:
- Hacemos clic con el botón derecho en [1] y, a continuación, en [2-3];
- en [4], se ven correctamente los datos de los tramos impositivos;
Repetimos el mismo proceso para la tabla de constantes [tbconstantes]:
20.2. Aplicación 2: cálculo del impuesto en modo batch
20.2.1. Arquitectura
La aplicación de cálculo de impuestos de version 4 utilizaba la siguiente arquitectura:
La capa [dao] implementa una interfaz [InterfaceImpôtsDao]. Hemos creado una clase que implementa esta interfaz:
- [ImpôtsDaoWithAdminDataInJsonFile], que recuperaba los datos fiscales de un archivo jSON. Se trataba de la version 3;
Vamos a implementar la interfaz [InterfaceImpôtsDao] mediante una nueva clase [ImpotsDaoWithTaxAdminDataInDatabase] que recuperará los datos de la administración tributaria de una base de datos. La capa [dao], al igual que antes, escribirá los resultados en un archivo jSON y buscará los datos de los contribuyentes en un archivo de texto. Sabemos que si seguimos respetando la interfaz [InterfaceImpôtsDao], la capa [métier] no tendrá que modificarse.
La nueva arquitectura será la siguiente:
20.2.2. Configuración de la aplicación
El archivo de configuración [config_database] sigue siendo el mismo que en la aplicación 1. La configuración [config] incorpora nuevos 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"
})
- líneas 6-8: las rutas absolutas de los archivos de texto utilizados por la aplicación 2;
La configuración de las capas [config_layers] evoluciona de la siguiente manera:
def configure(config: dict) -> dict:
# instanciación de la capa dao
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instanciación de capa [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# se devuelve el config
return config
- líneas 3-4: la capa [dao] ahora se implementa mediante la clase [ImpotsDaoWithAdminDataInDatabase]. Esta clase es nueva, pero implementa la misma interfaz [InterfaceDao] que la version 4 del ejercicio de aplicación;
- líneas 7-8: la capa [métier] está implementada por la clase [ImpôtsMétier]. Es la clase utilizada en la version 4 del ejercicio de aplicación;
20.2.3. La capa [dao]
La clase de implementación [ImpotsDaoWithAdminDataInDatabase] de la interfaz [InterfaceImpôtsDao] será la siguiente:
# importaciones
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):
# constructor
def __init__(self, config: dict):
# config["taxPayersFilename"]: el nombre del archivo de texto de los contribuyentes
# config["taxPayersResultsFilename"]: el nombre del archivo jSON de los resultados
# config["errorsFilename"]: registra los errores encontrados en taxPayersFilename
# config["database"]: configuración de la base de datos
# inicialización de la clase Parent
AbstractImpôtsDao.__init__(self, config)
# almacenamiento de parámetros
self.__config = config
# admindata
self.__admindata = None
# implementación de la interfaz
def get_admindata(self):
# ¿Se ha memorizado admindata?
if self.__admindata:
return self.__admindata
# se realiza una consulta en BD
session = None
config = self.__config
try:
# una sesión
database_config = config["database"]
session = database_config["session"]
# se lee la tabla de tramos del impuesto
tranches = session.query(Tranche).all()
# se lee la tabla de constantes (una sola línea)
constantes = session.query(Constantes).first()
# se crea la instancia admindata
admindata = AdminData()
# se crean en ella las tablas de límites, 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))
# se añaden las constantes
admindata.fromdict(constantes.asdict())
# se guarda admindata
self.__admindata = admindata
# se devuelve el valor
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# se vuelve a lanzar la excepción de otra forma
raise ImpôtsError(27, f"{erreur}")
finally:
# se cierra la sesión
if session:
session.close()
Notas
- línea 11: la clase [ImpotsDaoWithAdminDataInDatabase] hereda de la clase [AbstractImpôtsDao] presentada en la version 4. Sabemos que esta última implementa la interfaz [InterfaceDao] presentada en esta misma version. Es el cumplimiento de esta interfaz lo que nos permite no cambiar la capa [métier];
- línea 13: el constructor de la clase recibe como parámetro el diccionario de configuración de la aplicación;
- línea 20: se inicializa la clase padre []. Esta implementa parcialmente la interfaz [InterfaceDao]:
- [get_taxpayers_data] lee el archivo [taxpayersdata.txt], que contiene los datos de los contribuyentes;
- [write_taxpayers_results] escribe los resultados en el archivo jSON [résultats.json];
- [get_admindata] no está implementado;
- línea 22: se almacena la configuración pasada como parámetros;
- línea 27: implementación del método [get_admindata] de la interfaz [InterfaceDao]:
- líneas 28-30: el método [get_admindata] recupera los datos de la administración tributaria en un objeto de tipo [AdminData] y almacena este objeto en [self.__admindata]. Si se llama al método [get_admindata] varias veces, no se consulta la base de datos varias veces. Solo se consulta la primera vez. Las veces siguientes, se devuelve el objeto [self.__admindata];
- líneas 36-37: se recupera la sesión [sqlalchemy] que se creó durante la configuración de la aplicación mediante [config_database];
- líneas 40: se recuperan los tramos del impuesto en una lista;
- líneas 43: se recuperan las constantes del cálculo del impuesto;
- línea 46: se crea una instancia de la clase [AdminData]. Recordemos que deriva de [BaseEntity];
- líneas 48-54: se inicializan las tablas [limites, coeffr, coeffn] de la instancia [AdminData];
- líneas 55-56: se inicializan las demás propiedades de [AdminData] con las constantes del cálculo del impuesto. Se ha tenido cuidado de dar los mismos nombres a las propiedades de las clases [AdminData] y [Constantes], lo que simplifica el código;
- líneas 57-58: la instancia [AdminData] se almacena en la capa [dao] para devolverla en las próximas llamadas al método [get_admindata];
- línea 60: se devuelve el valor solicitado por el código llamante;
- líneas 61-63: gestión de un posible error;
- líneas 64-67: la base de datos solo es objeto de una única consulta. Por lo tanto, se puede cerrar la sesión [sqlalchemy];
20.2.4. Prueba de la capa [dao]
En la version 4 de esta aplicación, habíamos creado una clase de prueba de la capa [métier]. Más concretamente, esta clase probaba tanto las capas [métier] como [dao]. Retomamos esta prueba para verificar que la capa [dao] funciona según lo esperado. De hecho, la capa [métier] no cambia.
La prueba [TestDaoMétier] es la siguiente:
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'casado': 'sí', 'hijos': 2, 'salario': 55555,
# 'impuesto': 2814, 'recargo': 0, 'descuento': 0, 'reducción': 0, 'tipo': 0,14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
métier.calculate_tax(taxpayer, admindata)
# verificación
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': 'sí', 'hijos': 3, 'salario': 200000,
# 'impuesto': 42842, 'recargo': 17283, 'descuento': 0, 'reducción': 0, 'tipo': 0,41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
métier.calculate_tax(taxpayer, admindata)
# verificaciones
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__':
# se espera un parámetro mysql o 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()
# se está configurando la aplicación
import config
config = config.configure({'sgbd': sgbd})
# capa de negocio
métier = config['métier']
try:
# datos de administración
admindata = config['dao'].get_admindata()
except BaseException as ex:
# visualización
print((f"L'erreur suivante s'est produite : {ex}"))
# fin
sys.exit()
# se envía el parámetro recibido por el script
sys.argv.pop()
# se ejecutan los métodos de prueba
print("tests en cours...")
unittest.main()
- No volveremos sobre las 11 pruebas descritas en el apartado |prueba de capa [métier] version 4|;
- líneas 37-66: vamos a ejecutar el script de pruebas como una aplicación normal y no como una prueba UnitTest. Es la línea 66 la que activará el framework UnitTest. En las pruebas anteriores, utilizábamos el método [setUp] para configurar la ejecución de cada prueba. Repetíamos 11 veces la misma configuración, ya que la función [setUp] se ejecuta antes de cada prueba. Aquí, realizamos la configuración una sola vez. Consiste en definir las variables globales [métier] en la línea 53 y [admindata] en la línea 56, que luego serán utilizadas por los métodos de [TestDaoMétier], por ejemplo, en la línea 12;
- líneas 39-47: el script de prueba espera un parámetro [mysql / pgres] que indica si se utiliza una base MySQL o PostgreSQL;
- líneas 50-51: se configura la prueba;
- línea 53: se recupera la capa [métier] de la configuración;
- línea 56: se hace lo mismo con la capa [dao]. A continuación, se recupera la instancia [admindata] que encapsula los datos necesarios para el cálculo del impuesto;
- las pruebas han demostrado que el método [unittest.main()] de la línea 66 no ignoraba el parámetro [mysql / pgres] recibido por el script, sino que le daba un significado diferente. La línea 63 hace que este método ya no tenga ningún parámetro;
Creamos dos configuraciones de ejecución:
Si ejecutamos una de estas dos configuraciones, obtenemos los siguientes 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
- líneas 5 y 7: las 11 pruebas se han superado;
Recordemos que estas pruebas solo verifican 11 casos del cálculo del impuesto. No obstante, su éxito puede bastar para que confiemos en la capa [dao].
20.2.5. El script principal
El script principal [main] es el mismo que en el version 4:
# se espera un parámetro mysql o 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()
# se configura la aplicación
import config
config = config.configure({'sgbd': sgbd})
# se establece la ruta del sistema (syspath); ya se pueden realizar las importaciones
from ImpôtsError import ImpôtsError
# se recuperan las capas de la aplicación (ya están instanciadas)
dao = config["dao"]
métier = config["métier"]
try:
# recuperación de los tramos del impuesto
admindata = dao.get_admindata()
# lectura de los datos de los contribuyentes
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# ¿de los contribuyentes?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# cálculo del impuesto de los contribuyentes
for taxPayer in taxpayers:
# taxPayer es tanto un parámetro de entrada como de salida
# taxPayer va a ser modificado
métier.calculate_tax(taxPayer, admindata)
# escritura de los resultados en un archivo de texto
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# Visualización del error
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# finalizado
print("Travail terminé...")
Notas
- líneas 1-10: se recupera el parámetro [mysql / pgres] que indica el SGBD que se debe utilizar;
- líneas 12-14: se configura la aplicación;
- líneas 16-17: se importa la clase [ImpôtsError]. La necesitamos en la línea 38;
- líneas 19-21: se recuperan referencias de las capas de la aplicación;
- línea 25: se solicitan a la capa [dao] los datos de la administración tributaria. La capa [métier] los necesita para el cálculo del impuesto;
- línea 27: se recuperan en una lista los datos (id, estado civil, hijos, salario) de los contribuyentes;
- líneas 29-30: si esta lista está vacía, se lanza una excepción;
- líneas 32-35: cálculo del impuesto de los elementos de la lista [taxpayers];
- línea 37: escritura de los resultados en el archivo jSON[résultats.json];
- líneas 38-40: gestión de posibles errores;
Para la ejecución del script, se crean dos |configuraciones de ejecución|:
Los resultados obtenidos en el archivo [résultats.json] son los de version 4.
20.3. Aplicación 3: cálculo del impuesto en modo interactivo
Ahora presentamos la aplicación que permite calcular el impuesto de forma interactiva. Se trata de una adaptación de la aplicación 2 de version 4.
- el script [main] inicia el diálogo con el usuario mediante el método [ui.run] de la capa [ui];
- la capa [ui]:
- utiliza la capa [dao] para obtener los datos necesarios para calcular el impuesto;
- solicita al usuario la información relativa al contribuyente cuyo impuesto se desea calcular;
- utiliza la capa [métier] para realizar dicho cálculo;
El archivo [config_layers] instancia una capa adicional:
def configure(config: dict) -> dict:
# instanciación de la capa dao
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instanciación de capa [métier]
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)
# se devuelve config
return config
La clase [ImpôtsConsole], líneas 11-12, es la misma que en la |version 4|.
El script principal [main] es el siguiente:
# se espera un parámetro mysql o 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()
# se configura la aplicación
import config
config = config.configure({'sgbd': sgbd})
# el syspath está configurado; se pueden realizar las importaciones
from ImpôtsError import ImpôtsError
# se recupera la capa [ui]
ui = config["ui"]
# código
try:
# ejecución de la capa [ui]
ui.run()
except ImpôtsError as ex1:
# se muestra el mensaje de error
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# se muestra el mensaje de error
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# ejecutado en todos los casos
print("Travail terminé...")
- líneas 1-10: el script espera un parámetro [mysql / pgres] que indique el SGBD que se debe utilizar;
- líneas 12-14: se configura la aplicación;
- líneas 19-20: se recupera la capa [ui] de la configuración;
- línea 25: se ejecuta;
Los resultados son idénticos a los de la |version 4|. No podía ser de otra manera, ya que todas las interfaces de la version 4 se han respetado en la version 5.