Skip to content

23. تمرين تدريبي: الإصدار 6

23.1. مقدمة

نعود الآن إلى تطبيق حساب الضرائب الخاص بنا. سنقوم بإنشاء العديد من تطبيقات الويب حوله.

في الإصدار 5 من تمرين التطبيق الخاص بنا، تم تخزين بيانات مصلحة الضرائب في قاعدة بيانات. كان هذا الإصدار 5 يتألف من تطبيقين منفصلين يشتركان في طبقات مشتركة:

  • تطبيق يحسب الضرائب في الوضع |الدفعي| للمكلفين المخزنين في ملف نصي؛
  • تطبيق يحسب الضرائب في الوضع |التفاعلي| للمكلفين الذين تم إدخال معلوماتهم عبر لوحة المفاتيح؛

كان الإصدار 5 من تطبيق حساب الضرائب الدفعي يتمتع بالبنية التالية:

Image

في النهاية، سيكون للإصدار الويب من هذا التطبيق البنية التالية:

Image

  • يتواصل عميل الويب [1] مع خادم الويب [2]، الذي يتواصل بدوره مع نظام إدارة قواعد البيانات [3]؛
  • يحتفظ خادم الويب [2] بطبقتي [الأعمال] [8] و[DAO] [9] من التطبيق الأصلي؛
  • يحتفظ التطبيق الأصلي بنصه البرمجي الرئيسي [4] وطبقة [الأعمال] الخاصة به [15]. طبقتا [الأعمال] [8] و[15] متطابقتان؛
  • يتطلب الاتصال بين العميل والخادم طبقتين إضافيتين:
    • طبقة [الويب] [7]، التي تنفذ تطبيق الويب؛
    • طبقة [DAO] [5]، التي تعمل كعميل لتطبيق الويب [7]؛

في الإصدار النهائي، يمكن إجراء حساب الضرائب الدفعي بطريقتين:

  • تتولى طبقة [الأعمال] في الخادم معالجة منطق الأعمال الخاص بحساب الضرائب. سيستخدم البرنامج النصي [الرئيسي] هذه الطريقة؛
  • يتم التعامل مع منطق الأعمال لحساب الضرائب بواسطة طبقة [الأعمال] الخاصة بالعميل. سيستخدم البرنامج النصي [main2] هذه الطريقة؛

من الآن فصاعدًا، سنقوم بتطوير العديد من تطبيقات العميل/الخادم من النوع الموصوف أعلاه، كل منها يوضح تقنية أو أكثر من تقنيات تطوير الويب الجديدة.

23.2. خادم الويب لحساب الضرائب

23.2.1. الإصدار 1

Image

النص البرمجي [server_01] هو تطبيق الويب التالي:

Image

  • في [1]، نستخدم عنوان URL معلماتي نمرر فيه ثلاث قيم:
    • [married] (نعم/لا) للإشارة إلى ما إذا كان دافع الضرائب متزوجًا أم لا؛
    • [children]: عدد أطفال المكلف؛
    • [salary]: الراتب السنوي للمكلف؛
  • في [2]، يعرض خادم الويب سلسلة JSON تحدد مبلغ الضريبة المستحقة مع مكوناتها المختلفة؛

بنية التطبيق هي كما يلي:

Image

  • يقوم المتصفح [1] بالاستعلام عن الخادم [2]. يقوم البرنامج النصي [server_01] بتنفيذ الطبقة [الويب] [2] للخادم؛
  • الطبقات [3-8] هي تلك المستخدمة بالفعل في |الإصدار 5| من تطبيق حساب الضرائب. نعيد استخدامها كما هي؛
    • يتم تعريف طبقة [الأعمال] [3] |هنا|؛
    • يتم تعريف طبقة [DAO] [4] |هنا|؛

يتم تكوين تطبيق الويب [server_01] باستخدام ثلاثة نصوص برمجية:

  • [config]، الذي يقوم بتكوين التطبيق بأكمله؛
  • [config_database]، الذي يقوم بتكوين الوصول إلى قاعدة البيانات. سنعمل مع أنظمة إدارة قواعد البيانات MySQL و PostgreSQL؛
  • [config_layers]، الذي يقوم بتكوين طبقات التطبيق؛

النص البرمجي [config] هو كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def configure(config: dict) -> dict:
    import os

    #  step 1 ------
    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))
    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  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",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  IndexController
        f"{script_dir}/../controllers",
        #  scripts [config_database, config_layers]
        script_dir,
    ]
    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  application configuration
    #  list of users authorized to use the application
    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

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

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

    #  we return the configuration
    return config
  • تأخذ الدالة [configure] قاموس [config] كحجة (السطر 1) وتعيده كنتيجة (السطر 54) بعد إثراء محتوياته. كان من الممكن الإشارة منذ وقت طويل إلى أنه لم يكن من الضروري إعادة النتيجة [config]. في الواقع، [config] هو مرجع قاموس يشترك فيه الكود المستدعي مع الكود المستدعى. وبالتالي، فإن الكود المستدعي يمتلك هذا المرجع بالفعل (السطر 1)، ولا داعي لإرجاعه مرة أخرى (السطر 54). وبالتالي، فإن كتابة: 

config=[module].configure(config) (1)

أمر زائد عن الحاجة. يكفي كتابة:


[module].configure(config) (2)

ومع ذلك، احتفظت بأسلوب الكتابة (1) لأنني اعتقدت أنه قد يوضح بشكل أفضل أن الكود الذي تم استدعاؤه يقوم بتعديل قاموس [config].

  • السطر 1: يحتوي قاموس [config] الذي تستلمه الدالة [configure] على مفتاح "sgbd" يتم أخذ قيمته من القائمة ["mysql"، "pgres"]. يعني [mysql] أن قاعدة البيانات المستخدمة تدار بواسطة MySQL، بينما يعني "pgres" أن قاعدة البيانات المستخدمة تدار بواسطة PostgreSQL؛
  • الأسطر 4–27: نسرد جميع الدلائل التي تحتوي على العناصر الضرورية لتطبيق الويب. وستكون هذه الدلائل جزءًا من مسار Python الخاص بالتطبيق (الأسطر 30–31)؛
  • الأسطر 33–40: سيُسمح لمستخدمين معينين فقط بالوصول إلى التطبيق. هنا، لدينا قائمة تحتوي على مستخدم واحد؛
  • الأسطر 43–46: يقوم البرنامج النصي [config_database] بإنشاء التكوين لقاعدة البيانات المستخدمة؛
  • السطر 46: التكوين الذي أنشأه البرنامج النصي [config_database] هو قاموس نقوم بتخزينه في التكوين العام المرتبط بمفتاح "database"؛
  • الأسطر 48–51: يقوم البرنامج النصي [config_layers] بإنشاء مثيلات لطبقات تطبيق الويب. ويعيد قاموسًا يتم تخزينه في التكوين العام تحت مفتاح "layers"؛

البرنامج النصي [config_database] هو نفسه المستخدم بالفعل في |الإصدار 5|. ونورده هنا كمرجع:

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

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

    #  metadata
    metadata = MetaData()

    #  the constants table
    constantes_table = Table("tbconstantes", metadata,
                             Column('id', Integer, primary_key=True),
                             Column('plafond_qf_demi_part', Float, nullable=False),
                             Column('plafond_revenus_celibataire_pour_reduction', Float, nullable=False),
                             Column('plafond_revenus_couple_pour_reduction', Float, nullable=False),
                             Column('valeur_reduc_demi_part', Float, nullable=False),
                             Column('plafond_decote_celibataire', Float, nullable=False),
                             Column('plafond_decote_couple', Float, nullable=False),
                             Column('plafond_impot_celibataire_pour_decote', Float, nullable=False),
                             Column('plafond_impot_couple_pour_decote', Float, nullable=False),
                             Column('abattement_dixpourcent_max', Float, nullable=False),
                             Column('abattement_dixpourcent_min', Float, nullable=False)
                             )

    #  tax bracket table
    tranches_table = Table("tbtranches", metadata,
                           Column('id', Integer, primary_key=True),
                           Column('limite', Float, nullable=False),
                           Column('coeffr', Float, nullable=False),
                           Column('coeffn', Float, nullable=False)
                           )
    #  mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

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

    #  a session
    session = session_factory()

    #  certain information is recorded and rendered in a dictionary
    return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
            "constantes_table": constantes_table, "session": session}

يقوم البرنامج النصي [config_layers] بتكوين طبقات خادم الويب. نعيد استخدام |script| الذي رأيناه من قبل:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def configure(config: dict) -> dict:
    #  instantiation of application layers

    #  dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    dao = ImpotsDaoWithAdminDataInDatabase(config)

    #  business
    from ImpôtsMétier import ImpôtsMétier
    métier = ImpôtsMétier()

    #  put layer instances in a dictionary and return them to the calling code
    return {
        "dao": dao,
        "métier": métier
    }
  • السطر 6: يتم تنفيذ طبقة [dao] باستخدام قاعدة بيانات؛
  • تم تعريف [ImpotsDaoWithAdminDataInDatabase] |هنا|؛
  • تم تعريف [BusinessTaxes] |هنا|؛

النص البرمجي الرئيسي [server_01] هو كما يلي:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

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

#  dependencies
from ImpôtsError import ImpôtsError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&children=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  initially no errors
    erreurs = []
    #  the query must have three parameters in the URL
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  retrieve marital status in URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  retrieve the number of children in the URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants [{enfants}] invalide")
        else:
            enfants = int(enfants)

    #  the salary is retrieved from the URL
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire [{salaire}] invalide")
        else:
            salaire = int(salaire)

    #  invalid parameters in the URL?
    for key in request.args.keys():
        if key not in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the customer
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)


#  hand only
if __name__ == '__main__':
    #  start the Flask server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • الأسطر 1–10: استرداد المعلمة التي تشير إلى نظام إدارة قواعد البيانات (DBMS) المطلوب استخدامه؛
  • الأسطر 12–14: باستخدام هذه المعلومات، يمكننا تكوين التطبيق. على وجه الخصوص، يتم إنشاء مسار Python؛
  • الأسطر 16–23: باستخدام مسار Python الجديد، نقوم باستيراد الوحدات النمطية الضرورية؛
  • الأسطر 25–31: استرداد البيانات من مصلحة الضرائب لحساب الضريبة؛
  • الأسطر 33–34: إنشاء مثيل لتطبيق Flask؛
  • السطر 38: لا يخدم تطبيق Flask سوى عنوان URL [/]. ويتوقع عنوان URL بتنسيق كما يلي: [/ ?married=xx&children=yy&salary=zz]، حيث:
    • xx: نعم / لا؛
    • yy: عدد الأطفال؛
    • zz: الراتب السنوي؛
  • السطور 40-89: نتحقق من صحة معلمات عنوان URL؛
  • السطر 41: سنقوم بتجميع رسائل الخطأ في قائمة [errors]؛
  • السطر 43: قد تتذكر أن معلمات عنوان URL موجودة في [request.args] (انظر |هنا|):
    • كائن [request] هو كائن Flask الذي تم استيراده في السطر 20؛
    • يعمل الكائن [request.args] كقاموس؛
  • الأسطر 43-44: نتحقق من وجود ثلاثة معلمات بالضبط (لا أقل ولا أكثر)؛
  • الأسطر 46-49: نتحقق من وجود المعلمة [married] في عنوان URL؛
  • الأسطر 50–54: إذا كانت موجودة، نتحقق من أن قيمتها بالخط الصغير، بعد حذف المسافات البيضاء في البداية والنهاية، هي "yes" أو "no"؛
  • الأسطر 56-59: نتحقق من وجود المعلمة [children] في عنوان URL؛
  • الأسطر 60–66: إذا كانت موجودة، نتحقق من أن قيمتها عدد صحيح موجب؛
  • السطر 66: تذكر أن معلمات URL وقيمها هي سلاسل نصية. يتم تحويل قيمة المعلمة [children] إلى "int"؛
  • الأسطر 68-78: بالنسبة لمعلمة [salary]، نقوم بإجراء نفس الفحوصات التي أجريناها لمعلمة [children]؛
  • الأسطر 81–83: نتحقق من عدم وجود معلمات أخرى غير [‘married’، ‘children’، ‘salary’] في عنوان URL؛
  • الأسطر 85–89: إذا لم تكن قائمة [errors] فارغة بعد كل هذه الفحوصات، نرسل قائمة الأخطاء هذه إلى العميل كسلسلة JSON مع رمز الحالة [400 Bad Request]؛

نظرًا لأننا سنحتاج غالبًا إلى إرسال سلسلة JSON ردًا على العميل لاحقًا، فقد تم تضمين الأسطر القليلة المطلوبة لذلك في الوحدة النمطية [myutils.py] التي استخدمناها بالفعل:

Image

يصبح البرنامج النصي [myutils.py] كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
#  imports
import json
import os
import sys

from flask import make_response


def set_syspath(absolute_dependencies: list):
    #  absolute_dependencies: a list of absolute folder names

    ….


#  response generation HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
    #  response body HTTP
    response = make_response(json.dumps(réponse, ensure_ascii=False))
    #  response body HTTP is jSON
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    #  we send the HTTP response
    return response, status_code
  • السطر 16: تتوقع الدالة [json_response] معلمتين:
    • [response]: القاموس الذي يحتوي على سلسلة JSON المراد إرسالها إلى عميل الويب؛
    • [status_code]: رمز حالة HTTP الخاص بالاستجابة؛
  • السطر 18: نحدد نص JSON للاستجابة؛
  • السطر 20: نضيف رأس HTTP الذي يُخبر عميل الويب أنه سيتلقى JSON؛
  • السطر 22: نرسل استجابة HTTP إلى الكود المستدعي. ويتعين على الكود المستدعي إرسالها إلى عميل الويب؛

يتغير ملف [__init__.py] على النحو التالي:


from .myutils import set_syspath, json_response

يتم تثبيت الإصدار الجديد من [myutils] ضمن الوحدات النمطية على مستوى الجهاز باستخدام الأمر [pip install .] في محطة PyCharm:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install .
Processing c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\packages
Using legacy setup.py install for myutils, since package 'wheel' is not installed.
Installing collected packages: myutils
  Attempting uninstall: myutils
    Found existing installation: myutils 0.1
    Uninstalling myutils-0.1:
      Successfully uninstalled myutils-0.1
    Running setup.py install for myutils ... done
Successfully installed myutils-0.1
  • السطر 1: يجب أن تكون في مجلد [packages] لإدخال هذا الأمر؛

يستمر كود البرنامج النصي [server_01] على النحو التالي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
…    
    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the client
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
  • السطر 10: في هذه المرحلة، تكون المعلمات المتوقعة في عنوان URL موجودة وصحيحة؛
  • السطر 10: نقوم بإنشاء كائن [TaxPayer] الذي يمثل نموذجًا للمكلف؛
  • السطر 11: نطلب من طبقة [business] حساب الضريبة. لاحظ أن العناصر التي تحسبها طبقة [business] يتم إدراجها في كائن [taxpayer] الذي يتم تمريره كمعلمة؛
  • السطر 13: يتم إرسال الاستجابة إلى عميل الويب كسلسلة JSON. هذه هي سلسلة JSON لقاموس. نربط القاموس الخاص بكائن [taxpayer] بالمفتاح [result]. لم نتمكن من وضع كائن [taxpayer] نفسه لأنه غير قابل للتسلسل في JSON؛

نقوم بإنشاء تكوينين للتنفيذ، أحدهما لـ MySQL والآخر لـ PostgreSQL:

Image

فيما يلي بعض أمثلة التنفيذ (لقد قمت بتشغيل تطبيق [server_01] ونظام إدارة قواعد البيانات، ثم طلبت عنوان URLhttp://localhost:5000/ باستخدام متصفح):

Image

Image

فيما يلي مثال على الطلب في وحدة التحكم Postman:

Image


GET /?mari%C3%A9=xx&enfants=yy&salaire=zz HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e4c5df8c-4bd6-4250-b789-b7b164db4eff
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 06:15:44 GMT
 
{"réponse": {"erreurs": ["paramètre marié [xx] invalide", "paramètre enfants [yy] invalide", "paramètre salaire [zz] invalide"]}}
  • السطر 1: تم طلب عنوان URL غير صحيح؛
  • السطر 10: يستجيب الخادم بالحالة 400 BAD REQUEST؛

23.2.2. الإصدار 2

Image

يعزل الإصدار 2 من الخادم معالجة عناوين URL في وحدة [index_controller] [5]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
#  import dependencies
import re

from flask_api import status
from werkzeug.local import LocalProxy


#  URL set: /?married=xx&children=yy&salary=zz
def execute(request: LocalProxy, config: dict) -> tuple:
    #  dependencies
    from TaxPayer import TaxPayer

    #  initially no errors
    erreurs = []
    #  the query must have three parameters
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  we retrieve the marital status of the URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  we retrieve the number of children in URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants {enfants} invalide")
        else:
            enfants = int(enfants)

    #  we recover the URL salary
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire {salaire} invalide")
        else:
            salaire = int(salaire)

    #  other parameters in the URL?
    for key in request.args.keys():
        if not key in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return résultats, status.HTTP_400_BAD_REQUEST

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
    #  we send the response to the customer
    return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
  • السطر 9: تتلقى الدالة [execute] معلمتين:
    • [request]: طلب HTTP الخاص بالعميل؛
    • [config]: قاموس تكوين التطبيق؛

النص البرمجي [server_02] هو كما يلي:

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

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

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • الأسطر 36–41: معالجة المسار /؛
  • السطر 39: استخدام الدالة [IndexController.execute]؛

سنستخدم الآن هذه التقنية: سيتم التعامل مع كل مسار بواسطة الوحدة النمطية الخاصة به.

نتائج التنفيذ هي نفسها كما في الإصدار 1.

23.2.3. الإصدار 3

تقدم النسخة 3 مفهوم المصادقة.

يصبح البرنامج النصي [server_03] كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

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

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
from flask_httpauth import HTTPBasicAuth
import index_controller

#  data recovery from tax authorities
try:
    #  config['admindata'] will be read-only application data
    config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  authentication manager
auth = HTTPBasicAuth()


#  authentication method
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
    #  user list
    users = config['users']
    #  browse this list
    for user in users:
        if user['login'] == login and user['password'] == password:
            return True
    #  we didn't find
    return False


#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • السطر 21: استيراد معالج المصادقة. هناك أنواع مختلفة من المصادقة لخادم الويب. النوع الذي نستخدمه هنا يسمى [HTTP Basic]. يتبع كل نوع من أنواع المصادقة حوارًا محددًا بين العميل والخادم؛
  • السطر 33: إنشاء مثيل لمعالج المصادقة؛
  • السطر 37: تشير التعليقة التوضيحية [@auth.verify_password] إلى الدالة التي سيتم تنفيذها عندما يرغب معالج المصادقة في التحقق من اسم المستخدم وكلمة المرور المرسلة من العميل وفقًا لبروتوكول [HTTP Basic]؛
  • السطر 55: تحدد التعليقة التوضيحية [@auth.login_required] مسارًا يجب أن يتم مصادقة عميل الويب عليه. إذا لم يكن عميل الويب قد أرسل بيانات اعتماده بعد، فسيطلبها خادم الويب تلقائيًا باستخدام بروتوكول HTTP Basic؛

يجب تثبيت الوحدة النمطية [flask_httpauth]:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install flask_httpauth
Collecting flask_httpauth
  Downloading Flask_HTTPAuth-4.1.0-py2.py3-none-any.whl (5.8 kB)
Requirement already satisfied: Flask in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from flask_httpauth) (1.1.2)
Requirement already satisfied: itsdangerous>=0.24 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.1.0)
Requirement already satisfied: click>=5.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (7.1.2)
Requirement already satisfied: Jinja2>=2.10.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (2.11.2)
Requirement already satisfied: Werkzeug>=0.15 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.0.1)
Requirement already satisfied: MarkupSafe>=0.23 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Jinja2>=2.10.1->Flask->flask_httpauth) (1.1.1
)
Installing collected packages: flask-httpauth
Successfully installed flask-httpauth-4.1.0

لنرى ما يحدث في وحدة التحكم في Postman. أنت:

  • قم بإنشاء تكوين تشغيل؛
  • قم بتشغيل تطبيق الويب؛
  • قم بتشغيل قاعدة البيانات التي تختارها؛
  • اطلب عنوان URL [/] باستخدام Postman؛

يكون الحوار بين العميل والخادم في وحدة التحكم Postman كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
GET / HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e65e2a28-4fe3-423b-88b3-b3e5a83092b1
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 19
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:05:37 GMT

Unauthorized Access
  • السطر 10: يرد الخادم بأننا غير مخولين بالوصول إلى عنوان URL [/]؛
  • السطر 13: يخبرنا ببروتوكول المصادقة الذي يجب استخدامه، وهو في هذه الحالة بروتوكول المصادقة الأساسية؛

من الممكن تكوين Postman لإرسال بيانات اعتماد المستخدم وفقًا لبروتوكول المصادقة الأساسية:

Image

  • في [6-7] ندخل بيانات الاعتماد الموجودة في البرنامج النصي [config]: Image 

    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

يصبح الحوار بين العميل والخادم في وحدة تحكم Postman كما يلي:


GET / HTTP/1.1
Authorization: Basic YWRtaW46YWRtaW4=
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 5ce20822-e87c-4eef-a2f4-b9eaec38d881
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 203
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:20:01 GMT
 
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
  • السطر 2: يرسل عميل Postman بيانات اعتماد المستخدم [admin / admin] في شكل مشفر؛
  • السطر 17: يستجيب الخادم بشكل صحيح. ويبلغ عن أخطاء لأن المعلمات [marié, enfants, salaire] لم يتم إرسالها (السطر 1)، لكنه لا يبلغ عن خطأ في المصادقة؛

الآن دعونا نطلب عنوان URL / باستخدام متصفح (Firefox أدناه):

Image

  • كما هو الحال مع Postman، تلقى Firefox استجابة HTTP من الخادم مع رؤوس HTTP التالية: 
1
2
3
4
HTTP/1.0 401 UNAUTHORIZED
…
WWW-Authenticate: Basic realm="Authentication Required"
…

لا يوقف Firefox، مثل المتصفحات الأخرى، نافذة الحوار عند تلقيه لهذه الرؤوس. بل يطلب من المستخدم بيانات الاعتماد التي يطلبها الخادم. في المثال أعلاه، سيؤدي مجرد كتابة admin / admin إلى تلقي استجابة الخادم:

Image

23.3. عميل الويب لخادم حساب الضرائب

23.3.1. مقدمة

في القسم السابق، كان عميل الويب لخادم حساب الضرائب عبارة عن متصفح. في هذا القسم، سيكون عميل الويب عبارة عن برنامج نصي للوحدة الطرفية. تصبح البنية كما يلي:

Image

  • يتكون عميل الويب من الطبقات [1-2]؛
  • يتكون خادم الويب من الطبقات [3-9]. كما ذكرنا في القسم السابق؛

لذلك نحتاج إلى كتابة الطبقات [1-2].

يجب أن تكون الطبقة [dao] [2] قادرة على التواصل مع خادم الويب [3]. نحن الآن نفهم بروتوكول HTTP ويمكننا كتابة، باستخدام الوحدة النمطية [pycurl] التي درسناها بالفعل على سبيل المثال، برنامج نصي يتواصل مع خادم الويب [3]. ومع ذلك، هناك وحدات نمطية متخصصة في اتصال عميل/خادم HTTP. سنستخدم إحداها، وهي الوحدة النمطية [requests]:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install requests
Collecting requests
  Downloading requests-2.24.0-py2.py3-none-any.whl (61 kB)
     || 61 kB 137 kB/s
Collecting idna<3,>=2.5
  Downloading idna-2.10-py2.py3-none-any.whl (58 kB)
     || 58 kB 692 kB/s
Collecting chardet<4,>=3.0.2
  Downloading chardet-3.0.4-py2.py3-none-any.whl (133 kB)
     || 133 kB 1.3 MB/s
Collecting urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1
  Downloading urllib3-1.25.9-py2.py3-none-any.whl (126 kB)
     || 126 kB 1.1 MB/s
Collecting certifi>=2017.4.17
  Downloading certifi-2020.6.20-py2.py3-none-any.whl (156 kB)
     || 156 kB 1.1 MB/s
Installing collected packages: idna, chardet, urllib3, certifi, requests
Successfully installed certifi-2020.6.20 chardet-3.0.4 idna-2.10 requests-2.24.0 urllib3-1.25.9

هيكل الدليل لبرامج نصية عميل الويب هو كما يلي:

Image

ستقوم البرنامج النصي بتنفيذ تطبيق حساب الضرائب في الوضع الدفعي الموصوف في |الإصدار 1|. أحدث إصدار من هذا التطبيق هو |الإصدار 5|. فيما يلي تذكير بكيفية عمله:

  • يتم سرد دافعي الضرائب الذين سيتم حساب الضريبة لهم في الملف النصي [taxpayersdata.txt]: 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# données valides : id, marié, enfants, salaire
1,oui,2,55555
2,oui,2,50000
3,oui,3,50000
4,non,2,100000
5,non,3,100000
6,oui,3,100000
7,oui,5,100000
8,non,0,100000
9,oui,2,30000
10,non,0,200000
11,oui,3,200000
# on crée des lignes erronées
# pas assez de valeurs
11,12
# des valeurs erronées
x,x,x,x
  • يتم حفظ النتائج في ملفين:
  • يُدرج الملف النصي [errors.txt] الأخطاء التي تم اكتشافها في ملف دافع الضرائب: 

Analyse du fichier C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt
 
Ligne 15, not enough values to unpack (expected 4, got 2)
Ligne 17, MyException[1, L'identifiant d'une entité <class 'TaxPayer.TaxPayer'> doit être un entier >=0]
  • (تابع)
    • يحتوي ملف JSON [results.json] على نتائج حساب الضرائب لمختلف دافعي الضرائب: 

[
  {
    "id": 0,
    "marié": "oui",
    "enfants": 2,
    "salaire": 55555,
    "impôt": 2814,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 0,
    "réduction": 0
  },
  {
    "id": 1,
    "marié": "oui",
    "enfants": 2,
    "salaire": 50000,
    "impôt": 1384,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 384,
    "réduction": 347
  },
  …
]

23.3.2. تكوين عميل الويب

Image

يتم إجراء التكوين باستخدام نصين برمجيين:

  • [config]، الذي يتولى جميع عمليات التكوين خارج طبقات البنية؛
  • [config_layers]، الذي يتولى تكوين طبقات البنية؛

النص البرمجي [config] هو كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def configure(config: dict) -> dict:
    import os

    #  step 1 ------

    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))

    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  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",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  ImpôtsDaoWithHttpClient
        f"{script_dir}/../services",
        #  configuration scripts
        script_dir,
    ]

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

    #  step 2 ------
    #  application configuration with constants
    config.update({
        "taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
        "resultsFilename": f"{script_dir}/../data/output/résultats.json",
        "errorsFilename": f"{script_dir}/../data/output/errors.txt",
        "server": {
            "urlServer": "http://127.0.0.1:5000/",
            "authBasic": True,
            "user": {
                "login": "admin",
                "password": "admin"
            }
        }
    }
    )

    #  step 3 ------
    #  layer instantiation
    import config_layers
    config['layers'] = config_layers.configure(config)

    #  we return the configuration
    return config
  • السطر 1: تأخذ الدالة [configure] كمعلمة القاموس الذي سيتم ملؤه بمعلومات التكوين. قد يكون هذا القاموس مملوءًا مسبقًا أو فارغًا. هنا، سيكون فارغًا؛
  • الأسطر 40–42: المسارات المطلقة للملفات النصية الثلاثة التي تديرها طبقة [dao]؛
  • الأسطر 43-50: مرتبطة بمفتاح [server]، المعلومات التي تحتاج طبقة [dao] إلى معرفتها عن خادم الويب الذي يجب أن تتواصل معه:
    • السطر 44: عنوان URL لخدمة الويب؛
    • السطر 45: يتم تعيين المفتاح [authBasic] على True إذا كان الوصول إلى عنوان URL يتطلب مصادقة أساسية؛
    • الأسطر 46–49: بيانات اعتماد المستخدم الذي سيقوم بالمصادقة إذا كانت المصادقة مطلوبة؛
  • السطور 56-57: نقوم بإنشاء مثيلات للطبقات — في هذه الحالة، طبقة [dao] الواحدة — ونضع مراجع الطبقات في [config] تحت مفتاح [layers]؛

النص البرمجي [config_layers] هو كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def configure(config: dict) -> dict:
    #  instantiation of applicatuon layers

    #  dao layer
    from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
    dao = ImpôtsDaoWithHttpClient(config)

    #  make the layer configuration
    return {
        "dao": dao
    }
  • السطر 1: تستقبل الدالة [configure] القاموس الذي يقوم بتكوين التطبيق؛
  • الأسطر 4–6: يتم إنشاء مثيل لطبقة [dao]. في السطر 6، نمرر إليها تكوين التطبيق، حيث ستجد المعلومات التي تحتاجها؛
  • الأسطر 8-11: يتم إرجاع قاموس يحتوي على مرجع إلى طبقة [dao]؛

23.3.3. البرنامج النصي الرئيسي [main]

البرنامج النصي الرئيسي [main] هو نسخة معدلة من البرنامج النصي الموجود في |الإصدار 5|:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#  configure the application
import config
config = config.configure({})

#  dependencies
from ImpôtsError import ImpôtsError

#  code
try:
    #  we recover the [dao] layer
    dao = config["layers"]["dao"]
    #  reading taxpayer data
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    #  taxpayers?
    if not taxpayers:
        raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    #  tax calculation
    for taxpayer in taxpayers:
        #  taxpayer is both an input and output parameter
        #  taxpayer will be modified
        dao.calculate_tax(taxpayer)
    #  writing results to a text file
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError  as erreur:
    #  error display
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    #  completed
    print("Travail terminé...")
  • السطران 2-3: تم تكوين التطبيق؛
  • السطر 13: توفر طبقة [dao] قائمة بالمكلفين الذين يجب حساب الضرائب عليهم؛
  • السطر 21: تحسب طبقة [dao] الضريبة لكل منهم؛
  • السطر 23: يتم حفظ النتائج في ملف JSON؛

23.3.4. تنفيذ طبقة [dao]

Image

دعونا نعيد النظر في بنية العميل/الخادم المستخدمة:

Image

  • في [2، 6]، نرى أن طبقة [dao] لها دوران:
    • تقوم بالوصول إلى نظام الملفات لقراءة بيانات دافعي الضرائب وكتابة نتائج حسابات الضرائب. لدينا بالفعل فئة |AbstractImpôtsDao| قادرة على القيام بذلك. وهي مستخدمة منذ |الإصدار 4|؛
    • تتواصل مع خادم الويب [3]؛

في |الإصدار 5|، كان البرنامج النصي الرئيسي [main] [1] يتواصل مباشرة مع طبقة [business] [4]. ونفضل عدم تغيير هذا البرنامج النصي. ولتحقيق ذلك، سنضمن أن طبقة [DAO] [2] تنفذ واجهة طبقة [business] [4]. وبهذه الطريقة، سيبدو أن البرنامج النصي الرئيسي [main] يتواصل مباشرة مع طبقة [business] [4] ويمكنه تجاهل حقيقة أنه موجود على جهاز آخر تمامًا.

يمكن أن يكون تعريف الفئة التي تنفذ طبقة [DAO] [2] كما يلي:


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
  • فئة [TaxDaoWithHttpClient]:
    • ترث من فئة [AbstractTaxDao]، مما يسمح لها بالتعامل مع نظام الملفات [6]؛
    • تنفذ واجهة [InterfaceImpôtsMétier] حتى لا تضطر إلى تغيير البرنامج النصي الرئيسي [main] في |الإصدار 5|؛

فيما يلي الكود الكامل لفئة [TaxDaoWithHttpClient]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
#  imports
import requests
from flask_api import status

from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from ImpôtsError import ImpôtsError
from InterfaceImpôtsMétier import InterfaceImpôtsMétier
from TaxPayer import TaxPayer


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):

    #  manufacturer
    def __init__(self, config: dict):
        #  parent initialization
        AbstractImpôtsDao.__init__(self, config)
        #  parameter memory
        self.__config_server = config["server"]

    #  unused [AbstractImpôtsDao] method
    def get_admindata(self) -> AdminData:
        pass

    #  tAX CALCULATION
    def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
        #  we let the exceptions rise
        #  get parameters
        params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
        #  connection with Auth Basic authentication?
        if self.__config_server['authBasic']:
            response = requests.get(
                #  URL of the queried server
                self.__config_server['urlServer'],
                #  URL parameters
                params=params,
                #  basic authentication
                auth=(
                    self.__config_server["user"]["login"],
                    self.__config_server["user"]["password"]))
        else:
            #  connection without Auth Basic authentication
            response = requests.get(self.__config_server['urlServer'], params=params)
        #  check
        print(response.text)
        #  response status code HTTP
        status_code = response.status_code
        #  we put the response jSON in a dictionary
        résultat = response.json()
        #  error if status code other than 200 OK
        if status_code != status.HTTP_200_OK:
            #  we know that the errors have been associated with the [errors] key in the response
            raise ImpôtsError(87, résultat['réponse']['erreurs'])
        #  we know that the result has been associated with the [result] key in the response
        #  modify the input parameter with this result
        taxpayer.fromdict(résultat["réponse"]["result"])
  • الأسطر 21–23: تحتوي فئة [AbstractTaxDao] (السطر 12) على طريقة مجردة [get_admindata]. يتعين علينا تنفيذها حتى لو لم نستخدمها (يتم إدارة admindata بواسطة الخادم، وليس بواسطة العميل)؛
  • السطر 26: تنتمي الطريقة [calculate_tax] إلى الواجهة [InterfaceImpôtsMétier] (السطر 12). يجب علينا تنفيذها؛
  • السطر 15: يتلقى المنشئ قاموس تكوين التطبيق كمعلمة وحيدة له؛
  • السطران 16-17: يتم تهيئة الفئة الأم [AbstractTaxDao] عن طريق تمرير تكوين التطبيق إليها، هنا أيضًا. وستجد هناك أسماء الملفات النصية الثلاثة التي تحتاج إلى إدارتها؛
  • السطران 18-19: يتم تخزين المعلومات المتعلقة بخادم الويب لحساب الضرائب محليًا داخل الفئة؛
  • السطر 26: تتلقى طريقة [calculate_tax] كائنًا من النوع |Taxpayer| كمعلمة. للامتثال لتوقيع طريقة [InterfaceImpôtsMétier.calculate_tax]، تتلقى أيضًا معلمة [admindata]، والتي من المفترض أن تغلف بيانات إدارة الضرائب. على جانب العميل، لا تتوفر لدينا هذه البيانات. ستظل هذه المعلمة دائمًا [None]. يشير هذا الحل البديل إلى أن فئة [ImpôtsMétier] كانت في البداية سيئة التصميم:
  • كان من المفترض أن تكون توقيع [calculate_tax] ببساطة: 

def calculate_tax(self, taxpayer: TaxPayer)

وكان يجب تمرير المعلمة [admindata: AdminData] إلى منشئ الفئة؛

  • السطر 27: لم يتم تغليف كود طريقة [calculate_tax] في كتلة try / catch / finally. وهذا يعني أن أي استثناءات لن يتم معالجتها وستنتقل إلى الكود المستدعي، وهو في هذه الحالة البرنامج النصي [main]. ويقوم هذا البرنامج النصي بالفعل بالتقاط جميع الاستثناءات المنقولة من طبقة [dao]؛
  • السطر 28: يتم حساب الضريبة على جانب الخادم. لذلك سنحتاج إلى التواصل معه. نقوم بذلك باستخدام وحدة [requests] المستوردة في السطر 2؛
  • الأسطر 31–43: لإرسال طلب GET إلى خادم الويب، نستخدم طريقة [requests.get]:
    • السطور 33–34: المعلمة الأولى للطريقة هي عنوان URL المراد الاتصال به؛
    • الأسطر 35-40: المعلمتان الأخريان هما معلمتان مسمايتان لا يهم ترتيبهما؛
    • السطور 35-36: يجب أن تكون قيمة المعلمة المسماة [params] عبارة عن قاموس يحتوي على المعلومات التي سيتم تضمينها في عنوان URL بالشكل [/url?param1=value1&param2=value2&…]؛
    • السطر 29: القاموس الذي يحتوي على المعلمات الثلاثة [married, children, salary] التي يتوقعها خادم الويب. لا داعي للقلق بشأن الترميز (المسمى urlencoded) الذي يجب أن تخضع له هذه المعلمات. تتولى [requests] هذه المهمة؛
    • الأسطر 37–40: المعلمة المسماة [auth] هي مجموعة مكونة من عنصرين (login، password). وهي تمثل بيانات الاعتماد للمصادقة الأساسية؛
  • السطران 44–45: هذان السطران مخصصان للأغراض التعليمية فقط (سنقوم بتعليقهما بمجرد اكتمال عملية تصحيح الأخطاء):
    • [response] تمثل استجابة HTTP للخادم؛
    • [response.text] يمثل نص المستند الموجود في هذا الرد. أثناء تصحيح الأخطاء، من المفيد التحقق مما أرسله لنا الخادم؛
  • السطر 47: [response.status_code] هو رمز حالة HTTP للاستجابة المستلمة. يرسل خادمنا ثلاثة رموز فقط:
    • 200 OK
    • 400 طلب غير صحيح
    • 500 خطأ داخلي في الخادم
  • السطر 49: يرسل خادمنا دائمًا JSON، حتى في حالة حدوث خطأ. تقوم الدالة [response.json()] بإنشاء قاموس من سلسلة JSON المستلمة. دعونا نستعرض الشكلين المحتملين لسلسلة JSON: 

{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}

{"réponse": {"result": {"id": 0, "marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
  • الأسطر 51–53: إذا لم يكن رمز الحالة 200، يتم إلقاء استثناء مع رسائل الخطأ المضمنة في الرد؛
  • السطر 56: استرجاع القاموس الناتج عن حساب الضريبة واستخدامه لتحديث معلمة الإدخال [taxpayer]؛

23.3.5. التنفيذ

لتشغيل العميل:

  • ابدأ تشغيل الخادم [server_03] باستخدام نظام إدارة قواعد البيانات (DBMS) الذي تختاره؛
  • قم بتشغيل البرنامج النصي [main] الخاص بالعميل؛

ستجد النتائج في المجلد [data/output]. وهي نفس النتائج الخاصة بالإصدار 5.

23.4. اختبارات طبقة [dao]

لنعد إلى بنية تطبيق العميل/الخادم:

Image

  • في كود العميل، تأكدنا من أن طبقة [dao] [1] توفر نفس واجهة طبقة [business] [3]. لذلك سنستخدم فئة الاختبار |TestDaoMétier|، التي درسناها سابقًا، لاختبار طبقة [business] [3]؛

سيتم تنفيذ فئة الاختبار في البيئة التالية:

Image

  • التكوين [2] مطابق للتكوين [1] الذي استعرضناه للتو؛

فئة الاختبار [TestHttpClientDao] هي كما يلي:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import unittest


class TestHttpClientDao(unittest.TestCase):

    def test_1(self) -> None:
        from TaxPayer import TaxPayer

        #  { 'married': 'yes', 'children': 2, 'salary': 55555,
        #  tax': 2814, 'surcôte': 0, 'décôte': 0, 'réduction': 0, 'taux': 0.14}
        taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
        dao.calculate_tax(taxpayer)
        #  check
        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

        #  { 'married': 'yes', 'children': 3, 'salary': 200000,
        #  tax': 42842, 'surcôte': 17283, 'décôte': 0, 'réduction': 0, 'taux': 0.41}
        taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
        dao.calculate_tax(taxpayer)
        #  checks
        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__':

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

    #  dao layer
    dao = config['layers']['dao']

    #  test methods are executed
    print("tests en cours...")
    unittest.main()

هذه الفئة مشابهة لتلك التي تمت دراستها بالفعل في الإصدار 4 من التطبيق.

  • السطران 40-41: تكوين بيئة الاختبار؛
  • السطر 44: نسترد مرجعًا إلى طبقة [DAO]؛
  • السطران 47-48: نجري الاختبارات؛

لتشغيل الاختبارات، نقوم بإنشاء |تكوين تشغيل|:

Image

  • نقوم بإنشاء تكوين تشغيل لبرنامج نصي وحدة التحكم، وليس لاختبار الوحدة؛

عند تشغيل هذا التكوين، يتم الحصول على النتائج التالية:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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/http-clients/01/tests/TestHttpClientDao.py
tests en cours...
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 55555, "impôt": 2814, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
....{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 200000, "impôt": 64210, "surcôte": 7498, "taux": 0.45, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 50000, "impôt": 1384, "surcôte": 0, "taux": 0.14, "décôte": 384, "réduction": 347}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 50000, "impôt": 0, "surcôte": 0, "taux": 0.14, "décôte": 720, "réduction": 0}}}
...{"réponse": {"result": {"marié": "non", "enfants": 2, "salaire": 100000, "impôt": 19884, "surcôte": 4480, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 3, "salaire": 100000, "impôt": 16782, "surcôte": 7176, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 100000, "impôt": 9200, "surcôte": 2180, "taux": 0.3, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 5, "salaire": 100000, "impôt": 4230, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 100000, "impôt": 22986, "surcôte": 0, "taux": 0.41, "décôte": 0, "réduction": 0}}}
....
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 30000, "impôt": 0, "surcôte": 0, "taux": 0.0, "décôte": 0, "réduction": 0}}}
----------------------------------------------------------------------
Ran 11 tests in 0.130s

OK

Process finished with exit code 0

تم اجتياز جميع الاختبارات الـ 11.

https://stahe.github.io
هذا البرنامج التعليمي الذي كتبه Serge Tahé متاح للجمهور وفقًا لشروط Creative Commons Attribution – Non-commercial – ShareAlike 3.0 Unported.