20. Ćwiczenie praktyczne: wersja 5
Opracujemy trzy aplikacje:
- aplikacja 1 zainicjuje bazę danych, która zastąpi plik [admindata.json] z wersji 4;
- aplikacja 2 obliczy podatki w trybie wsadowym;
- aplikacja 3 obliczy podatki w trybie interaktywnym;
20.1. Aplikacja 1: inicjalizacja bazy danych
Aplikacja 1 będzie miała następującą architekturę:
Jest to rozwinięcie architektury wersji 4 (punkt |Version 4|): dane podatkowe będą pobierane z bazy danych zamiast z pliku jSON. Warstwa [dao] zostanie zmodyfikowana w celu wdrożenia tej zmiany.
20.1.1. Plik [admindata.json]
Plik [admindata.json] jest taki sam jak w wersji 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
}
Jako kolumny bazy danych wykorzystamy klucze z tego słownika.
20.1.2. Tworzenie baz danych
Jak pokazano w paragrafie |création d’une base de données MySQL|, tworzymy bazę danych o numerze MySQL o nazwie [dbimpots-2019], należącą do użytkownika [admimpots] z hasłem [mdpimpots]. W [phpMyAdmin] wygląda to następująco:
Podobnie, jak pokazano w akapicie |création d’une base de données PostgreSQL|, tworzymy bazę danych o nazwie PostgreSQL, należącą do użytkownika o nazwie [dbimpots-2019] i hasle [admimpots]. W [pgAdmin] wygląda to następująco:
Bazy danych zostały utworzone, ale na razie nie zawierają żadnych tabel. Zostaną one utworzone przez ORM i [sqlalchemy].
20.1.3. Entities mapowane przez [sqlalchemy]
Utworzymy dwie tabele w celu zebrania danych z [admindata.json]:
Zdefiniowana przez [sqlalchemy] tabela [tbtranches] będzie gromadzić dane z tablic [limites, coeffr, coeffn] ze słownika [admindata.json]:
# tabela przedziałów podatkowych
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)
)
Tabela [tbconstantes], zdefiniowana przez [sqlalchemy], będzie zawierać stałe ze słownika [admindata.json]:
# tabela stałych
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)
)
Entities, które zostaną przypisane do tych dwóch tabel, to:
Entyteta [Constantes] zawiera stałe ze słownika [admindata.json]:
from BaseEntity import BaseEntity
# klasa kontenerowa danych administracji podatkowej
class Constantes(BaseEntity):
# klucze wykluczone ze stanu klasy
excluded_keys = ["_sa_instance_state"]
# klucze dozwolone
@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"]
- wiersz 5: klasa [Constantes] rozszerza klasę [BaseEntity];
- wiersz 7: poprzez mapowanie [sqlalchemy] klasa [Constante] otrzyma właściwość [_sa_instance_state]. Wykluczamy ją ze słownika [asdict] tej encji;
- wiersze 11–23: właściwości encji. Aby ułatwić pisanie kodu, wykorzystaliśmy nazwy użyte w słowniku [admindata.json];
Entyteta [Tranche] zawiera wiersz z trzech tabel [limites, coeffr, coeffn] ze słownika [admindata.json]:
from BaseEntity import BaseEntity
# klasa kontenerowa danych administracji podatkowej
class Tranche(BaseEntity):
# klucze wykluczone ze stanu klasy
excluded_keys = ["_sa_instance_state"]
# klucze dozwolone
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
- wiersz 5: klasa [Tranche] rozszerza klasę [BaseEntity];
- wiersz 7: z właściwości słownika [asdict] danego obiektu wyklucza się właściwość [_sa_instance_state] dodaną przez [sqlalchemy];
- wiersze 10–12: właściwości klasy;
Mapowanie między encjami [Constantes, Tranche] a tabelami [constantes, tranches] będzie wyglądało następująco:
…
# tabela stałych
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)
)
# tabela przedziałów podatkowych
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)
)
# mapowania
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
- mapowania zostały wykonane w wierszach 24–29. Pominięto tam przyporządkowanie właściwości mapowanych encji do tabel bazy danych. Jest to możliwe, gdy nazwy kolumn tabel są takie same jak nazwy właściwości, z którymi mają być powiązane. Z tego powodu w tabelach umieściliśmy nazwy właściwości mapowanych encji. Ułatwia to pisanie kodu i jego zrozumienie;
20.1.4. Plik konfiguracyjny [sqlalchemy]
Właśnie omówiliśmy część konfiguracji pliku [sqlalchemy]. Pełna treść pliku [config_database] wygląda następująco:
def configure(config: dict) -> dict:
# konfiguracja SQLAlchemy
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# ciągi połączeń z wykorzystywanymi bazami danych
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': „postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019”
}
# ciąg połączenia z używaną bazą danych
engine = create_engine(connection_strings[config['sgbd']])
# metadane
metadata = MetaData()
# tabela stałych
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)
)
# tabela przedziałów podatkowych
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)
)
# mapowania
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
# fabryka sesji
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# sesja
session = session_factory()
# zapisywanie niektórych informacji
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# wynik
return config
- wiersz 1: funkcja [configure] przyjmuje jako parametr słownik, którego klucz [sgbd] określa, którego z SGBD należy użyć: MySQL (mysql) lub PostgreSQL (pgres);
- wiersze 6–12: wybierana jest baza danych wymagana przez konfigurację;
- wiersze 14–44: mapowania encji na tabele. Mapowania te są proste, ponieważ nie ma żadnego powiązania między tabelami [tranches] i [constantes]. Są one niezależne. Nie ma zatem potrzeby zarządzania kluczem obcym między nimi;
- wiersze 46–51: tworzona jest sesja robocza aplikacji o nazwie [session];
- wiersze 53–58: umieszcza się niezbędne informacje w słowniku konfiguracyjnym, a następnie zwraca się ten słownik;
20.1.5. Warstwa [dao]
Wróćmy do architektury aplikacji nr 1, którą należy zbudować:
Warstwa [dao] [1] musi odczytać plik [admindata.json] [2] i przenieść jego zawartość do jednej z baz [3, 4];
Warstwa [dao] udostępnia interfejs [1] i jest implementowana przez klasę [2].
Interfejs [InterfaceDao4TransferAdminData2Database] ma następującą postać:
# importy
from abc import ABC, abstractmethod
# interfejs InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# przeniesienie danych podatkowych do bazy danych
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
- wiersze 8–10: interfejs zawiera tylko jedną metodę [transfer_admindata_in_database] bez parametrów. Ponieważ metoda ta wymaga parametrów (jaki plik?, jaka baza danych?), oznacza to, że parametry te zostaną przekazane do konstruktora klas implementujących ten interfejs;
Klasa [DaoTransferAdminDataFromJsonFile2Database] implementuje interfejs [InterfaceDao4TransferAdminData2Database] w następujący sposób:
# importy
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):
# producent
def __init__(self, config: dict):
self.config = config
# przesyłanie
def transfer_admindata_in_database(self) -> None:
# inicjalizacje
session = None
config = self.config
try:
# pobieranie danych z urzędu skarbowego
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# przeniesienie treści do słownika
admindata = json.load(fd)
# pobieranie konfiguracji bazy danych
database = config["database"]
# usuwanie dwóch tabel z bazy danych
# checkfirst=True: najpierw sprawdza, czy tabela istnieje
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# ponowne utworzenie tabel na podstawie mapowań
database["metadata"].create_all(database["engine"])
# bieżąca sesja [sqlalchemy]
session = database["session"]
# wypełnianie tabeli przedziałów podatkowych
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]}))
# wypełnianie tabeli stałych
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"]
}))
# weryfikacja sesji [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# wyjątek jest ponownie zgłaszany w innej postaci
raise ImpôtsError(17, f"{erreur}")
finally:
# zwolniono zasoby sesji
if session:
session.close()
- wiersz 13: klasa [DaoTransferAdminDataFromJsonFile2Database] implementuje interfejs [InterfaceDao4TransferAdminData2Database];
- wiersze 15–17: konstruktor klasy przyjmuje jako parametr słownik konfiguracji. Wykorzystane zostaną następujące klucze:
- [admindataFilename] (wiersz 27): nazwa pliku jSON zawierającego dane administracji podatkowej, które mają zostać przeniesione do bazy danych;
- [database], wiersz 32: konfiguracja aplikacji [sqlalchemy];
- wiersze 34–37: usunięcie tabel [constantes] i [tranches], jeśli istnieją;
- wiersze 39–40: ponowne utworzenie obu tabel;
- wiersz 43: pobieranie sesji [sqlalchemy] obecnej w konfiguracji;
- wiersze 45–51: tabele [limites, coeffr, coeffn] ze słownika [admindata] są umieszczane w sesji. W tym celu do sesji dodaje się instancje encji [Tranche];
- wiersze 52–64: instancja encji [Constantes] zostaje umieszczona w sesji;
- wiersze 66–67: sesja zostaje zatwierdzona. Jeśli dane sesji nie znajdowały się jeszcze w bazie, są tam umieszczane w tym momencie;
- wiersze 68–70: obsługa ewentualnego błędu;
- wiersze 71–74: sesja zostaje zamknięta. Jest to możliwe, ponieważ warstwa [dao] jest wykorzystywana tylko raz;
20.1.6. Konfiguracja aplikacji
Aplikacja jest konfigurowana za pomocą trzech plików o nazwie [1]:
- [config] to plik konfiguracji ogólnej. To właśnie on konfiguruje aplikację [main]. Wspierają go dwa pozostałe pliki:
- [config_database], który już omówiliśmy i który konfiguruje pliki ORM oraz [sqlalchemy];
- [config_layers], który konfiguruje warstwy aplikacji;
Plik [config] ma następującą treść:
def configure(config: dict) -> dict:
# [config] ma klucz [sgbd] o wartości:
# [mysql] do zarządzania bazą MySQL
# [pgres] do zarządzania bazą PostgreSQL
import os
# krok 1 ---
# ustalamy ścieżkę Python Path aplikacji
# absolutna ścieżka do folderu zawierającego ten skrypt
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (w razie potrzeby należy zmienić)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# absolutne ścieżki do bibliotek zależnych
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",
# foldery lokalne
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# ustawiamy ścieżkę systemową
from myutils import set_syspath
set_syspath(absolute_dependencies)
# krok 2 ------
# uzupełnianie konfiguracji aplikacji
config.update({
# absolutne ścieżki do plików danych
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# krok 3 ------
# konfiguracja bazy danych
import config_database
config = config_database.configure(config)
# krok 4 ------
# instancjonowanie warstw aplikacji
import config_layers
config = config_layers.configure(config)
# przekazujemy konfigurację
return config
- wiersze 8–36: tworzy się ścieżkę Python Path aplikacji;
- wiersze 38–43: w konfiguracji podaje się ścieżkę do pliku [admindata.json];
- wiersze 45–48: konfiguracja pliku [sqlalchemy];
- wiersze 50–53: instancjonowanie warstw aplikacji;
- wiersz 56: zwracamy ogólną konfigurację;
Plik [config_layers] ma następującą treść:
def configure(config: dict) -> dict:
# instancjonowanie warstwy [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# zwracamy konfigurację
return config
- wiersze 3–4: instancja warstwy [dao]. Widzieliśmy, że konstruktor klasy [DaoTransferAdminDataFromJsonFile2Database] oczekuje jako parametr słownika ogólnej konfiguracji aplikacji;
- wiersz 4: odwołanie do warstwy [dao] jest umieszczane w konfiguracji;
- wiersz 7: zwracamy konfigurację;
20.1.7. Skrypt [main] aplikacji
Główny skrypt [main] ma następującą postać:
# oczekuje się na parametr mysql lub 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()
# konfiguracja aplikacji
import config
config = config.configure({'sgbd': sgbd})
# ścieżka systemowa została ustalona – można przeprowadzić importy
from ImpôtsError import ImpôtsError
# pobieramy warstwę [dao]
dao = config["dao"]
# kod
try:
# przesyłanie danych do bazy
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# wyświetlany jest komunikat o błędzie
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# wyświetlany jest błąd
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# koniec
print("Terminé...")
- wiersze 1–10: oczekuje się na parametr. Sprawdzane jest, czy istnieje i czy jest poprawny;
- wiersze 12–14: konfiguruje się aplikację (ogólne ustawienia, SQLAlchemy, warstwy), przekazując jako parametr wybrany typ SGBD;
- wiersze 19–20: będziemy potrzebować warstwy [dao]. Pobieramy ją;
- wiersz 25: dokonujemy przeniesienia do bazy danych. Wszystkie informacje niezbędne dla metody [transfer_admindata_in_database] są dostępne we właściwościach warstwy [dao] z wiersza 20. To właśnie stamtąd metoda ta je pobierze;
Po wykonaniu z bazą MySQL zawiera ona następujące elementy (phpMyAdmin):
W kolumnie [3] widoczne są wartości przypisane przez MySQL do klucza głównego [id]. Numeracja zaczyna się od 1. Powyższy zrzut ekranu uzyskano po kilkukrotnym uruchomieniu skryptu.
W przypadku bazy PostgreSQL wyniki są następujące:
- klikamy prawym przyciskiem myszy na [1], a następnie na [2-3];
- w [4] widoczne są dane dotyczące przedziałów podatkowych;
Powtarzamy tę samą procedurę dla tabeli stałych [tbconstantes]:
20.2. Zastosowanie 2: obliczanie podatku w trybie wsadowym
20.2.1. Architektura
Aplikacja do obliczania podatku w wersji 4 wykorzystywała następującą architekturę:
Warstwa [dao] implementuje interfejs [InterfaceImpôtsDao]. Stworzyliśmy klasę implementującą ten interfejs:
- [ImpôtsDaoWithAdminDataInJsonFile], która pobierała dane podatkowe z pliku jSON. Była to wersja 3;
Zaimplementujemy interfejs [InterfaceImpôtsDao] za pomocą nowej klasy [ImpotsDaoWithTaxAdminDataInDatabase], która będzie pobierać dane z urzędu skarbowego z bazy danych. Warstwa [dao], tak jak dotychczas, zapisze wyniki do pliku jSON i pobierze dane podatników z pliku tekstowego. Wiemy, że jeśli nadal będziemy przestrzegać interfejsu [InterfaceImpôtsDao], warstwa [métier] nie będzie wymagała modyfikacji.
Nowa architektura będzie wyglądać następująco:
20.2.2. Konfiguracja aplikacji
Plik konfiguracyjny [config_database] pozostaje taki sam jak w aplikacji 1. Konfiguracja [config] zawiera nowe elementy:
# krok 2 ------
# uzupełniono konfigurację aplikacji
config.update({
# absolutne ścieżki do plików danych
"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"
})
- wiersze 6–8: bezwzględne ścieżki do plików tekstowych używanych przez aplikację 2;
Konfiguracja warstw [config_layers] zmienia się w następujący sposób:
def configure(config: dict) -> dict:
# instancjonowanie warstwy DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instancjonowanie warstwy [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# przekazujemy konfigurację
return config
- wiersze 3–4: warstwa [dao] jest teraz implementowana przez klasę [ImpotsDaoWithAdminDataInDatabase]. Klasa ta jest nowa, ale implementuje ten sam interfejs [InterfaceDao], co wersja 4 ćwiczenia aplikacyjnego;
- wiersze 7–8: warstwa [métier] jest implementowana przez klasę [ImpôtsMétier]. Jest to klasa używana w wersji 4 ćwiczenia praktycznego;
20.2.3. Warstwa [dao]
Klasa implementacyjna [ImpotsDaoWithAdminDataInDatabase] interfejsu [InterfaceImpôtsDao] będzie wyglądać następująco:
# importy
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):
# konstruktor
def __init__(self, config: dict):
# konfiguracja["taxPayersFilename"]: nazwa pliku tekstowego zawierającego dane podatników
# konfiguracja["taxPayersResultsFilename"]: nazwa pliku wyników jSON
# config["errorsFilename"]: zapisuje błędy wykryte w pliku taxPayersFilename
# config["database"]: konfiguracja bazy danych
# inicjalizacja klasy Parent
AbstractImpôtsDao.__init__(self, config)
# zapisywanie parametrów
self.__config = config
# dane administracyjne
self.__admindata = None
# implementacja interfejsu
def get_admindata(self):
# czy admindata zostało zapisane?
if self.__admindata:
return self.__admindata
# wysyłamy zapytanie w BD
session = None
config = self.__config
try:
# sesja
database_config = config["database"]
session = database_config["session"]
# odczytano tabelę przedziałów podatkowych
tranches = session.query(Tranche).all()
# odczytano tabelę stałych (tylko 1 wiersz)
constantes = session.query(Constantes).first()
# tworzymy instancję admindata
admindata = AdminData()
# tworzy się w niej tabele limitów: 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))
# dodaje się do niej stałe
admindata.fromdict(constantes.asdict())
# zapisuje się instancję admindata
self.__admindata = admindata
# zwracamy wartość
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# ponownie wywołujemy wyjątek w innej postaci
raise ImpôtsError(27, f"{erreur}")
finally:
# zamyka się sesję
if session:
session.close()
Uwagi
- wiersz 11: klasa [ImpotsDaoWithAdminDataInDatabase] dziedziczy po klasie [AbstractImpôtsDao] przedstawionej w wersji 4. Wiadomo, że ta ostatnia implementuje interfejs [InterfaceDao] przedstawiony w tej samej wersji. To właśnie zgodność z tym interfejsem pozwala nam nie zmieniać warstwy [métier];
- wiersz 13: konstruktor klasy otrzymuje jako parametr słownik konfiguracji aplikacji;
- wiersz 20: klasa nadrzędna [] jest inicjowana. Częściowo implementuje ona interfejs [InterfaceDao]:
- [get_taxpayers_data] odczytuje plik [taxpayersdata.txt] zawierający dane podatników;
- [write_taxpayers_results] zapisuje wyniki do pliku jSON [résultats.json];
- [get_admindata] nie jest zaimplementowany;
- wiersz 22: zapisujemy konfigurację przekazaną w parametrach;
- wiersz 27: implementacja metody [get_admindata] interfejsu [InterfaceDao]:
- wiersze 28–30: metoda [get_admindata] pobiera dane z urzędu skarbowego do obiektu typu [AdminData] i zapisuje ten obiekt w [self.__admindata]. Jeśli metoda [get_admindata] jest wywoływana wielokrotnie, nie wykonuje się wielokrotnych zapytań do bazy danych. Zapytanie jest wysyłane tylko za pierwszym razem. Przy kolejnych wywołaniach zwracany jest obiekt [self.__admindata];
- wiersze 36–37: pobierana jest sesja [sqlalchemy], która została utworzona podczas konfiguracji aplikacji przez metodę [config_database];
- wiersz 40: pobierane są przedziały podatkowe do listy;
- wiersze 43: pobierane są stałe służące do obliczania podatku;
- wiersz 46: tworzy się instancję klasy [AdminData]. Przypominamy, że klasa ta wywodzi się z klasy [BaseEntity];
- wiersze 48–54: inicjuje się tablice [limites, coeffr, coeffn] instancji [AdminData];
- wiersze 55–56: inicjuje się pozostałe właściwości klasy [AdminData] za pomocą stałych służących do obliczania podatku. Zadbano o to, aby właściwości klas [AdminData] i [Constantes] miały te same nazwy, co upraszcza kod;
- wiersze 57–58: instancja [AdminData] jest zapisywana w warstwie [dao] w celu zwrócenia jej podczas kolejnych wywołań metody [get_admindata];
- wiersz 60: zwracana jest wartość żądana przez kod wywołujący;
- wiersze 61–63: obsługa ewentualnego błędu;
- wiersze 64–67: do bazy danych kierowane jest tylko jedno zapytanie. Można zatem zamknąć sesję [sqlalchemy];
20.2.4. Test warstwy [dao]
W wersji 4 tej aplikacji stworzyliśmy klasę testującą warstwę [métier]. Dokładniej rzecz biorąc, klasa ta testowała jednocześnie warstwy [métier] i [dao]. Wykorzystujemy ten test, aby sprawdzić, czy warstwa [dao] działa zgodnie z oczekiwaniami. Warstwa [métier] pozostaje bowiem bez zmian.
Test [TestDaoMétier] wygląda następująco:
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'żonaty': 'tak', 'dzieci': 2, 'wynagrodzenie': 55555,
# 'podatek': 2814, 'dopłata': 0, 'ulga': 0, 'obniżka': 0, 'stawka': 0,14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
métier.calculate_tax(taxpayer, admindata)
# weryfikacja
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
# {'stan cywilny': 'żonaty', 'dzieci': 3, 'wynagrodzenie': 200000,
# 'podatek': 42842, 'dopłata': 17283, 'ulga': 0, 'obniżka': 0, 'stawka': 0,41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
métier.calculate_tax(taxpayer, admindata)
# weryfikacje
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__':
# oczekuje się parametru mysql lub 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()
# konfiguracja aplikacji
import config
config = config.configure({'sgbd': sgbd})
# warstwa biznesowa
métier = config['métier']
try:
# admindata
admindata = config['dao'].get_admindata()
except BaseException as ex:
# wyświetlanie
print((f"L'erreur suivante s'est produite : {ex}"))
# koniec
sys.exit()
# przekazujemy parametr otrzymany przez skrypt
sys.argv.pop()
# wykonuje się metody testowe
print("tests en cours...")
unittest.main()
- nie powracamy do 11 testów opisanych w akapicie |test warstwy [métier] wersja 4|;
- wiersze 37–66: uruchomimy skrypt testowy jako zwykłą aplikację, a nie jako test UnitTest. To właśnie wiersz 66 spowoduje uruchomienie frameworka UnitTest. W poprzednich testach używaliśmy metody [setUp] do konfiguracji wykonania każdego testu. Tę samą konfigurację powtarzaliśmy 11 razy, ponieważ funkcja [setUp] jest uruchamiana przed każdym testem. W tym przypadku konfigurację wykonujemy tylko raz. Polega ona na zdefiniowaniu zmiennych globalnych [métier] w wierszu 53 oraz [admindata] w wierszu 56, które będą następnie wykorzystywane przez metody, na przykład [TestDaoMétier] w wierszu 12;
- wiersze 39–47: skrypt testowy oczekuje parametru [mysql / pgres], który wskazuje, czy używana jest baza MySQL, czy PostgreSQL;
- wiersze 50–51: test jest skonfigurowany;
- wiersz 53: pobieramy warstwę [métier] z konfiguracji;
- wiersz 56: podobnie postępuje się z warstwą [dao]. Następnie pobierana jest instancja [admindata], która zawiera dane niezbędne do obliczenia podatku;
- testy wykazały, że metoda [unittest.main()] z wiersza 66 nie ignorowała parametru [mysql / pgres] otrzymanego przez skrypt, ale nadawała mu inne znaczenie. Wiersz 63 sprawia, że metoda ta nie ma już żadnego parametru;
Tworzymy dwie konfiguracje wykonania:
Jeśli uruchomimy jedną z tych dwóch konfiguracji, otrzymamy następujące wyniki:
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
- wiersze 5 i 7: wszystkie 11 testów zakończyło się powodzeniem;
Przypomnijmy, że testy te sprawdzają jedynie 11 przypadków obliczania podatku. Ich pomyślny wynik może jednak wystarczyć, abyśmy mieli zaufanie do warstwy [dao].
20.2.5. Skrypt główny
Główny skrypt [main] jest taki sam jak w wersji 4:
# oczekuje się na parametr mysql lub 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()
# konfiguruje się aplikację
import config
config = config.configure({'sgbd': sgbd})
# ustalono ścieżkę systemową – można przeprowadzić importy
from ImpôtsError import ImpôtsError
# pobierane są warstwy aplikacji (są one już zainicjowane)
dao = config["dao"]
métier = config["métier"]
try:
# pobieranie przedziałów podatkowych
admindata = dao.get_admindata()
# odczyt danych podatników
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# podatników?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# obliczanie podatku podatników
for taxPayer in taxpayers:
# taxPayer jest zarówno parametrem wejściowym, jak i wyjściowym
# taxPayer zostanie zmodyfikowany
métier.calculate_tax(taxPayer, admindata)
# zapis wyników do pliku tekstowego
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# wyświetlenie błędu
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# zakończono
print("Travail terminé...")
Uwagi
- wiersze 1–10: pobierany jest parametr [mysql / pgres], który wskazuje, że należy użyć skryptu SGBD;
- wiersze 12–14: aplikacja jest konfigurowana;
- wiersze 16–17: importowana jest klasa [ImpôtsError]. Jest ona potrzebna w wierszu 38;
- wiersze 19–21: pobierane są odniesienia do warstw aplikacji;
- wiersz 25: wysyłamy zapytanie do warstwy [dao] o dane z urzędu skarbowego. Warstwa [métier] potrzebuje ich do obliczenia podatku;
- wiersz 27: pobieramy z listy dane (identyfikator, stan cywilny, liczba dzieci, wynagrodzenie) podatników;
- wiersze 29–30: jeśli lista jest pusta, zgłaszany jest wyjątek;
- wiersze 32–35: obliczanie podatku na podstawie elementów listy [taxpayers];
- wiersz 37: zapis wyników do pliku jSON[résultats.json];
- wiersze 38–40: obsługa ewentualnego błędu;
W celu wykonania skryptu tworzymy dwa pliki |configurations d’exécution|:
Wyniki uzyskane w pliku [résultats.json] dotyczą wersji 4.
20.3. Aplikacja 3: obliczanie podatku w trybie interaktywnym
Przedstawiamy teraz aplikację umożliwiającą interaktywne obliczanie podatku. Jest to adaptacja aplikacji 2 z wersji 4.
- skrypt [main] uruchamia okno dialogowe z użytkownikiem za pomocą metody [ui.run] z warstwy [ui];
- warstwa [ui]:
- wykorzystuje warstwę [dao] w celu uzyskania danych niezbędnych do obliczenia podatku;
- prosi użytkownika o podanie informacji dotyczących podatnika, dla którego ma zostać obliczony podatek;
- wykorzystuje warstwę [métier] do wykonania tego obliczenia;
Plik [config_layers] instancjonuje dodatkową warstwę:
def configure(config: dict) -> dict:
# utworzenie instancji warstwy DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# utworzenie instancji warstwy [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# interfejs użytkownika
from ImpôtsConsole import ImpôtsConsole
config['ui'] = ImpôtsConsole(config)
# zwracamy konfigurację
return config
Klasa [ImpôtsConsole], wiersze 11–12, jest taka sama jak w klasie |version 4|.
Główny skrypt [main] wygląda następująco:
# oczekuje się na parametr mysql lub 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()
# konfigurujemy aplikację
import config
config = config.configure({'sgbd': sgbd})
# ścieżka systemowa jest skonfigurowana – można przeprowadzić importy
from ImpôtsError import ImpôtsError
# pobieramy warstwę [ui]
ui = config["ui"]
# kod
try:
# uruchomienie warstwy [ui]
ui.run()
except ImpôtsError as ex1:
# wyświetla się komunikat o błędzie
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# wyświetlany jest komunikat o błędzie
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# wykonane w każdym przypadku
print("Travail terminé...")
- w wierszach 1–10 skrypt oczekuje parametru [mysql / pgres], który wskazuje, którego pliku SGBD należy użyć;
- wiersze 12–14: aplikacja jest konfigurowana;
- wiersze 19–20: pobierana jest warstwa [ui] z konfiguracji;
- wiersz 25: uruchamia się ją;
Wyniki są identyczne jak w przypadku |version 4|. Nie mogło być inaczej, ponieważ wszystkie interfejsy wersji 4 zostały zachowane w wersji 5.