19. ORM 的使用 SQLALCHEMY

上一章已说明，在某些情况下可以编写独立于 SGBD 的代码，并采用以下架构：

在本章中，我们将使用 ORM（对象关系映射器）[sqlalchemy]，以便无论使用何种 SGBD，都能以统一的方式访问 SGBD。 一个 ORM 能够实现两点：

• 它允许脚本与 SGBD 进行交互，而无需发出 SQL 命令；
• 它使脚本无需了解每个 SGBD 的具体特性；

架构如下：

脚本现通过 ORM 与连接器分离。它通过类和方法与 ORM 进行交互。它不执行 SQL 的代码。 由ORM通过与其连接的连接器来执行此操作。它向脚本隐藏了这些连接器的具体特性。因此，脚本代码不受连接器（即SGBD）变更的影响；

所研究的脚本树结构如下：

19.1. 安装 ORM [sqlalchemy]

ORM [sqlalchemy] 以 Python 包的形式提供，需在 Python 终端中进行安装：

(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\databases\sqlalchemy>pip install sqlalchemy
Collecting sqlalchemy
  Downloading SQLAlchemy-1.3.18-cp38-cp38-win_amd64.whl (1.2 MB)
     || 1.2 MB 3.3 MB/s
Installing collected packages: sqlalchemy
Successfully installed sqlalchemy-1.3.18

19.2. 脚本 01：基础篇

• 在[1]中，将要研究的脚本。这些脚本将使用[2]中的类：BaseEntity、MyException、Personne、Utils；

19.2.1. 配置

文件 [config] 按以下方式配置应用程序：

def configure():
    # root_dir
    # 配置中相对路径的绝对路径
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    # 依赖项的绝对路径
    absolute_dependencies = [
        # BaseEntity, MyException, 人员, 工具
        f"{root_dir}/classes/02/entities",
    ]

    # 设置 syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # 类配置
    from Personne import Personne
    Personne.excluded_keys = ['_sa_instance_state']

    # 生成配置
    return {}

注释

• 第 8 行：将包含 [BaseEntity, MyException, Personne, Utils] 类的文件夹添加到 Python 路径中；
• 第 12-13 行：设置应用程序的 Python 路径；
• 第 16-17 行：可能还记得类 |BaseEntity| 有一个名为 [excluded_keys] 的类属性。 该属性是一个列表，用于存放我们不希望出现在该类字典中的类属性（asdict 函数）。此处将属性 [_sa_instance_state] 从类 [Personne] 的状态中排除。稍后我们将了解原因；

19.2.2. 脚本 [démo]

脚本 [démo] 展示了 ORM 和 [sqlalchemy] 的首次应用：

# 获取应用程序配置
import config

config = config.configure()

# 导入
from sqlalchemy import Table, Column, Integer, String, MetaData, UniqueConstraint
from sqlalchemy.orm import mapper

from Personne import Personne

# 元数据
metadata = MetaData()

# 表
personnes_table = Table("personnes", metadata,
                        Column('id', Integer, primary_key=True),
                        Column('prenom', String(30), nullable=False),
                        Column("nom", String(30), nullable=False),
                        Column("age", Integer, nullable=False),
                        UniqueConstraint('nom', 'prenom', name='uix_1')
                        )
# 映射前的 Personne 类
personne1 = Personne().fromdict({"id": 67, "prénom": "x", "nom": "y", "âge": 10})
print(f"personne1={personne1.__dict__}")

# 映射
mapper(Personne, personnes_table, properties={
    'id': personnes_table.c.id,
    '姓': personnes_table.c.prenom,
    '姓氏：personnes_table.c.nom,
    '年龄：personnes_table.c.age
})

# personne1 未被修改
print(f"personne1={personne1.__dict__}")
# Personne类已修改——它已被“增强”
personne2 = Personne().fromdict({"id": 68, "prénom": "x1", "nom": "y1", "âge": 11})
print(f"personne2={personne2.__dict__}")

注释

• 第 1-4 行：配置应用程序；
• 第 6-10 行：导入脚本所需的模块；
• 第13行：[MetaData]是[sqlalchemy]的子类；
• 第15-22行：[Table]是[sqlalchemy]的子类。它用于描述数据库中的表。 在此，我们将描述数据库 MySQL [dbpersonnes] 中的表 [personnes]，该数据库已在章节 |MySQL| 中进行过探讨；
  • 第 16 行：第一个参数 [personnes] 是被描述的表名；
  • 第 16 行：第二个参数 [metadata] 是第 13 行创建的实例 [MetaData]；
  • 第17-22行：以下每个参数均采用[sqlalchemy]特有的语法（但与SQL的语法相似）来描述表中的一列；
  • 每列均通过 [sqlalchemy] 类的实例 [Column] 进行描述；
    • 第一个参数是列名；
    • 第二个参数是其类型；
    • 后续参数均为命名参数：
      • 第 17 行：[primary_key=True]，用于指定列 [id] 是表 [personnes] 的主键；
      • 第 18 行：[nullable=False]，用于指定当向表中插入一行时，该列必须具有值；
  • 第21行：最后，类[UniqueConstraint]用于描述唯一性约束。此处指定表中的(姓, 名)列必须唯一。名为[name]的属性用于为该约束命名。此处需区分两种情况：
    • 描述现有表时，需在表的属性中查找约束名称（如 phpMyAdmin 或 pgAdmin）；
    • 描述即将创建的表。此时可自定义名称；
• 第23-25行：创建一个名为[personne1]的人员，并显示其字典[__dict__]。此处将显示：

personne1={'_BaseEntity__id': 67, '_Personne__prénom': 'x', '_Personne__nom': 'y', '_Personne__âge': 10}

• 第27-33行：进行映射，即在类[Personne]与表[personnes]之间建立对应关系。这本质上是一种[propriétés de la classe  colonnes de la table]的映射。 函数 [mapper] 在此接受三个参数：
  • 第 28 行：第一个参数是进行映射的类名称；
  • 第28行：第二个参数是要关联的表。该表即第16行创建的对象[Table]；
  • 第28行：第三个参数是一个名为[properties]的参数。这是一个字典，其中键是映射类的属性，值是映射表的列。 要指定表 [personnes_table] 的 X 列，应写为 [personnes_table.c.X]；
• 第35-36行：映射完成后，再次显示人员[personne1]。可以看到其内容未发生变化：

personne1={'_BaseEntity__id': 67, '_Personne__prénom': 'x', '_Personne__nom': 'y', '_Personne__âge': 10}

• 第37-39行：创建新人员[personne2]并显示。此时显示如下：

personne2={'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x00000259A6747FA0>, 'id': 68, 'prénom': 'x1', 'nom': 'y1', 'âge': 11}

可见字典 [__dict__] 已发生重大变更：

• （续）
  • 出现了一个新的属性 [_sa_instance_state]。可以看到它是 ORM [sqlalchemy] 的一个对象；
  • 其他属性已去除了表示其所属类的 prefix；

因此可以得出结论：第27至33行的映射操作修改了类[Personne]。

当需要显示对象 [Personne] 的状态时，通常不需要属性 [_sa_instance_state]。该属性实际上仅用于 [sqlalchemy] 的内部处理，通常与我们无关。 因此，我们在脚本 [config] 中写道：

    # 类配置
    from Personne import Personne
    Personne.excluded_keys = ['_sa_instance_state']

19.2.3. 脚本 [main]

脚本 [main] 将通过与 [sqlalchemy] 进行交互，来操作数据库 MySQL [dbpersonnes] 中的表 [personnes]。 要理解后续内容，需回顾此处的架构：

如果 [Database1] 是 [dbpersonnes] 数据库，则可见脚本与该数据库之间的连接需经过两个实体：

• 连接 SGBD 和 MySQL 的 Python 连接器；
• SGBD 与 MySQL；

脚本 [main] 将与 ORM 进行交互，后者随后将与 Python 连接器进行交互。 ORM 将通过 |MySQL| 和 |PostgreSQL| 段落中描述的工具与该连接器进行交互，特别是通过发送 SQL 命令。 脚本 [main] 不会使用 SQL 命令。 它将基于 ORM 的 API（应用程序编程接口），该接口由类和接口组成。

[main]脚本如下：

# 配置应用程序
import config

config = config.configure()

# 导入
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, UniqueConstraint
from sqlalchemy.exc import IntegrityError, InterfaceError
from sqlalchemy.orm import mapper, sessionmaker

from Personne import Personne

# 数据库连接字符串MySQL
engine = create_engine("mysql+mysqlconnector://admpersonnes:nobody@localhost/dbpersonnes")

# 元数据
metadata = MetaData()

# 表
personnes_table = Table("personnes", metadata,
                        Column('id', Integer, primary_key=True),
                        Column('prenom', String(30), nullable=False),
                        Column("nom", String(30), nullable=False),
                        Column("age", Integer, nullable=False),
                        UniqueConstraint('nom', 'prenom', name='uix_1')
                        )

# 映射
mapper(Personne, personnes_table, properties={
    'id': personnes_table.c.id,
    '名字：personnes_table.c.prenom,
    '姓氏：personnes_table.c.nom,
    '年龄：personnes_table.c.age
})

# 会话工厂
Session = sessionmaker()
Session.configure(bind=engine)

session = None
try:
    # 一个会话
    session = Session()

    # 删除表 [personnes]
    session.execute("drop table if exists personnes")

    # 根据映射重新创建表
    metadata.create_all(engine)

    # 一次插入
    session.add(Personne().fromdict({"id": 67, "prénom": "x", "nom": "y", "âge": 10}))
    # session.commit()

    # 一个查询
    personnes = session.query(Personne).all()

    # 显示
    print("Liste des personnes ---------")
    for personne in personnes:
        print(personne)

    # 另外两次插入，其中第二次因 (名字, 姓氏) 唯一性检查失败
    session.add(Personne().fromdict({"id": 68, "prénom": "x1", "nom": "y1", "âge": 10}))
    session.add(Personne().fromdict({"id": 69, "prénom": "x1", "nom": "y1", "âge": 10}))

    # 一个查询
    personnes = session.query(Personne).all()

    # 显示
    print("Liste des personnes ---------")
    for personne in personnes:
        print(personne)

    # 会话验证
    session.commit()

except (InterfaceError, IntegrityError) as erreur:
    # 显示
    print(f"L'erreur suivante s'est produite : {erreur}")
    # 取消上一次会话
    if session:
        print("rollback...")
        session.rollback()
finally:
    # 释放会话资源
    if session:
        session.close()

注释

• 第 1-4 行：配置应用程序；
• 第 7-9 行：导入 [sqlalchemy] 库中的一系列类和接口；
• 第 11 行：导入类 [Personne]；
• 第14行：数据库连接字符串。其中指定：
  • 使用的 SGBD（mysql）；
  • 使用的 Python 连接器（mysql.connector，不带 .）；
  • 登录用户（admpersonnes）；
  • 其密码（nobody）；
  • SGBD所在的机器（localhost=运行脚本的机器）；
  • 数据库名称（dbpersonnes）；

凭借这些信息，[sqlalchemy] 即可连接到数据库。请注意，所使用的 Python 连接器必须已安装。[sqlalchemy] 不会执行此操作。

• 第 19-26 行：[personnes] 表的描述；
• 第 28-34 行：[Personne] 类与 [personnes] 表之间的映射；
• 第36-38行：大多数[sqlalchemy]操作都在一个会话中进行。[sqlalchemy]中的“会话”概念与SQL中的“事务”概念相近。 会话由第37行中的函数[sessionmaker]返回的类[Session]创建；
• 第38行：类[Session]通过第14行的连接字符串与数据库[dbpersonnes]关联；
• 第 43 行：创建一个会话。如前所述，会话可类比于事务；
• 第45-46行：方法[Session.execute]用于执行命令SQL。 这并不常见，因为我们曾提到 ORM 可避免使用 SQL 语言；
• 第48-49行：方法[metadata.create_all]用于创建所有使用第17行实例[MetaData]的表。 我们只有一个：第20-26行定义的表[personnes]。[sqlalchemy]将利用这些行的信息来创建该表。这体现了ORM的一个主要优势： 它隐藏了 SGBD 的具体细节。实际上，由于列的类型不同，SQL 和 [create] 可能与 SGBD 存在很大差异。 数据类型并未实现标准化。因此，不同 [create] 之间的顺序各不相同。在此，借助 [sqlalchemy]：
  • 我们能够以唯一的方式描述所需的表；
  • [sqlalchemy] 能够生成与当前的 SGBD 相匹配的 [create]；
• 第 52 行：向会话中添加了一个 [Personne] 对象。这不会自动将其添加到数据库中。实际上，ORM 会遵循自身的规则与数据库进行同步。它始终会试图优化其发出的查询次数。 举个例子。脚本在会话中添加（add）两个人（personne1、personne2），随后发出一个查询：它希望查看表中所有人员。[sqlalchemy] 可以这样处理：
  • [personne1] 的添加操作可在内存中完成。目前无需将其写入数据库；
  • [personne2] 亦同；
  • 接下来是 [select] 类型的查询。此时需要检索表 [personnes] 中的所有行。[sqlalchemy] 将把 [personne1, personne2] 写入数据库，然后执行该查询；

[sqlalchemy] 因此会进行对开发者而言透明的优化。

• 第56行：要执行[select]类型的查询（我想查看……），需使用[Session.query]方法。 方法 [query] 的参数是与被查询表映射的类。该方法返回类型为 [Query]。 方法 [Query.all] 请求会话中所有 [Personne] 对象。该方法返回表 [personnes] 中的所有行，每行以 [Personne] 对象的形式呈现。 为此，[sqlalchemy] 利用了 [Personne] 类与 [personnes] 表之间已建立的映射关系。第 56 行返回的结果是一组 [Personne] 对象；
• 第58-61行：显示列表[personnes]中的元素。由于类[Personne]继承自类[BaseEntity]， 因此第61行中隐式调用的[Personne.__str__]方法，实际上是[BaseEntity.__str__]方法，该方法返回调用对象的jSON字符串。 该字符串即字典 [Personne.asdict] 中的字符串 jSON（参见 |BaseEntity|）。 我们曾提到，映射完成后，每个 [Personne] 对象中都会包含 [_sa_instance_state] 属性。然而，该属性的值并非 [BaseEntity] 类型。 因此必须将其从类[Personne]的字典中排除，否则会导致显示异常。这已在脚本[config]中实现；
• 第63-65行：添加了另外两名姓名完全相同的人员。但这两个列的联合上存在唯一性约束。因此理应触发错误。这正是我们想要验证的；
• 第67-68行：再次查询数据库中所有人员的列表；
• 第70-73行：并将其显示出来；
• 第75-76行：提交（commit）该事务。顾名思义，底层事务将被提交；
• 运行时可见，由于第65行触发的异常，第67-76行将不会被执行。随后程序将跳转至第78-84行处理该异常；
• 第 78 行：如果 [sqlalchemy] 无法连接到数据库 [dbpersonnes]，则会触发异常 [InterfaceError]。 第65行触发了异常[IntegrityError]；
• 第 80 行：显示错误；
• 第 82-84 行：如果会话存在，则将其注销。这相当于回滚底层事务；
• 第 85-88 行：无论是否发生错误，都会关闭会话以释放资源；

执行结果如下：

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/databases/sqlalchemy/01/main.py
Liste des personnes ---------
{"nom": "y", "prénom": "x", "id": 67, "âge": 10}
L'erreur suivante s'est produite : (raised as a result of Query-invoked autoflush; consider using a session.no_autoflush block if this flush is occurring prematurely)
(mysql.connector.errors.IntegrityError) 1062 (23000): Duplicate entry 'y1-x1' for key 'uix_1'
[SQL: INSERT INTO personnes (id, prenom, nom, age) VALUES (%(id)s, %(prenom)s, %(nom)s, %(age)s)]
[parameters: ({'id': 68, 'prenom': 'x1', 'nom': 'y1', 'age': 10}, {'id': 69, 'prenom': 'x1', 'nom': 'y1', 'age': 10})]
(Background on this error at: http://sqlalche.me/e/13/gkpj)
rollback...

Process finished with exit code 0

• 第2-3行：首次插入后的用户列表；
• 第5行：当添加两名姓名完全相同的人员时触发的异常[IntegrityError]；
• 第6-7行：请注意失败的任务SQL。 这是一个已配置的 INSERT 命令：[sqlalchemy] 已使用唯一的 INSERT 将这两个人插入。 由此可见，系统曾尝试优化已发出的 SQL 订单；

现在，让我们通过 phpMyAdmin 查看 [personnes] 表的内容：

从 [6] 可见，该表为空。 甚至连脚本最初放入该会话中的第一位用户都不存在。这是因为该操作是在事务中进行的，而该事务已在脚本 [main] 的 [except] 子语句中被回滚。

现在，让我们在 [main] 中进行以下修改：

    # 插入
    session.add(Personne().fromdict({"id": 67, "prénom": "x", "nom": "y", "âge": 10}))
    # session.commit()

在第 2 行添加了一条记录后，我们将第 3 行解注。操作 [session.commit] 将提交底层事务，并启动一个新事务。执行后，表 [personnes] 的内容如下：

从 [6] 中可以看到，第一次插入操作已被保留。这是因为该操作是在事务 1 中执行的，而随后的错误发生在事务 2 中。

19.3. 脚本 02：[sqlalchemy] 的映射

02脚本是01脚本的一种变体。我们尝试在[config.py]中进行尽可能多的配置。现在，我们在此配置应用程序的[sqlalchemy]环境：

def configure():
    # 配置中相对路径的绝对路径
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    # 依赖项的绝对路径
    absolute_dependencies = [
        # BaseEntity, MyException, 用户, 工具
        f"{root_dir}/classes/02/entities",
    ]

    # 设置 syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # 导入
    from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, UniqueConstraint
    from sqlalchemy.orm import mapper, sessionmaker

    # 链接到数据库MySQL
    engine = create_engine("mysql+mysqlconnector://admpersonnes:nobody@localhost/dbpersonnes")

    # 元数据
    metadata = MetaData()

    # 表
    personnes_table = Table("personnes", metadata,
                            Column('id', Integer, primary_key=True),
                            Column('prenom', String(30), nullable=False),
                            Column("nom", String(30), nullable=False),
                            Column("age", Integer, nullable=False),
                            UniqueConstraint('nom', 'prenom', name='uix_1')
                            )

    # 映射
    from Personne import Personne

    mapper(Personne, personnes_table, properties={
        'id': personnes_table.c.id,
        '名字：personnes_table.c.prenom,
        '姓氏：personnes_table.c.nom,
        '年龄：personnes_table.c.age
    })

    # 会话工厂
    Session = sessionmaker()
    Session.configure(bind=engine)

    # 将这些信息放入配置中
    config = {}
    config["Session"] = Session
    config["metadata"] = metadata
    config["engine"] = engine
    config["personnes_table"] = personnes_table

    # 类配置
    from Personne import Personne
    Personne.excluded_keys = ['_sa_instance_state']

    # 加载配置
    return config

注释

• 第 2-12 行：配置 Python 路径；
• 第 14-45 行：配置 [sqlalchemy] 环境；
• 第 47-52 行：将 [sqlalchemy] 环境添加到配置字典中；
• 第 54-56 行：配置类 [Personne]；

通过此配置，脚本 [main] 变为如下内容：

# 配置应用程序
import config

config = config.configure()

# 系统路径已配置 - 进行导入
from sqlalchemy.exc import IntegrityError, DatabaseError, InterfaceError
from sqlalchemy.orm.exc import FlushError

from Personne import Personne

session = None
try:
    # 一个会话
    session = config["Session"]()

    # 删除表 [personnes]
    session.execute("drop table if exists personnes")

    # 根据映射重新创建表
    config["metadata"].create_all(config["engine"])

    # 两次插入
    session.add(Personne().fromdict({"prénom": "x", "nom": "y", "âge": 10}))
    personne = Personne().fromdict({"prénom": "x1", "nom": "y1", "âge": 7})
    session.add(personne)

    # 验证两次插入
    session.commit()

    # 一个查询
    personnes = session.query(Personne).all()

    # 显示
    print("Liste des personnes-----------")
    for personne in personnes:
        print(personne)

    # 另外两次插入，其中第二次失败
    session.add(Personne().fromdict({"prénom": "x2", "nom": "y2", "âge": 10}))
    session.add(Personne().fromdict({"prénom": "x2", "nom": "y2", "âge": 10}))

    # 一个查询
    personnes = session.query(Personne).all()

    # 显示
    print("Liste des personnes-----------")
    for personne in personnes:
        print(personne)

    # 会话验证
    session.commit()

except (FlushError, DatabaseError, InterfaceError, IntegrityError) as erreur:
    # 显示
    print(f"L'erreur suivante s'est produite : {erreur}")
    # 取消上一次会话
    if session:
        print("rollback...")
        session.rollback()
finally:
    # 显示
    print("Travail terminé...")
    # 释放会话资源
    if session:
        session.close()

执行结果如下：

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/databases/sqlalchemy/02/main.py
Liste des personnes-----------
{"âge": 10, "nom": "y", "prénom": "x", "id": 1}
{"âge": 7, "nom": "y1", "prénom": "x1", "id": 2}
L'erreur suivante s'est produite : (raised as a result of Query-invoked autoflush; consider using a session.no_autoflush block if this flush is occurring prematurely)
(mysql.connector.errors.IntegrityError) 1062 (23000): Duplicate entry 'y2-x2' for key 'uix_1'
[SQL: INSERT INTO personnes (prenom, nom, age) VALUES (%(prenom)s, %(nom)s, %(age)s)]
[parameters: {'prenom': 'x2', 'nom': 'y2', 'age': 10}]
(Background on this error at: http://sqlalche.me/e/13/gkpj)
rollback...
Travail terminé...

Process finished with exit code 0

在 phpMyAdmin 中，表 [personnes] 变为如下所示：

现在，让我们查看由 [sqlalchemy] 生成的表 [personnes]：

• 在 [6] 中，各列使用的类型；
• 在 [7] 中，可以看到 [id] 列具有 [AUTO_INCREMENT] 属性。 这意味着在向表中插入一行时，如果该行[id]列没有值，该值将由MySQL按递增方式生成：1、2、3、…… 这一特性使我们在向表中插入数据时无需关注主键的值：由 MySQL 负责生成；
• 在 [8] 中，可以看到 [id] 列是主键；
• 在 [9] 中，我们可以看到针对字段 [nom, prenom] 的唯一性约束；

19.4. 脚本 03：操作 [sqlalchemy] 会话中的实体

配置文件 [config] 与前例中的相同。 在脚本 [main] 中，我们利用 [sqlalchemy] 的方法，对表 [personnes] 执行常规操作 [INSERT, UPDATE, DELETE, SELECT]：

# 配置应用程序
import config

config = config.configure()

# 导入
from sqlalchemy import func
from sqlalchemy.exc import IntegrityError, DatabaseError, InterfaceError
from sqlalchemy.orm.session import Session
from Personne import Personne

# 显示表内容 [personnes]
def affiche_table(session: Session):
    print("----------------")
    # 一个查询
    personnes = session.query(Personne).all()
    # 显示
    affiche_personnes(personnes)

# 显示人员列表
def affiche_personnes(personnes: list):
    print("----------------")
    # 显示
    for personne in personnes:
        print(personne)


# 主 ---------------------------
session = None
try:
    # 一个会话
    session = config["Session"]()

    # 删除表 [personnes]
    # checkfirst=True：首先检查表是否存在
    config["personnes_table"].drop(config["engine"], checkfirst=True)

    # 根据映射重新创建表
    config["metadata"].create_all(config["engine"])

    # 插入数据
    session.add(Personne().fromdict({"prénom": "Pierre", "nom": "Nicazou", "âge": 35}))
    session.add(Personne().fromdict({"prénom": "Géraldine", "nom": "Colou", "âge": 26}))
    session.add(Personne().fromdict({"prénom": "Paulette", "nom": "Girondé", "âge": 56}))

    # 显示会话内容
    affiche_table(session)

    # 按姓氏字母顺序列出人员，若姓氏相同则按名字字母顺序排序
    personnes = session.query(Personne).order_by(Personne.nom.desc(), Personne.prénom.desc())

    # 显示
    affiche_personnes(personnes)

    # 按年龄降序排列，年龄在 [20,40] 区间内的人员列表
    # 然后按姓氏字母顺序排列，若姓氏相同则按名字字母顺序排列
    personnes = session.query(Personne). \
        filter(Personne.âge >= 20, Personne.âge <= 40). \
        order_by(Personne.âge.desc(), Personne.nom.asc(), Personne.prénom.asc())

    # 显示
    affiche_personnes(personnes)

    # 插入布鲁诺夫人
    bruneau = Personne().fromdict({"prénom": "Josette", "nom": "Bruneau", "âge": 46})
    session.add(bruneau)
    # 修改其年龄
    bruneau.âge = 47

    # 姓布鲁诺的人员列表
    personne = session.query(Personne).filter(func.lower(Personne.nom) == "bruneau").first()

    # 显示
    affiche_personnes([personne])

    # 删除布鲁诺夫人
    session.delete(personne)

    # 姓布鲁诺的人员列表
    personnes = session.query(Personne).filter(func.lower(Personne.nom) == "bruneau")

    # 显示
    affiche_personnes(personnes)

    # 验证会话
    session.commit()

except (DatabaseError, InterfaceError, IntegrityError) as erreur:
    # 显示
    print(f"L'erreur suivante s'est produite : {erreur}")
    # 取消上次会话
    if session:
        session.rollback()

finally:
    # 显示
    print("Travail terminé...")
    # 释放会话资源
    if session:
        session.close()

注释

• 第 20-25 行：函数 [affiche_personnes] 显示人员列表中的项目；
• 第 12-18 行：函数 [affiche_table] 显示表 [personnes] 的内容；
• 第 34-36 行：删除表 [personnes]。与以前的版本不同，这里不使用 SQL 命令，而是使用 [sqlalchemy] 的方法：
  • config["personnes_table"] 是描述表 [personnes] 的对象 [Table]；
  • config["engine"] 是连接数据库 [dbpersonnes] 的连接字符串；
  • 名为 [checkfirst=True] 的参数要求仅在表 [personnes] 存在时才执行该操作；
• 第 38-39 行：重新创建表 [personnes]；
• 第41-44行：将三名人员添加到会话中。需注意，他们未必会立即插入到表[personnes]中。这取决于[sqlalchemy]的策略，该策略旨在优化性能；
• 第46-47行：显示表[personnes]的内容。如果这三人的插入操作尚未完成，则会因本次查询而立即执行；
• 第49-50行：这是[order_by]方法的一个使用示例，该方法可按特定顺序呈现查询结果。 语法 [order_by(critère1, critère2)] 首先根据条件 [critère1] 显示结果，当行具有相同的 [critère1] 值时，则根据条件 [critère2] 进行排序。 可以像这样设置多个条件；
• 第55-59行：引入了基于[filter]方法的筛选概念。[filter(critère1, critère2)]的表示法在所用条件之间建立了一个逻辑ET（即AND）；
• 第64-67行：登录了一位新用户；
• 第70-71行：另一个过滤查询的示例。函数[func.lower(param)]将[param]转换为小写。此外还有其他可用函数，如[func.xx]。在第71行的表达式中：
  • [session.query.filter] 返回一个 [Personne] 对象列表；
  • [session.query.filter.first] 返回该列表的第一个元素；
• 第 77 行：从会话中删除一个元素；
• 第86行：会话被提交；

执行结果如下：

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/databases/sqlalchemy/03/main.py
----------------
----------------
{"âge": 35, "nom": "Nicazou", "prénom": "Pierre", "id": 1}
{"âge": 26, "nom": "Colou", "prénom": "Géraldine", "id": 2}
{"âge": 56, "nom": "Girondé", "prénom": "Paulette", "id": 3}
----------------
{"âge": 35, "nom": "Nicazou", "prénom": "Pierre", "id": 1}
{"âge": 56, "nom": "Girondé", "prénom": "Paulette", "id": 3}
{"âge": 26, "nom": "Colou", "prénom": "Géraldine", "id": 2}
----------------
{"âge": 35, "nom": "Nicazou", "prénom": "Pierre", "id": 1}
{"âge": 26, "nom": "Colou", "prénom": "Géraldine", "id": 2}
----------------
{"prénom": "Josette", "nom": "Bruneau", "âge": 47, "id": 4}
----------------
Travail terminé...

Process finished with exit code 0

• 第4-6行：会话内容；
• 第8-10行：按姓名降序排列的会话内容；
• 第12-13行：年龄在[20, 40]区间内的人员的会话内容；
• 第15行：名为“bruneau”的人员；

在 phpMyAdmin 中，执行结束时表 [personnes] 的内容如下：

19.5. 脚本 04：使用数据库 [PostgreSQL]

文件 [04] 是文件 [03] 的副本。我们仅修改了一处内容，即文件 [config] 中的连接字符串：

    # 连接到数据库PostgreSQL
    engine = create_engine("postgresql+psycopg2://admpersonnes:nobody@localhost/dbpersonnes")

现在，该连接字符串指向 [dbpersonnes] 数据库，该数据库属于 SGBD 和 [PostgreSQL]。请注意使用了连接器 [psycopg2]，必须已安装该连接器。

执行脚本 [main] 会得到以下结果：

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/databases/sqlalchemy/04/main.py
----------------
----------------
{"nom": "Nicazou", "prénom": "Pierre", "id": 1, "âge": 35}
{"nom": "Colou", "prénom": "Géraldine", "id": 2, "âge": 26}
{"nom": "Girondé", "prénom": "Paulette", "id": 3, "âge": 56}
----------------
{"nom": "Nicazou", "prénom": "Pierre", "id": 1, "âge": 35}
{"nom": "Girondé", "prénom": "Paulette", "id": 3, "âge": 56}
{"nom": "Colou", "prénom": "Géraldine", "id": 2, "âge": 26}
----------------
{"nom": "Nicazou", "prénom": "Pierre", "id": 1, "âge": 35}
{"nom": "Colou", "prénom": "Géraldine", "id": 2, "âge": 26}
----------------
{"prénom": "Josette", "nom": "Bruneau", "âge": 47, "id": 4}
----------------
Travail terminé...

Process finished with exit code 0

使用工具 [pgAdmin]（参见段落 |pgAdmin|），表 [personnes] 处于以下状态：

表 [personnes] 是使用以下代码 SQL 生成的：

• 在 [4-5] 中，可以看到 [id] 列是主键。 同时可见其默认值为 [mot clé DEFAULT]，这意味着若插入一行数据时未指定主键，系统将自动生成 SGBD 作为主键。 这是一种常见的操作方式：让 SGBD 负责生成主键；

[sqlalchemy]脚本的05版本清晰地展示了从SGBD切换到另一个版本的便捷性：只需在配置脚本中更改连接字符串即可。 其他内容均未改变。若将上文的 [id, nom, prenom, age] 列类型与示例 |02| 中 MySQL 表的列类型进行对比，可见二者存在差异。 [sqlalchemy] 会将其适配为所使用的 SGBD。 这种能够轻松适配新 SGBD 的特性，足以成为采用 [sqlalchemy] 或其他 ORM 的充分理由。

19.6. 脚本 05：完整示例

本示例延续了第 |troiscouches-v01| 节中的研究内容。该示例展示了一种三层架构 [ui, métier, dao]，用于处理 [Classe, Elève, Matière, Note] 实体。 实体原先硬编码在 [dao] 层中。现在我们将它们放入数据库中。我们将使用两个 SGBD：MySQL 和 PostgreSQL。

19.6.1. 应用程序架构

应用程序架构如下：

• 在 [1-3] 中，包含了示例 |troiscouches-v01| 中已存在的 [ui, métier, dao] 层。[dao] 层现与 [ORM] 层进行通信；
• [1-5] 层由 Python 代码实现；

19.6.2. 数据库

我们创建一个名为 MySQL 的 [dbecole] 数据库，该数据库归用户 [admecole] 所有，其密码为 [mdpecole]。 为此，我们将遵循“创建数据库”一节中描述的步骤：

• 在 [1] 中，数据库 [dbecole] 不包含表 [3]；
• 在 [7] 中，用户 [admecole] 对该数据库拥有所有权限；

对 SGBD 和 PostgreSQL 也进行同样的操作。 我们创建一个名为 [dbecole] 的数据库，所有者为用户 [admecole]，密码为 [mdpecole]。为此，我们遵循“创建数据库”一节中描述的步骤：

• 在 [1] 中，数据库 [dbecole]；
• 将 [2] 用户更改为 [admecole]；
• 在 [3-4] 中，数据库 [dbecole] 属于用户 [admecole]；

19.6.3. 应用程序处理的实体

在 |troiscouches v01| 应用程序中，处理的实体如下（参见 |实体|）。这些实体将被存储在上述数据库中。我们不会在新应用程序中重复创建这些实体，而是从它们已定义的位置获取。

类 [Classe]：

# 导入
from BaseEntity import BaseEntity
from MyException import MyException
from Utils import Utils


class Classe(BaseEntity):
    # 从类状态中排除的属性
    excluded_keys = []

    # 类属性
    @staticmethod
    def get_allowed_keys() -> list:
        # id：类标识符
        # 名称：类名
        return BaseEntity.get_allowed_keys() + ["nom"]

    # 获取器
    @property
    def nom(self: object) -> str:
        return self.__nom

    #  设置器
    @nom.setter
    def nom(self: object, nom: str):
        # 名称必须为非空字符串
        if Utils.is_string_ok(nom):
            self.__nom = nom
        else:
            raise MyException(11, f"Le nom de la classe {self.id} doit être une chaîne de caractères non vide")

类 [Elève]：

# 导入
from BaseEntity import BaseEntity
from Classe import Classe
from MyException import MyException

from Utils import Utils


class Elève(BaseEntity):
    # 从类状态中排除的属性
    excluded_keys = []

    # 类的属性
    @staticmethod
    def get_allowed_keys() -> list:
        # id：学生标识符
        # 姓：学生的姓
        # 名字：学生的名字
        # 班级：学生的班级
        return BaseEntity.get_allowed_keys() + ["nom", "prénom", "classe"]

    # 获取器
    @property
    def nom(self: object) -> str:
        return self.__nom

    @property
    def prénom(self: object) -> str:
        return self.__prénom

    @property
    def classe(self: object) -> Classe:
        return self.__classe

    #  设置器
    @nom.setter
    def nom(self: object, nom: str) -> str:
        # 姓必须为非空字符串
        if Utils.is_string_ok(nom):
            self.__nom = nom
        else:
            raise MyException(41, f"Le nom de l'élève {self.id} doit être une chaîne de caractères non vide")

    @prénom.setter
    def prénom(self: object, prénom: str) -> str:
        # 名字必须为非空字符串
        if Utils.is_string_ok(prénom):
            self.__prénom = prénom
        else:
            raise MyException(42, f"Le prénom de l'élève {self.id} doit être une chaîne de caractères non vide")

    @classe.setter
    def classe(self: object, value):
        try:
            # 预期类型为 Class
            if isinstance(value, Classe):
                self.__classe = value
            # 或字典类型
            elif isinstance(value,dict):
                self.__classe=Classe().fromdict(value)
            # 或 json 类型
            elif isinstance(value,str):
                self.__classe = Classe().fromjson(value)
        except BaseException as erreur:
            raise MyException(43, f"L'attribut [{value}] de l'élève {self.id} doit être de type Classe ou dict ou json. Erreur : {erreur}")

类 [Matière]：

# 导入
from BaseEntity import BaseEntity
from MyException import MyException
from Utils import Utils


class Matière(BaseEntity):
    # 类状态中不包含的属性
    excluded_keys = []

    # 类的属性
    @staticmethod
    def get_allowed_keys() -> list:
        # id：课程标识符
        # 名称：课程名称
        # 系数：课程系数
        return BaseEntity.get_allowed_keys() + ["nom", "coefficient"]

    # 获取器
    @property
    def nom(self: object) -> str:
        return self.__nom

    @property
    def coefficient(self: object) -> float:
        return self.__coefficient

    #  设置器
    @nom.setter
    def nom(self: object, nom: str):
        # 名称必须为非空字符串
        if Utils.is_string_ok(nom):
            self.__nom = nom
        else:
            raise MyException(21, f"Le nom de la matière {self.id} doit être une chaîne de caractères non vide")

    @coefficient.setter
    def coefficient(self, coefficient: float):
        # 系数必须为大于等于0的实数
        erreur = False
        if isinstance(coefficient, (int, float)):
            if coefficient >= 0:
                self.__coefficient = coefficient
            else:
                erreur = True
        else:
            erreur = True
        # 错误？
        if erreur:
            raise MyException(22, f"Le coefficient de la matière {self.nom} doit être un réel >=0")

[Note] 类：

# 导入
from BaseEntity import BaseEntity
from Elève import Elève
from Matière import Matière
from MyException import MyException


class Note(BaseEntity):
    # 类状态中不包含的属性
    excluded_keys = []

    # 类属性
    @staticmethod
    def get_allowed_keys() -> list:
        # id：评分标识符
        # 值：分数本身
        # 学生：与该成绩相关的学生（类型为学生）
        # 科目：与该成绩相关的科目（类型为 Matière）
        # 因此，Note 对象即为某学生在某学科中的成绩
        return BaseEntity.get_allowed_keys() + ["valeur", "élève", "matière"]

    # 获取器
    @property
    def valeur(self: object) -> float:
        return self.__valeur

    @property
    def élève(self: object) -> Elève:
        return self.__élève

    @property
    def matière(self: object) -> Matière:
        return self.__matière

    # 获取器
    @valeur.setter
    def valeur(self: object, valeur: float):
        # 成绩必须是 0 到 20 之间的实数
        if isinstance(valeur, (int, float)) and 0 <= valeur <= 20:
            self.__valeur = valeur
        else:
            raise MyException(31,
                f"L'attribut {valeur} de la note {self.id} doit être un nombre dans l'intervalle [0,20]")

    @élève.setter
    def élève(self: object, value):
        try:
            # 期望类型为 Elève
            if isinstance(value, Elève):
                self.__élève = value
            # 或字典类型
            elif isinstance(value, dict):
                self.__élève = Elève().fromdict(value)
            # 或 json 类型
            elif isinstance(value, str):
                self.__élève = Elève().fromjson(value)
        except BaseException as erreur:
            raise MyException(32,
                f"L'attribut [{value}] de la note {self.id} doit être de type Elève ou dict ou json. Erreur : {erreur}")

    @matière.setter
    def matière(self: object, value):
        try:
            # 预期类型为 Matière
            if isinstance(value, Matière):
                self.__matière = value
            # 或字典类型
            elif isinstance(value, dict):
                self.__matière = Matière().fromdict(value)
            # 或 json 类型
            elif isinstance(value, str):
                self.__matière = Matière().fromjson(value)
        except BaseException as erreur:
            raise MyException(33,
                f"L'attribut [{value}] de la note {self.id} doit être de type Matière ou dict ou json. Erreur : {erreur}")

19.6.4. 配置

配置已拆分为多个文件：

• 通用配置位于 [config.py] 中：它设置了应用程序的 Python 路径并实例化了架构层；
• [sqlalchemy]的配置位于[config_database]中：负责类与表的映射；
• 应用程序的各层配置在 [config_layers] 中；

文件 [config] 内容如下：

def configure(config: dict) -> dict:
    import os

    # 步骤 1 ---
    # 设置应用程序的 Python 路径
    # 此脚本所在文件夹的绝对路径
    script_dir = os.path.dirname(os.path.abspath(__file__))

    # 配置中相对路径的绝对路径
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    # 依赖项的绝对路径
    absolute_dependencies = [
        # BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        # 三层项目 v01
        f"{root_dir}/troiscouches/v01/interfaces",
        f"{root_dir}/troiscouches/v01/services",
        f"{root_dir}/troiscouches/v01/entities",
        # 本项目文件
        script_dir,
        f"{script_dir}/../services",
    ]

    # syspath 更新
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # 步骤 2------
    # 数据库配置
    import config_database
    config = config_database.configure(config)

    # 步骤 3 ------
    # 应用层实例化
    import config_layers
    config = config_layers.configure(config)

    # 提交配置
    return config

• 第 4-27 行：构建应用程序的 Python 路径；
• 第29-32行：配置[sqlalchemy]；
• 第34-37行：应用程序层的配置；

文件 [config_database] 内容如下：

def configure(config: dict) -> dict:
    # config['sgbd'] 是所用 SGBD 的名称
    # MySQL：MySQL
    # pgres：PostgreSQL

    # SQLAlchemy 配置
    from sqlalchemy import Table, Column, Integer, MetaData, String, Float, ForeignKey, create_engine

    from sqlalchemy.orm import mapper, relationship, sessionmaker

    # 所用数据库的连接字符串
    engines = {
        'mysql': "mysql+mysqlconnector://admecole:mdpecole@localhost/dbecole",
        'pgres': "postgresql+psycopg2://admecole:mdpecole@localhost/dbecole"
    }
    # 正在使用的数据库连接字符串
    engine = create_engine(engines[config['sgbd']])

    # 元数据
    metadata = MetaData()

    # 数据库表
    tables = {}
    # 映射类
    from Classe import Classe
    from Elève import Elève
    from Note import Note
    from Matière import Matière

    # 类表
    tables['classes'] = classes_table = \
        Table("classes", metadata,
              Column('id', Integer, primary_key=True),
              Column('nom', String(30), nullable=False),
              )

    mapper(Classe, tables['classes'], properties={
        'id': classes_table.c.id,
        '姓名：classes_table.c.nom
    })

    # 学生表
    tables['élèves'] = élèves_table = \
        Table("élèves", metadata,
              Column('id', Integer, primary_key=True),
              Column('nom', String(30), nullable=False),
              Column('prénom', String(30), nullable=False),
              # 学生所属班级
              Column('classe_id', Integer, ForeignKey('classes.id')),
              )
    # 映射
    mapper(Elève, tables['élèves'], properties={
        'id': élèves_table.c.id,
        '姓氏：élèves_table.c.nom,
        '姓': élèves_table.c.prénom,
        '班级：关系(班级, backref="学生", lazy="select")
    })

    # 目录
    tables['matières'] = matières_table = \
        Table("matières", metadata,
              Column('id', Integer, primary_key=True),
              Column('nom', String(30), nullable=False),
              Column('coefficient', Float, nullable=False)
              )
    # 映射
    mapper(Matière, tables['matières'], properties={
        'id': matières_table.c.id,
        '名称：matières_table.c.nom,
        "coefficient": matières_table.c.coefficient
    })

    # 成绩表
    tables['notes'] = notes_table = \
        Table("notes", metadata,
              Column('id', Integer, primary_key=True),
              Column('valeur', Float, nullable=False),
              # 某条成绩属于某位学生
              Column('élève_id', Integer, ForeignKey('élèves.id')),
              # 某科目的成绩
              Column('matière_id', Integer, ForeignKey('matières.id')),
              )

    # 映射
    mapper(Note, tables['notes'], properties={
        'id': notes_table.c.id,
        '值：notes_table.c.valeur,
        '学生': relationship(学生, backref="笔记", lazy="select"),
        '科目': relationship(科目, backref="成绩", lazy="select")
    })

    # 实体配置 [BaseEntity]
    Elève.excluded_keys = ['_sa_instance_state', 'notes', 'classe']
    Classe.excluded_keys = ['_sa_instance_state', 'élèves']
    Matière.excluded_keys = ['_sa_instance_state', 'notes']
    Note.excluded_keys = ['_sa_instance_state', 'matière', 'élève']

    # 会话工厂
    Session = sessionmaker()
    Session.configure(bind=engine)

    # 一个会话
    session = Session()

    # 将某些信息记录到配置字典中
    config['database'] = {"engine": engine, "metadata": metadata, "tables": tables, "session": session}

    # 加载配置
    return config

注释

• 第 1-4 行：函数 [configure] 接收一个字典作为参数。仅使用键 [sgbd]。 若数据库为 MySQL，则该键值为 [mysql]；若数据库为 [pgres]，则该键值为 [pgres]；
• 第 6-9 行：导入 [sqlalchemy] 中的元素。 脚本 [config_database] 负责在数据库 [dbecole] 的表与实体 [Classes, Elève, Matière, Note] 之间建立映射关系。在表中，实体的数据被封装在一行中。 而在 Python 代码中，这些数据被封装在一个对象中。因此得名 ORM（对象关系映射器）：ORM 在关系型数据库的行与对象之间建立映射（关联）。 在此应用中，我们有四个实体 [Classe, Elève, Matière, Note]，它们将与四个表 [classes, élèves, matières, notes] 建立关联。请注意，表名中可以包含带重音的字符；
• 第 11-17 行：所用数据库的连接字符串。该字符串取决于 config[‘sgbd’] 元素；
• 第 24-28 行：将进行映射的应用程序实体 [sqlalchemy]。当执行这些行时，Python 路径已由脚本 [config] 建立；
• 第 30-40 行：实体 [Classe] 与表 [classes] 之间的映射；
• 第 30-35 行：定义了表 [classes]，其基于 [sqlalchemy] 的类 [Table]。我们指定该表包含两列：
  • 列 [id] 作为主键，即类编号，第 33 行；
  • [nom] 列，包含类名，第 34 行；
• 第 31-32 行：请注意，x=y=z 这种语法在 Python 中是合法的：z 的值被赋给 y，然后 y 的值被赋给 x；
• 第 37-40 行：列出了表 [classes] 的各列与实体 [Classe] 的属性之间的映射关系；
• 第 42-57 行：实体 [Elève] 与表 [élèves] 之间的映射；
• 第 51-57 行：表 [élèves] 通过 [sqlalchemy] 的类 [Table] 进行定义。我们指定该表包含四个列：
  • 列 [id] 作为主键，表示学生编号，第 45 行；
  • [nom] 列，包含学生姓氏，第 46 行；
  • [prénom] 列，包含学生的名字，第 47 行。请注意，列名可以包含带重音的字符；
  • 第 49 行，列 [classe_id] 将包含该学生所属的班级编号。 这被称为外键。[élèves.classe_id] 是列 [classes.id] 上的外键（ForeignKey）。 这意味着 [élèves.classe_id] 的值必须存在于 [classes.id] 列中；
• 第 51-57 行：列出了表 [élèves] 的各列与实体 [Elève] 的属性之间的对应关系：
  • 第 53-55 行很容易理解；
  • 第 56 行较为复杂：它定义了属性 [Elève.classe] 的值，该值由连接表 [élèves] 和 [classes] 的外键关系计算得出。 函数 [relationship] 的参数如下：
    • [Classe]：这是与实体 [Elève] 建立外键关系的实体的名称。 该关系应通过在表 [classes] 上设置外键，在表 [élèves] 中体现。我们知道该外键确实存在；
    • [backref="élèves"]：将添加到实体 [Classe] 上的属性名称。[Classe.élèves] 将包含该班级所有学生的列表。 该属性不应已存在。如果已存在，只需在此为 [backref] 选择另一个名称即可。 开发者无需管理该属性。[sqlalchemy]将负责此项工作。开发者只需知道该属性由[sqlalchemy]添加且已存在，并可在代码中使用它；
    • [lazy=’select’]：这意味着 ORM 不应试图立即为属性 [Elève.classe] 赋值。它只应在代码显式请求时才计算该值。因此：
    • 如果代码请求所有学生的列表，这些学生会被返回，但其 [classe] 属性不会被计算；
    • 稍后，代码关注特定学生 [e] 并引用其班级 [e.classe]。 该引用将强制 [sqlalchemy] 向数据库发起查询以获取该学生的班级信息，这一过程对开发者而言是透明的；
    • 同时，添加 [lazy=’select’] 旨在避免不必要的数据库查询；
• 第56行：当ORM从[élèves]表中检索一行数据时，它会获取[id, nom, prénom, classe_id]中的信息。基于这些信息，它需要构建一个Elève对象（包含id、姓、名、班级）。 对于 [id, nom, prénom] 属性，这并不困难。但对于 [classe] 属性，情况则更为复杂。其值是一个 [Classe] 类型的对象引用。 然而，ORM 仅包含 [élèves.classe_id] 这一条信息。 由于 [élèves.classe_id] 是 [classes.id] 列的外键，因此这里要求它利用该关系，从 [classes] 表中检索 id=[élèves.classe_id] 的行 （该行必然存在），并基于此行创建属性 [Elève.classe] 所期望的对象 [Classe]；
• 第59-71行：实体[Matière]与表[matières]之间的映射；
• 第 59-65 行：定义名为 [matières] 的表 [sqlalchemy]；
• 第 66-71 行：列出了表 [matières] 的列与实体 [Matière] 的属性之间的对应关系。此处没有难点；
• 第73-90行：实体[Note]与表[notes]之间的映射；
• 第 73-82 行：定义名为 [notes] 的表 [sqlalchemy]。该表有两个外键：
  • 第 79 行，[notes.élève_id] 列的值取自 [élèves.id] 列。该外键体现了某条评分属于特定学生的这一事实；
  • 第81行，[notes.matière_id]列的值来自[matières.id]列。该外键体现了某条成绩属于特定学科的事实；
• 第 84-90 行：实体 [Note] 与表 [notes] 之间的映射：
  • 第88行：属性[Note.élève]的值必须是类型为[Elève]的实例。 ORM 在 [notes] 表的该行中仅包含 [notes.élève_id] 列，该列引用了 [élèves.id] 列。 这里要求利用该外键关系，根据已知的成绩查找出 [Elève] 实例。 此外，[relationship(Elève, backref="notes", …)] 将创建新属性 [Elève.notes]，该属性将作为学生的成绩列表。该属性在类 [Elève] 中不应已存在；
  • 第89行：属性[Note.matière]的值必须为[Matière]类型的实例。 在表 [notes] 的该行中，ORM 仅包含 [notes.matière_id] 列的信息，该列引用了 [matières.id] 列。 这里说明，应利用该外键关系来查找具有该评分的 [Matière] 实例。 此外，[relationship(Matière, backref="notes", …)] 将创建新属性 [Matière.notes]，该属性将包含该课程的评分列表。该属性在类 [Matière] 中不应已存在；
• 第 92-96 行：为每个从 [BaseEntity] 派生的实体，定义了应从该实体的属性字典（BaseEntity.asdict）中排除的属性列表。 我们看到，[sqlalchemy] 会向所有映射实体添加属性 [_sa_instance_state]。我们不希望该属性出现在属性字典中。此外，我们还看到之前的映射已向实体添加了新属性：
  • [Elève.notes]：该学生的所有成绩；
  • [Classe.élèves]：班级内所有学生；
  • [Matière.notes]：该学科的所有成绩；

通常，我们不希望这些属性被添加到实体的状态中。事实上，计算这些属性的值会产生开销，而且这些值往往毫无用处。因此，如果我们检索名为“X”的学生：

• （续）
  • ORM 将返回一个 [Elève(id, nom, prénom, classe, notes)] 实体。由于 [lazy=’select’] 的存在，与数据库外键相关的 [classe, notes] 属性将不会被计算；
  • 现在，如果我显示该学生的字符串 jSON，我们知道这将是该实体字典 [asdict] 中的字符串 jSON。 如果其中包含属性 [classe] 和 [notes]，[sqlalchemy] 将不得不发起 SQL 查询来计算它们的值。这会消耗大量资源。 如果能避免这些查询，那是更好的；
  • 在此，我们已排除所有与外键相关的属性；
• 第 98-100 行：实例化并配置 [Session factory]（factory=工厂）。[Session] 对象用于创建基于事务的 [sqlalchemy] 会话；
• 第 102-103 行：创建一个 sqlalchemy 会话；
• 第 106 行：将 [sqlalchemy] 配置中的某些元素放入应用程序配置的全局字典中；
• 第 109 行：返回该配置字典；

文件 [config_layers] 配置应用程序的各层：

def configure(config: dict) -> dict:
    # 实例化 [dao] 层
    from DatabaseDao import DatabaseDao
    dao = DatabaseDao(config)

    # 实例化 [métier] 层
    from Métier import Métier
    métier = Métier(dao)

    # 实例化图层 [ui]
    from Console import Console
    ui = Console(métier)

    # 将图层放入配置
    config['dao'] = dao
    config['métier'] = métier
    config['ui'] = ui

    # 生成配置
    return config

• 第 1 行：函数 [configure] 接收应用程序全局配置字典；
• 第 2-12 行：实例化应用程序的层；
• 第 15-17 行：将各层的引用放入全局配置中；
• 第 20 行：返回新的配置；

19.6.5. [dao] 层 - 1

这里需要理解的是，[dao] [3]层与ORM [sqlalchemy] [4] 层进行通信，该层的配置如前一段所述。 在 |troiscouches v01| 应用程序的三个层 [ui, métier, dao] 中，仅需重写 [dao] 层。[ui, métier] 层予以保留。

[dao] 层的实现已放置在 [services] 文件夹中：

[InterfaceDatabaseDao] 是 [dao] 层的接口：

from abc import ABC, abstractmethod

from InterfaceDao import InterfaceDao


class InterfaceDatabaseDao(InterfaceDao, ABC):

    # 初始化数据库
    @abstractmethod
    def init_database(self, data: dict):
        pass

• 第 6 行：接口 [InterfaceDatabaseDao] 既继承自类 [ABC]（作为抽象类），又继承自项目 |troiscouches v01| 中的接口 [InterfaceDao]；
• 第8-11行：在[InterfaceDao]继承的方法中添加了[init_database]方法。 该方法的作用是使用第 10 行作为参数传递的 [data] 字典中的数据初始化数据库；

回顾一下，[InterfaceDao] 接口如下：

# 导入
from abc import ABC, abstractmethod

# DAO 接口
from Elève import Elève


class InterfaceDao(ABC):
    # 班级列表
    @abstractmethod
    def get_classes(self: object) -> list:
        pass

    # 学生列表
    @abstractmethod
    def get_élèves(self: object) -> list:
        pass

    # 科目列表
    @abstractmethod
    def get_matières(self: object) -> list:
        pass

    # 成绩列表
    @abstractmethod
    def get_notes(self: object) -> list:
        pass

    # 某学生成绩列表
    @abstractmethod
    def get_notes_for_élève_by_id(self: object, élève_id: int) -> list:
        pass

    # 按ID查询学生
    @abstractmethod
    def get_élève_by_id(self: object, élève_id: int) -> Elève:
        pass

[dao] 层的实现如下：

from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

from Classe import Classe
from Elève import Elève
from InterfaceDatabaseDao import InterfaceDatabaseDao
from Matière import Matière
from MyException import MyException
from Note import Note


class DatabaseDao(InterfaceDatabaseDao):

    def __init__(self, config: dict):
        # database = {"engine": engine, "metadata": metadata, "tables": tables, "session": session}
        self.database = config['database']
        self.session = self.database['session']

    def init_database(self, data: dict):
        …
…

• 第 11 行：类 [DatabaseDao] 实现了接口 [InterfaceDatabaseDao]；
• 第 13-16 行：类的构造函数。它接收应用程序配置字典作为参数；
• 第 15 行：保存配置 [sqlalchemy]；
• 第 16 行：保存 [sqlalchemy] 会话，将通过该会话操作数据库；
• 第18行：方法[init_database]使用字典[data]初始化数据库；

字典 [data] 由以下脚本 [data.py] 实现：

def configure():
    from Classe import Classe
    from Elève import Elève
    from Matière import Matière
    from Note import Note

    # 实例化类
    classe1 = Classe().fromdict({"id": 1, "nom": "classe1"})
    classe2 = Classe().fromdict({"id": 2, "nom": "classe2"})
    classes = [classe1, classe2]
    # 学科
    matière1 = Matière().fromdict({"id": 1, "nom": "matière1", "coefficient": 1})
    matière2 = Matière().fromdict({"id": 2, "nom": "matière2", "coefficient": 2})
    matières = [matière1, matière2]
    # 学生
    élève11 = Elève().fromdict({"id": 11, "nom": "nom1", "prénom": "prénom1", "classe": classe1})
    élève21 = Elève().fromdict({"id": 21, "nom": "nom2", "prénom": "prénom2", "classe": classe1})
    élève32 = Elève().fromdict({"id": 32, "nom": "nom3", "prénom": "prénom3", "classe": classe2})
    élève42 = Elève().fromdict({"id": 42, "nom": "nom4", "prénom": "prénom4", "classe": classe2})
    élèves = [élève11, élève21, élève32, élève42]
    # 学生在各科目的成绩
    note1 = Note().fromdict({"id": 1, "valeur": 10, "élève": élève11, "matière": matière1})
    note2 = Note().fromdict({"id": 2, "valeur": 12, "élève": élève21, "matière": matière1})
    note3 = Note().fromdict({"id": 3, "valeur": 14, "élève": élève32, "matière": matière1})
    note4 = Note().fromdict({"id": 4, "valeur": 16, "élève": élève42, "matière": matière1})
    note5 = Note().fromdict({"id": 5, "valeur": 6, "élève": élève11, "matière": matière2})
    note6 = Note().fromdict({"id": 6, "valeur": 8, "élève": élève21, "matière": matière2})
    note7 = Note().fromdict({"id": 7, "valeur": 10, "élève": élève32, "matière": matière2})
    note8 = Note().fromdict({"id": 8, "valeur": 12, "élève": élève42, "matière": matière2})
    notes = [note1, note2, note3, note4, note5, note6, note7, note8]
    # 将所有数据合并
    data = {"élèves": élèves, "classes": classes, "matières": matières, "notes": notes}
    # 生成数据
    return data

• 第 34 行：将传递给 [init_database] 方法的字典。该字典由以下键组成（第 32 行）：
  • [élèves]：学生列表；
  • [classes]：班级列表；
  • [matières]：科目列表；
  • [notes]：所有学生所有科目的成绩列表；

让我们回到方法 [init_database]：

def init_database(self, data: dict):
        # 数据库配置
        database = self.database
        engine = database['engine']
        metadata = database['metadata']
        tables = database['tables']

        try:
            # 删除现有表
            # checkfirst=True：首先检查表是否存在
            tables["notes"].drop(engine, checkfirst=True)
            tables["matières"].drop(engine, checkfirst=True)
            tables["élèves"].drop(engine, checkfirst=True)
            tables["classes"].drop(engine, checkfirst=True)

            # 根据映射重新创建表
            metadata.create_all(engine)

            # 填充表
            session = self.session

            # 类
            classes = data["classes"]
            for classe in classes:
                session.add(classe)

            # 学科
            matières = data["matières"]
            for matière in matières:
                session.add(matière)

            # 学生
            élèves = data["élèves"]
            for élève in élèves:
                session.add(élève)

            # 成绩
            notes = data["notes"]
            for note in notes:
                session.add(note)

            # 提交
            session.commit()
        except (DatabaseError, InterfaceError, IntegrityError) as erreur:
            # 会话取消
            if session:
                session.rollback()
            # 抛出异常
            raise MyException(23, f"{erreur}")

• 第 3-6 行：从数据库配置中获取信息；
• 第9-14行：我们看到配置[sqlalchemy]已将四个实体映射到四个表[élèves, matières, classes, notes]上。如果这些表存在，则首先将其删除；
• 第16-17行：重新创建刚才删除的四个表；
• 第22-25行：将所有类放入会话中；
• 第27-30行：将所有科目放入会话中；
• 第 32-35 行：将所有学生放入会话中；
• 第37-40行：将所有成绩放入该届；
• 在进行这些添加操作时，我们遵循了一定的顺序。我们首先处理那些与其他实体没有关联的实体，最后处理那些与其他实体有关联的实体。因此，当我们将学生添加到会话中时，他们所隶属的班级已经存在于会话中；
• 第43行：会话[sqlalchemy]已验证通过。完成此操作后，可确保会话中的所有数据均已与数据库同步。简而言之，这些数据已成功导入表中。 这得益于在 [sqlalchemy] 配置中设置的映射关系。[sqlalchemy] 知道每个实体应如何存储在表中。[sqlalchemy] 还生成了表中可能包含的外键；
• 第44-49行：若遇到问题，将取消[sqlalchemy]会话，并在第49行抛出异常；

19.6.6. 数据库初始化

脚本 [main_init_database] 使用脚本 [data.py] 的内容初始化数据库。其代码如下：

# 等待 mysql 或 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()

# 正在配置应用程序
import config
config = config.configure({'sgbd': sgbd})

# 已配置 syspath - 可以进行导入
from MyException import MyException

# 获取待导入数据库的数据
import data
data = data.configure()

# 正在检索数据层 [dao]
dao = config["dao"]

# ----------- 主
try:
    # 创建并初始化数据库表
    dao.init_database(data)
except MyException as ex:
    # 显示错误
    print(f"L'erreur suivante s'est produite : {ex}")
finally:
    # 释放应用程序占用的资源
    import shutdown
    shutdown.execute(config)
# 结束
print("Travail terminé...")

• 第 1-11 行：脚本等待参数 [mysql] 或 [pgres]，具体取决于是要初始化数据库 MySQL 还是 PostgreSQL；
• 第13-15行：应用程序已配置为使用作为参数传递的SGBD；
• 第 20-22 行：获取要写入数据库的数据；
• 第 25 行：[dao] 层已实例化，可在应用程序配置中访问；
• 第30行：初始化数据库；
• 第 34-37 行：无论是否发生错误，均使用模块 [shutdown] 释放应用程序资源；

[shutdown.py] 模块如下：

def execute(config: dict):
    # 释放应用程序占用的资源
    sqlalchemy_session = config['database']['session']
    if sqlalchemy_session:
        sqlalchemy_session.close()

函数 [shutdown.execute] 关闭用于初始化数据库的会话 [sqlalchemy]。

我们创建第一个执行配置（参见 |执行配置|）以执行 [main_init_database]，其中包含 SGBD 和 MySQL：

在 phpMyAdmin 中，此配置的执行结果如下：

对于 SGBD 和 [PostgreSQL]，我们使用以下执行配置：

执行后，[pgAdmin]中的结果如下：

值得注意的是，我们能够非常轻松地切换到 SGBD。

19.6.7. [dao] 层 – 2

我们回到实现 [dao] 层的 [DatabaseDao] 类。目前我们仅展示了 [init_database] 方法的实现。现在我们展示其他方法的实现：

from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

from Classe import Classe
from Elève import Elève
from InterfaceDatabaseDao import InterfaceDatabaseDao
from Matière import Matière
from MyException import MyException
from Note import Note


class DatabaseDao(InterfaceDatabaseDao):

    def __init__(self, config: dict):
        # database = {"engine": engine, "metadata": metadata, "tables": tables, "session": session}
        self.database = config['database']
        self.session = self.database['session']

    def init_database(self, data: dict):
        …

    # 所有类的列表
    def get_classes(self: object) -> list:
        # 查询
        return self.session.query(Classe).all()

    # 所有学生的列表
    def get_élèves(self: object) -> list:
        # 查询
        return self.session.query(Elève).all()

    # 所有科目的列表
    def get_matières(self: object) -> list:
        # 查询
        return self.session.query(Matière).all()

    # 所有学生的成绩单
    def get_notes(self: object) -> list:
        # 查询
        return self.session.query(Note).all()

    # 特定学生的成绩单
    def get_notes_for_élève_by_id(self: object, élève_id: int) -> list:
        # 查询学生——若不存在则抛出异常
        # 允许其返回
        élève = self.get_élève_by_id(élève_id)
        # 获取其成绩（延迟加载）
        notes = élève.notes
        # 返回一个字典
        return {"élève": élève, "notes": notes}

    # 通过编号定位学生
    def get_élève_by_id(self, élève_id: int) -> Elève:
        # 搜索该学生
        élèves = self.session.query(Elève).filter(Elève.id == élève_id).all()
        # 是否已找到？
        if élèves:
            return élèves[0]
        else:
            raise MyException(11, f"L'élève d'identifiant {élève_id} n'existe pas")

    # 通过姓名定位学生
    def get_élève_by_name(self, élève_name: str) -> Elève:
        # 正在寻找该学生
        élèves = self.session.query(Elève).filter(Elève.nom == élève_name).all()
        # 找到了吗？
        if élèves:
            return élèves[0]
        else:
            raise MyException(12, f"L'élève de nom {élève_name} n'existe pas")

    # 通过班级编号定位的班级
    def get_classe_by_id(self, classe_id: int) -> Classe:
        # 正在寻找该班级
        classes = self.session.query(Classe).filter(Classe.id == classe_id).all()
        # 找到了吗？
        if classes:
            return classes[0]
        else:
            raise MyException(13, f"La classe d'identifiant {classe_id} n'existe pas")

    # 通过名称定位的班级
    def get_classe_by_name(self, classe_name: str) -> Classe:
        # 正在寻找该类
        classes = self.session.query(Classe).filter(Classe.nom == classe_name).all()
        # 是否已找到？
        if classes:
            return classes[0]
        else:
            raise MyException(14, f"La classe de nom {classe_name} n'existe pas")

    # 通过编号定位的学科
    def get_matière_by_id(self, matière_id: int) -> Matière:
        # 正在查找该科目
        matières = self.session.query(Matière).filter(Matière.id == matière_id).all()
        # 找到了吗？
        if matières:
            return matières[0]
        else:
            raise MyException(11, f"La matière d'identifiant {matière_id} n'existe pas")

    # 通过名称定位的物质
    def get_matière_by_name(self, matière_name: str) -> Matière:
        # 正在寻找该物质
        matières = self.session.query(Matière).filter(Matière.nom == matière_name).all()
        # 是否已找到？
        if matières:
            return matières[0]
        else:
            raise MyException(15, f"La matière de nom {matière_name} n'existe pas")

• 第21-24行：方法[get_classes]需返回学校的班级列表。第20行，我们使用了一个之前出现过的查询；
• 第26-39行：另外三个类似的方法，用于获取学生、科目和成绩的列表；
• 第51-59行：方法[get_élève_by_id]应返回一个通过编号标识的学生。若该学生不存在，则抛出异常；
  • 第54行：使用带筛选条件的查询。返回的列表可能为空，也可能包含一个元素；
  • 第57行：若获取的列表不为空，则返回列表中的第一个元素；
  • 否则（第59行），将抛出异常；
• 第41-49行：方法[get_notes_for_élève_by_id]应返回由编号标识的学生的成绩：
  • 第 45 行，使用方法 [get_élève_by_id] 获取该学生的 Elève 实体；
  • 第 47 行，使用由实体 [Note] 与表 [notes] 之间的映射创建的属性 [Elève.notes]（参见 |SQLAlchemy 配置| 段落），该属性表示该学生的成绩；
  • 第 49 行：返回一个字典；
• 第 61-109 行：一系列类似的方法，用于：
  • 按姓名查找学生，第61-69行；
  • 查找班级，第71-89行；
  • 查找学科，第91-109行；

19.6.8. 脚本 [main_joined_queries]

脚本 [main_joined_queries] 之所以如此命名，是因为它旨在揭示 [sqlalchemy] 为检索属于多个表的信息而隐式发出的查询。 这些对程序员而言不可见的查询，会在实体的映射中将某个属性关联到函数 [relationship] 时每次触发。例如：

    # 映射
    mapper(Note, tables['notes'], properties={
        'id': notes_table.c.id,
        '值：notes_table.c.valeur,
        '学生': relationship(学生, backref="笔记", lazy="select"),
        '科目': relationship(科目, backref="成绩", lazy="select")
    })

上文展示了实体 [Note] 与表 [notes] 之间的映射：

• 第 5 行，当首次请求实体 [Note] 的属性 [élève] 时，系统将通过查询 SQL 在表 [élèves] 中查找该属性。 只要该属性尚未被请求，它就保持未定义状态（延迟加载）。一旦获取该属性，其值将保存在 ORM 的内存中。 当该属性被第二次引用时，ORM将立即返回其值，而无需再次发出SQL请求。这一过程对开发者而言是透明的；
• 反向属性 [Elève.notes]（backref）亦是如此，第 5 行；
• 属性 [Note.matière] 及其反向引用属性 [Matière.notes]（backref）亦是如此，见第 6 行；

脚本 [main_joined_queries] 如下：

# 需要一个 mysql 或 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()

# 正在配置应用程序
import config
config = config.configure({"sgbd": sgbd})

# 已配置 syspath - 可以进行导入
from MyException import MyException

# [dao] 层
dao = config["dao"]
try:
    # 按 ID 查询学生
    print("élève id=11 -----------")
    élève = dao.get_élève_by_id(11)
    print(f"élève={élève}")
    # 学生的班级（延迟加载）
    classe = élève.classe
    print(f"classe de l'élève : {classe}")
    # 同班同学（延迟加载）
    print("élèves dans la même classe :")
    for élève in classe.élèves:
        print(f"élève={élève}")

    # 按姓名查询某位学生
    print("élève nom='nom2' -----------")
    print(f"élève={dao.get_élève_by_name('nom2')}")
    # 该学生的班级（延迟加载）
    print(f"classe de l'élève : {élève.classe}")

    # 某学生的成绩
    print("notes de l'élève id=11 -----------")
    # 先显示学生
    élève = dao.get_élève_by_id(11)
    # 然后是成绩（延迟加载）
    for note in élève.notes:
        # 成绩
        print(f"note={note}, "
              # 成绩所属科目（延迟加载）
              f"matière={note.matière}")

    # 班级中的学生
    print("élèves de la classe nom='classe1' -----------")
    # 先显示班级
    classe = dao.get_classe_by_name('classe1')
    # 然后是学生（延迟加载）
    for élève in classe.élèves:
        print(élève)

    # [classe2] 也是如此
    print("élèves de la classe de nom 'classe2' -----------")
    classe = dao.get_classe_by_name('classe2')
    for élève in classe.élèves:
        print(élève)

    # 某门学科的成绩
    print("matière de nom='matière1' -----------")
    # 先显示科目
    matière = dao.get_matière_by_name('matière1')
    print(f"matière={matière}")
    # 然后是该科目的成绩（延迟加载）
    print("Notes dans la matière : ")
    for note in matière.notes:
        print(note)

    # 科目2也一样
    print("matière de nom='matière2' -----------")
    matière = dao.get_matière_by_name('matière2')
    print(f"matière={matière}")
    print("Notes dans la matière : ")
    for note in matière.notes:
        print(f"note={note}")
except MyException as ex1:
    # 显示错误
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # 显示错误
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # 释放资源
    import shutdown
    shutdown.execute(config)

注释足以说明代码。

为 MySQL 创建一个运行配置：

执行结果如下：

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/databases/sqlalchemy/05/main/main_joined_queries.py mysql
élève id=11 -----------
élève={"classe_id": 1, "nom": "nom1", "prénom": "prénom1", "id": 11}
classe de l'élève : {"nom": "classe1", "id": 1}
élèves dans la même classe :
élève={"classe_id": 1, "nom": "nom1", "prénom": "prénom1", "id": 11}
élève={"classe_id": 1, "nom": "nom2", "prénom": "prénom2", "id": 21}
élève nom='nom2' -----------
élève={"classe_id": 1, "nom": "nom2", "prénom": "prénom2", "id": 21}
classe de l'élève : {"nom": "classe1", "id": 1}
notes de l'élève id=11 -----------
note={"matière_id": 1, "valeur": 10.0, "élève_id": 11, "id": 1}, matière={"coefficient": 1.0, "nom": "matière1", "id": 1}
note={"matière_id": 2, "valeur": 6.0, "élève_id": 11, "id": 5}, matière={"coefficient": 2.0, "nom": "matière2", "id": 2}
élèves de la classe nom='classe1' -----------
{"classe_id": 1, "nom": "nom1", "prénom": "prénom1", "id": 11}
{"classe_id": 1, "nom": "nom2", "prénom": "prénom2", "id": 21}
élèves de la classe de nom 'classe2' -----------
{"classe_id": 2, "nom": "nom3", "prénom": "prénom3", "id": 32}
{"classe_id": 2, "nom": "nom4", "prénom": "prénom4", "id": 42}
matière de nom='matière1' -----------
matière={"coefficient": 1.0, "nom": "matière1", "id": 1}
Notes dans la matière : 
{"matière_id": 1, "valeur": 10.0, "élève_id": 11, "id": 1}
{"matière_id": 1, "valeur": 12.0, "élève_id": 21, "id": 2}
{"matière_id": 1, "valeur": 14.0, "élève_id": 32, "id": 3}
{"matière_id": 1, "valeur": 16.0, "élève_id": 42, "id": 4}
matière de nom='matière2' -----------
matière={"coefficient": 2.0, "nom": "matière2", "id": 2}
Notes dans la matière : 
note={"matière_id": 2, "valeur": 6.0, "élève_id": 11, "id": 5}
note={"matière_id": 2, "valeur": 8.0, "élève_id": 21, "id": 6}
note={"matière_id": 2, "valeur": 10.0, "élève_id": 32, "id": 7}
note={"matière_id": 2, "valeur": 12.0, "élève_id": 42, "id": 8}

Process finished with exit code 0

要理解这些结果，需注意我们已从实体字典中排除了某些属性（参见 |配置|）：

    # 实体配置 [BaseEntity]
    Elève.excluded_keys = ['_sa_instance_state', 'notes', 'classe']
    Classe.excluded_keys = ['_sa_instance_state', 'élèves']
    Matière.excluded_keys = ['_sa_instance_state', 'notes']
    Note.excluded_keys = ['_sa_instance_state', 'matière', 'élève']

因此，当在代码第26行编写[print(f"élève={élève}")]时，上文第1行表明['_sa_instance_state', 'notes', 'classe']的属性将不会显示。这在结果的第3行中可见。 所有其他属性均被显示。因此，仍以第3行为例，我们发现了一个新的属性[classe_id]，该属性最初并不存在于实体[Elève]中。 该属性直接对应于表 [élèves] 中的列 [classe_id]。 因此，[sqlalchemy] 向实体 [Elève] 添加了以下属性：[classe_id, _sa_instance_state, notes]。必须注意这一点，特别是因为这些属性在映射实体中不应已存在。

从实体字典中排除的属性非常重要。例如，如果未将属性 [notes, élève] 从实体 [Elève] 中排除，那么操作 [print(f"élève={élève}")] 将显示这些属性，并因此 如前所述，会触发隐式查询 SQL（延迟加载）以获取这些属性的值。 如果像这里一样显示的是学生列表，那么针对每位学生都会执行隐式的 SQL 操作。这一方面可能毫无必要，另一方面肯定会消耗大量执行时间。

若要使用 PostgreSQL 数据库执行脚本，需创建以下执行配置：

执行结果与使用 MySQL 时相同。

19.6.9. 脚本 [main_stats_for_élève]

脚本 [main_stats_for_élève] 即 [troiscouches v01] 应用程序中已使用的脚本。 当时其名称为 [main]。这是一个控制台应用程序，用于获取学生成绩的某些指标：[moyenne pondérée, min, max, liste]。它属于以下架构：

在此分层架构中，与应用程序 |troiscouches v01| 相比，仅 [dao] 层发生了变更。 由于新层 [dao] 遵循了旧层 [dao] 的接口 [InterfaceDao]，因此 [ui, métier] 层无需更改。 因此，我们可以继续使用 |troiscouches v01| 应用程序中定义的层。

脚本 [main_stats_for_élève] 通过以下方式实现了上图所示的 [main] 层：

# 等待 mysql 或 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()

# 正在配置应用程序
import config
config = config.configure({'sgbd': sgbd})

# 已配置 syspath - 可以进行导入
from MyException import MyException

# [ui] 层
ui = config["ui"]
try:
    # 正在执行 [ui] 层
    ui.run()
except MyException as ex1:
    # 显示错误
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # 显示错误
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # 释放资源
    import shutdown
    shutdown.execute(config)

• 第 20 行：从应用程序配置中获取 [ui] 层的引用；
• 第 24 行：使用 [ui] 层的唯一方法启动用户对话；

PostgreSQL 的运行配置如下：

以下是使用此配置的执行示例：

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/databases/sqlalchemy/05/main/main_stats_for_élève.py pgres
Numéro de l'élève (>=1 et * pour arrêter) : 11
Elève={"prénom": "prénom1", "id": 11, "classe_id": 1, "nom": "nom1"}, notes=[10.0 6.0], max=10.0, min=6.0, moyenne pondérée=7.33
Numéro de l'élève (>=1 et * pour arrêter) : 1
L'erreur suivante s'est produite : MyException[11, L'élève d'identifiant 1 n'existe pas]
Numéro de l'élève (>=1 et * pour arrêter) : *

Process finished with exit code 0