20. Ejercicio práctico: versión 5
Vamos a desarrollar tres aplicaciones:
- la aplicación 1 inicializará la base de datos que reemplazará al archivo [admindata.json] de la versión 4;
- la aplicación 2 realizará el cálculo de impuestos en modo por lotes;
- la aplicación 3 calculará 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 la versión 4 (párrafo |Versión 4|): los datos fiscales se encontrarán en una base de datos en lugar de estar en un archivo jSON. La capa [dao] se modificará para implementar este cambio.
20.1.1. El archivo [admindata.json]
El archivo [admindata.json] es el mismo que en la versión 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 columnas de la base de datos las claves de este diccionario.
20.1.2. Creación de las bases de datos
Tal como se mostró en el párrafo |creación de una base de datos MySQL|, creamos una base de datos MySQL llamada [dbimpots-2019], propiedad del usuario [admimpots] con la contraseña [mdpimpots]. En [phpMyAdmin], esto da como resultado lo siguiente:
Del mismo modo, tal como se mostró en el párrafo |creación de una base de datos PostgreSQL|, creamos una base de datos PostgreSQL llamada [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 contienen ninguna tabla. Estas serán generadas 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] reunirá los datos de las tablas [limites, coeffr, coeffn] del diccionario [admindata.json]:
# la 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)
)
Definida por [sqlalchemy], la tabla [tbconstantes] reunirá las constantes del diccionario [admindata.json]:
# 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)
)
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 informe 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 tomado los nombres utilizados en el diccionario [admindata.json] para facilitar la escritura del código;
La entidad [Tranche] encapsula una fila 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 estado 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] agregada por [sqlalchemy];
- líneas 10-12: las propiedades de la clase;
La asignación 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)
)
# la 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)
- Las asignaciones se realizan en las líneas 24-29. Allí se omitió 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 esta razón, 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] completo 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)
)
# la 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 a 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 relevante se ingresa 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 se va a 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. Como este método necesita parámetros (¿qué archivo?, ¿qué base de datos?), esto significa que dichos parámetros 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: primero verifica que la tabla exista
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 actual [sqlalchemy]
session = database["session"]
# se llena 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 completa 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 transferirán 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: se incorporan a la sesión las tablas [limites, coeffr, coeffn] del diccionario [admindata]. Para ello, se incorporan a la sesión instancias de la entidad [Tranche];
- líneas 52-64: se incorpora a 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 se encontraban en la base de datos, se insertan en ese momento;
- líneas 68-70: manejo 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 este el que configura la aplicación [main]. Para ello, cuenta con la 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], cuyo valor es:
# [mysql] para administrar una base de datos MySQL
# [pgres] para administrar una base de datos PostgreSQL
import os
# paso 1 ---
# se configura la ruta de Python de la aplicación
# ruta absoluta de la carpeta de este script
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (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 aplica la configuración
return config
- líneas 8-36: se construye la ruta de Python de la aplicación;
- líneas 38-43: se incluye en la configuración la ruta del archivo [admindata.json];
- líneas 45-48: configuración de [sqlalchemy];
- líneas 50-53: se instancian 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 la configuración
return config
- líneas 3-4: instanciación de la capa [dao]. Ya vimos 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:
# transferencia de 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 verifica 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 de ahí de donde la obtendrá;
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 claramente los datos de los tramos de impuestos;
Repetimos el mismo procedimiento para la tabla de constantes [tbconstantes]:
20.2. Aplicación 2: cálculo del impuesto en modo por lotes
20.2.1. Arquitectura
La aplicación de cálculo de impuestos de la versión 4 utilizaba la siguiente arquitectura:
La capa [dao] implementa una interfaz [InterfaceImpôtsDao]. Creamos una clase que implementa esta interfaz:
- [ImpôtsDaoWithAdminDataInJsonFile], que obtenía los datos fiscales de un archivo jSON. Esa era la versión 3;
Vamos a implementar la interfaz [InterfaceImpôtsDao] mediante una nueva clase [ImpotsDaoWithTaxAdminDataInDatabase] que obtendrá 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 obtendrá los datos de los contribuyentes de un archivo de texto. Sabemos que, si seguimos respetando la interfaz [InterfaceImpôtsDao], no será necesario modificar la capa [métier].
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 elementos nuevos:
# 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",
"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] cambia 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 la capa [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# se devuelve la configuración
return config
- líneas 3-4: la capa [dao] ahora está implementada por la clase [ImpotsDaoWithAdminDataInDatabase]. Esta clase es nueva, pero implementa la misma interfaz [InterfaceDao] que la versión 4 del ejercicio de aplicación;
- líneas 7-8: la capa [métier] está implementada por la clase [ImpôtsMétier]. Esta es la clase utilizada en la versión 4 del ejercicio práctico;
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
# datos de administración
self.__admindata = None
# implementación de la interfaz
def get_admindata(self):
# ¿Se ha guardado 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 impositivos
tranches = session.query(Tranche).all()
# se lee la tabla de constantes (solo una fila)
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 agregan 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 versión 4. Sabemos que esta última implementa la interfaz [InterfaceDao] presentada en esa misma versión. Es el cumplimiento de esta interfaz lo que nos permite no modificar 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 el método [get_admindata] se invoca varias veces, no se consulta la base de datos varias veces. Solo se consulta la primera vez. Las siguientes veces, 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 para el cálculo del impuesto;
- línea 46: se crea una instancia de la clase [AdminData]. Cabe recordar que deriva de [BaseEntity];
- líneas 48-54: se inicializan los arreglos [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 tuvo cuidado de asignar 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 que realiza la llamada;
- líneas 61-63: manejo de un posible error;
- líneas 64-67: solo se realiza una única consulta a la base de datos. Por lo tanto, se puede cerrar la sesión [sqlalchemy];
20.2.4. Prueba de la capa [dao]
En la versión 4 de esta aplicación, habíamos creado una clase de prueba para la capa [métier]. Más precisamente, esta clase probaba tanto la capa [métier] como la [dao]. Retomamos esta prueba para verificar que la capa [dao] funcione como se espera. 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, 'tasa': 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, 'tasa': 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 elimina 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 a abordar las 11 pruebas descritas en el párrafo |prueba de capa [métier] versión 4|;
- líneas 37-66: ejecutaremos el script de pruebas como una aplicación normal y no como una prueba UnitTest. Será la línea 66 la que active el marco de trabajo UnitTest. En las pruebas anteriores, utilizábamos el método [setUp] para configurar la ejecución de cada prueba. Repetíamos la misma configuración 11 veces, 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 demostraron 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 asignaba un significado diferente. La línea 63 se encarga de que este método ya no tenga ningún parámetro;
Creamos dos configuraciones de ejecución:
Si ejecutamos cualquiera 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. Sin embargo, su éxito puede ser suficiente para que confiemos en la capa [dao].
20.2.5. El script principal
El script principal [main] es el mismo que en la versión 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 se modificará
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 obtienen 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 genera una excepción;
- líneas 32-35: cálculo del impuesto de los elementos de la lista [taxpayers];
- línea 37: se escriben los resultados en el archivo jSON[résultats.json];
- líneas 38-40: manejo de posibles errores;
Para ejecutar el script, se crean dos |configuraciones de ejecución|:
Los resultados obtenidos en el archivo [résultats.json] corresponden a la versión 4.
20.3. Aplicación 3: cálculo del impuesto en modo interactivo
Ahora presentamos la aplicación que permite calcular el impuesto de manera interactiva. Se trata de una adaptación de la aplicación 2 de la versión 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 este 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 la capa [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# interfaz de usuario
from ImpôtsConsole import ImpôtsConsole
config['ui'] = ImpôtsConsole(config)
# se devuelve la configuración
return config
La clase [ImpôtsConsole], líneas 11-12, es la misma que en la |versión 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; ya 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 |versión 4|. No podía ser de otra manera, ya que todas las interfaces de la versión 4 se han respetado en la versión 5.