Lab 8：数据库应用开发与事务管理实战

课程：数据库技术与应用：AI时代的基石数据技能
授课教师：王健楠 (计算机科学与技术系)
学期：2026年（春季）

实验目的与学习目标

本实验分为两大部分，完成后你应能够：

一、数据库应用开发

• 掌握基本的 Python 数据库编程：能够使用 mssql-python 驱动连接 Azure SQL 数据库，通过 Python 代码建表并批量导入数据。
• 理解数据库应用的典型架构：能够使用 Flask 框架将数据库操作封装为 Web API，供外部程序通过 HTTP 接口调用。

二、事务管理

• 掌握事务的核心概念：理解 ACID 特性，能够在 Python 中正确使用 COMMIT 和 ROLLBACK，保证多步数据库操作的原子性。
• 解决并发访问中的数据一致性问题：能够识别多用户同时操作数据库时产生的竞争条件，并通过事务加以防范。

实验背景

假设你正在为一个在线影片借阅平台开发数据库管理系统。客户可以通过订阅不同套餐方案在线借阅电影，系统需要追踪哪些客户当前正在借阅哪些影片，并通过网络接口提供查询与借阅服务。你的工作分为两个阶段：

• 第一阶段（任务 1–2）：在 Azure 云端 SQL 数据库中建表，并将套餐方案、客户信息、影片目录和借阅记录批量导入；
• 第二阶段（任务 3–5）：开发 Web API，实现查询和借阅功能，并确保系统在多客户并发使用时的事务一致性。

---

关键业务约束： 由于版权方对每部影片的并发授权数量有严格限制，系统有两条必须严格遵守的规则，你需要通过数据库事务来保证它们在任何情况下都不被违反：

• 约束 C1：由于版权授权限制，任何时刻一部影片最多只能被一位客户同时借阅。
• 约束 C2：任何时刻，一位客户同时借阅的影片数量不能超过其套餐上限。

---

评分规则： 共 100 分（5 个任务）。

任务	内容	分值
任务 1	Azure SQL 环境搭建	10 分
任务 2	数据加载（4 个子任务）	40 分
任务 3	查询当前借阅者	15 分
任务 4	查询剩余可借阅数量	15 分
任务 5	借阅电影	20 分

作答要求：

• 本实验需要你完成三个 Python 文件：connect_db.py、loaddata.py 和 query.py，本 Notebook 仅作为实验说明文档，不需要在 Notebook 中填写代码。
• 请仔细阅读每个任务的说明，在对应的 .py 文件中找到需要填写的函数，完成代码后按照说明中提供的测试方式进行验证。

---

第 0 步：文件结构与环境准备

本实验的工作目录包含以下文件：

Lab8/
├── Lab8.ipynb          ← 本文件（实验说明文档）
├── connect_db.py        ← 需要修改：数据库连接配置
├── loaddata.py          ← 需要修改：数据加载函数
├── query.py             ← 需要修改：Web API 实现
├── RentalPlan.txt       ← 数据文件：套餐方案
├── Customer.txt         ← 数据文件：客户信息
├── Movie.txt            ← 数据文件：影片目录
└── Rental.txt           ← 数据文件：借阅记录

在开始实验之前，请先安装所需的 Python 依赖：

pip install mssql-python flask

---

第一部分：数据库编程

---

任务 1：Azure SQL 环境搭建（10 分）

任务背景

在开始编写业务代码之前，你首先需要搭建云端数据库。本实验使用 Microsoft Azure SQL——一个全托管的关系型数据库云服务，无需自行维护服务器。

请按照以下步骤完成环境搭建：

步骤 1：登录 Azure 学生账号，创建一个空的 SQL 数据库，数据库名称设置为 VideoStore。

步骤 2：为了让实验评测系统能够验证你的结果，请前往数据库的防火墙设置页面，添加一条允许所有 IP 访问的规则（如下图所示）。

步骤 3：安装 Python 数据库驱动：

pip install mssql-python

步骤 4：打开同目录下的 connect_db.py 文件，将 MSSQL_CONNECTION_STRING 替换为你的 Azure SQL 连接字符串。

提示： 连接字符串可以在 Azure 门户中，进入你的 SQL 数据库页面，在左侧菜单「Connection strings」中找到。

提交内容：完成配置后，提交 connect_db.py 文件。

---

任务 2：数据加载（40 分）

任务背景

实验提供了四份初始数据文件，分别包含套餐方案、客户信息、影片目录和历史借阅记录。你需要将这些数据从本地文件导入到 Azure SQL 数据库中。

首先，你需要在 VideoStore 数据库中设计并创建以下四张表：

• RentalPlan（套餐方案）：pid INT，pname VARCHAR(50)，monthly_fee FLOAT，max_movies INT
• Customer（客户信息）：cid INT，pid^{FK→RentalPlan} INT，username VARCHAR(50)，password VARCHAR(50)
• Movie（影片目录）：mid INT，mname VARCHAR(50)，year INT
• Rental（借阅记录）：cid^{FK→Customer} INT，mid^FK→Movie INT，date_and_time DATETIME，status VARCHAR(6)

数据库设计说明：

• 所有外键声明请使用 ON DELETE CASCADE 选项。
• Rental 表不需要设置主键。
• RentalPlan 表：套餐共有四种——"Basic"（最多 1 部）、"Rental Plus"（最多 3 部）、"Super Access"（最多 5 部）、"Ultra Access"（最多 10 部）。
• Rental 表：status 字段取值为 "open"（正在借阅）或 "closed"（已归还）。客户借出影片时插入一条 "open" 记录，归还时更新为 "closed"（不删除）。date_and_time 字段记录借出时间，用于区分同一客户对同一影片的多次借阅。

请打开 loaddata.py 文件，完成以下四个函数。

2.a 完成 LoadRentalPlan 函数（10 分）

def LoadRentalPlan(filename, conn):
    """
    参数：
        filename: "RentalPlan.txt"
        conn: 调用 connect_db() 获取的数据库连接对象
    功能：
        1. 在 VideoStore 数据库中创建名为 "RentalPlan" 的表；
        2. 读取 RentalPlan.txt 中的数据并插入表中。
           - 各列之间以 '|' 分隔；
           - 请使用 executemany() 进行批量插入。
    """
    # 在此填写你的代码

2.b 完成 LoadCustomer 函数（10 分）

def LoadCustomer(filename, conn):
    """
    参数：
        filename: "Customer.txt"
        conn: 调用 connect_db() 获取的数据库连接对象
    功能：
        1. 在 VideoStore 数据库中创建名为 "Customer" 的表；
        2. 读取 Customer.txt 中的数据并插入表中。
           - 各列之间以 '|' 分隔；
           - 请使用 executemany() 进行批量插入。
    """
    # 在此填写你的代码

2.c 完成 LoadMovie 函数（10 分）

def LoadMovie(filename, conn):
    """
    参数：
        filename: "Movie.txt"
        conn: 调用 connect_db() 获取的数据库连接对象
    功能：
        1. 在 VideoStore 数据库中创建名为 "Movie" 的表；
        2. 读取 Movie.txt 中的数据并插入表中。
           - 各列之间以 '|' 分隔；
           - 请使用 executemany() 进行批量插入。
    """
    # 在此填写你的代码

2.d 完成 LoadRental 函数（10 分）

def LoadRental(filename, conn):
    """
    参数：
        filename: "Rental.txt"
        conn: 调用 connect_db() 获取的数据库连接对象
    功能：
        1. 在 VideoStore 数据库中创建名为 "Rental" 的表；
        2. 读取 Rental.txt 中的数据并插入表中。
           - 各列之间以 '|' 分隔；
           - 请使用 executemany() 进行批量插入。
    """
    # 在此填写你的代码

验证方式：

在终端中运行以下命令：

python loaddata.py

若四张表均成功创建并导入数据，你将在 Azure 门户中看到如下图所示的结果：

提交内容：完成后，提交 loaddata.py 文件。

---

第二部分：事务管理

任务背景

数据库已经就绪，接下来需要开发 Web API。就像豆瓣 API、微博开放平台一样，你需要通过 HTTP 接口将数据库的查询和更新功能暴露出来，供前端页面调用。

你将使用 Flask 框架来实现这些 Web API。Flask 是一个轻量级的 Python Web 框架，代码简洁，在业界被广泛用于快速开发 API 服务。

特别提醒： 在实现借阅功能（任务 5）时，你必须使用 SQL 事务来保证数据一致性——想象一下，如果两位客户同时抢借同一部影片会发生什么？正确使用事务是本次实验最核心的技能点。

快速上手：Flask 入门示例

前提： 请确保已完成任务 1 和任务 2（数据库已创建并导入数据），再进行以下操作。

步骤 1：阅读 Flask 官方文档：最小应用，了解 Flask 的基本用法。

步骤 2：在本实验目录下打开终端，运行：

FLASK_APP=query.py flask run

在浏览器中访问以下地址，确认能看到响应 {"cid": 1}：

http://127.0.0.1:5000/login?username=mike&password=mike123

步骤 3：阅读 query.py 中已有的 login 函数代码，理解为什么上面的请求会返回 {"cid": 1}。只要弄懂这一个例子，后续三个任务的实现模式完全相同，无需学习 Flask 的其他功能。

开发建议： 调试阶段推荐开启 Flask 的调试模式——代码改动后服务会自动重启：

FLASK_APP=query.py FLASK_DEBUG=1 flask run

---

任务 3：查询当前借阅者（15 分）

任务背景

前端页面需要显示某部影片当前的借阅状态。你需要提供一个 API 接口：给定一个影片 ID（mid），返回当前正在借阅该影片的客户 ID（cid）。如果该影片目前没有被任何人借阅，返回 cid = -1。

请在 query.py 中完成 getRenterID 函数：

@app.route('/getRenterID')
def getRenterID():
    """
    接收参数：mid（影片 ID）
    返回：cid（正在借阅该影片的客户 ID）
         如果该影片当前没有被任何人借阅，返回 cid = -1
    """
    mid = int(request.args.get('mid', -1))

    # 在此填写你的代码

    response = {'cid': cid}
    return response

测试（在初始数据集上）：

请求 URL	预期响应
http://127.0.0.1:5000/getRenterID?mid=1	{'cid': 2}
http://127.0.0.1:5000/getRenterID?mid=4	{'cid': -1}

---

任务 4：查询剩余可借阅数量（15 分）

任务背景

前端页面需要告知客户：你还可以再借几部影片？ 每位客户的借阅配额取决于其订阅的套餐方案（max_movies），减去当前正处于 "open" 状态的借阅数量。

与任务 3 不同，这个查询可能需要联合多张表，你可以用多条 SQL 语句分步实现，也可以用一条 JOIN 查询完成。在 Azure SQL Server 中，默认情况下每条语句自动提交（autocommit）。如果需要将多条语句放在同一个事务中执行，需要先关闭自动提交模式：conn.autocommit = False。

请在 query.py 中完成 getRemainingRentals 函数：

@app.route('/getRemainingRentals')
def getRemainingRentals():
    """
    接收参数：cid（客户 ID）
    返回：n（该客户还可以借阅的影片数量）
         n = 0 表示已达到套餐上限，无法再借。
    """
    cid = int(request.args.get('cid', -1))
    # 告知数据库驱动：开启多语句事务模式
    conn.autocommit = False

    # 在此填写你的代码

    conn.autocommit = True
    response = {"remain": n}
    return jsonify(response)

测试（在初始数据集上）：

请求 URL	预期响应
http://127.0.0.1:5000/getRemainingRentals?cid=1	{"remain": 5}
http://127.0.0.1:5000/getRemainingRentals?cid=2	{"remain": 2}
http://127.0.0.1:5000/getRemainingRentals?cid=3	{"remain": 9}
http://127.0.0.1:5000/getRemainingRentals?cid=4	{"remain": 1}

---

任务 5：借阅电影（20 分）

任务背景

这是本次实验最核心、也最具挑战性的任务。当一位客户发起借阅请求时，系统需要：

1. 检查该影片是否已被他人借阅（约束 C1）；
2. 检查该客户是否已达到套餐借阅上限（约束 C2）；
3. 如果两个约束都满足，在 Rental 表中插入一条 status = "open" 的新记录；
4. 否则，拒绝此次借阅请求。

为什么必须使用事务？ 以上步骤必须在一个原子事务中完成。试想：两位客户同时发起请求，争抢同一部影片。如果没有事务保护，两个请求可能都通过了 C1 的检查，然后同时完成插入——这样一部影片就同时被两人持有，严重违反业务规则。

当需要拒绝请求时，建议使用 conn.rollback() 回滚事务。

请在 query.py 中完成 rent 函数：

@app.route('/rent')
def rent():
    """
    接收参数：cid（客户 ID）、mid（影片 ID）
    返回："success" 或 "fail"

    以下情况返回 "fail"：
        - 违反 C1：该影片已被他人借阅；
        - 违反 C2：该客户已达到套餐借阅上限；
        - 或同时违反两者。
    否则更新数据库并返回 "success"。
    """
    cid = int(request.args.get('cid', -1))
    mid = int(request.args.get('mid', -1))

    conn = get_db()

    # 开启多语句事务模式
    conn.autocommit = False

    # 在此填写你的代码

    conn.autocommit = True

    # response = {"rent": "success"} 或 response = {"rent": "fail"}
    return jsonify(response)

测试（在初始数据集上，请严格按顺序依次执行）：

请求 URL	预期响应
http://127.0.0.1:5000/rent?cid=4&mid=5	{"rent": "success"}
http://127.0.0.1:5000/rent?cid=4&mid=6	{"rent": "fail"}
http://127.0.0.1:5000/rent?cid=1&mid=3	{"rent": "success"}
http://127.0.0.1:5000/rent?cid=1&mid=2	{"rent": "fail"}

注意： 每条请求都会改变数据库状态，后续结果依赖于前序操作，请勿打乱顺序。

思考题（无需提交）： 如果两位客户同时发送 rent?cid=4&mid=5 请求，你的实现能保证只有一人成功吗？试着分析你的代码是否真正解决了并发问题。

---

提交说明

请将以下文件打包后上传至网络学堂：

• connect_db.py
• loaddata.py
• query.py
• 四个数据文件：Customer.txt、Movie.txt、Rental.txt、RentalPlan.txt

---

实验圆满结束！

恭喜你完成了数据库应用开发与事务管理的全部实战任务！

通过本次实验，你亲身经历了从零搭建一个真实数据库应用的全过程——从云端数据库配置、数据批量导入，到 Web API 开发和并发事务控制。这些技能在实际的软件工程工作中会反复用到，希望这次动手实践能让你对课堂上的事务理论有更深刻的理解。

注意： 提交作业后，请不要删除或关闭你的 Azure SQL 数据库，评分期间我们需要访问你的数据库进行验证。待本次作业评分完成后，再关闭数据库资源。