Flask + Connexion + SQLAlchemy 实战（Part 3）：为 REST API 添加关联笔记模型与完整 CRUD-平芜编程栈

示例工程
教程

【免费下载链接】materials

Bonus materials, exercises, and example projects for our Python tutorials

项目地址：https://gitcode.com/gh_mirrors/ma/materials

本篇技术指南以开源仓库 materials 中flask-connexion-rest-part-3项目为蓝本，完整讲解「Python REST APIs With Flask, Connexion, and SQLAlchemy」系列第三部分的实战代码：如何基于 OpenAPI 3.0 规范（swagger.yml）由 Connexion 自动路由请求，通过 Flask-SQLAlchemy 建立Person（人员）与Note（笔记）的一对多关联模型，并提供人员与笔记两套完整的增删改查（CRUD）接口。读完本文，你将掌握从虚拟环境搭建、依赖安装、应用启动，到数据模型、API 处理器、Swagger UI 交互验证与演示数据库构建的完整落地流程。

项目定位与目录结构

Part 3 是系列教程中承上启下的关键一版：Part 1 建立了 Flask 基础骨架，Part 2 引入 Connexion 与 OpenAPI 规范实现/people接口，而Part 3 的核心增量是笔记（notes）——通过Note模型与Person建立一对多关联，让 API 从单表查询升级为关联数据操作。项目在仓库中的结构如下：

flask-connexion-rest-part-3/ ├── README.md # 项目说明与快速启动指南 ├── requirements.txt # 锁定版本的依赖清单 └── rp_flask_api/ ├── app.py # 应用入口：加载 API 规范并启动服务 ├── config.py # Connexion 应用、数据库与序列化配置 ├── models.py # Person / Note 数据模型与 Schema ├── people.py # /people 相关 API 处理器 ├── notes.py # /notes 相关 API 处理器 ├── build_database.py # 演示数据构建脚本（可选） ├── swagger.yml # OpenAPI 3.0 规范，定义全部路由 ├── people.db # SQLite 数据库文件 └── templates/ └── home.html # 首页模板，展示人员及其笔记

其中 README.md 给出了最简启动路径，而 rp_flask_api/ 目录内的源码则是本文逐层剖析的对象。

环境准备与快速启动

1. 创建虚拟环境

为保证依赖隔离，先创建并激活虚拟环境：

$ python -m venv venv $ source venv/bin/activate

2. 安装锁定依赖

项目的 requirements.txt 已固定全部依赖版本，直接安装即可：

(venv) $ python -m pip install -r requirements.txt

其中与本文主题直接相关的核心依赖包括：

依赖	锁定版本	作用
connexion	2.14.1	依据 OpenAPI 规范自动路由与校验请求
Flask	2.2.2	Web 应用框架
Flask-SQLAlchemy	3.0.2	ORM，管理 SQLite 数据模型
flask-marshmallow	0.14.0	对象序列化/反序列化
marshmallow-sqlalchemy	0.28.1	生成基于 SQLAlchemy 模型的 Schema
SQLAlchemy	1.4.42	ORM 底层引擎
PyYAML	6.0	解析`swagger.yml`
swagger-ui-bundle	0.0.9	内置 Swagger UI 交互文档页面

3. 启动应用

进入项目代码目录并启动开发服务器：

(venv) $ cd rp_flask_api (venv) $ python app.py

服务默认监听0.0.0.0:8000，启动后可访问两个关键入口：

首页：http://127.0.0.1:8000，展示数据库中的人员及各自的笔记列表；
Swagger UI 文档：http://127.0.0.1:8000/api/ui，可交互式调用全部 API。

应用入口与配置层：config.py

整个应用的初始化集中在 config.py，其核心逻辑是创建 Connexion 应用并串联 SQLAlchemy 与 Marshmallow：

import pathlib import connexion from flask_marshmallow import Marshmallow from flask_sqlalchemy import SQLAlchemy basedir = pathlib.Path(__file__).parent.resolve() connex_app = connexion.App(__name__, specification_dir=basedir) app = connex_app.app app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{basedir / 'people.db'}" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db = SQLAlchemy(app) ma = Marshmallow(app)

关键点说明：

specification_dir=basedir指定 OpenAPI 规范文件所在目录，因此后续app.add_api(config.basedir / "swagger.yml")可直接定位规范文件；
SQLALCHEMY_DATABASE_URI指向rp_flask_api/目录下的people.db，使用 SQLite 无需额外数据库服务；
SQLALCHEMY_TRACK_MODIFICATIONS = False关闭不必要的修改追踪，避免额外开销；
db与ma作为全局实例导出，供 models.py 中的模型与 Schema 复用。

入口文件 app.py

app.py 是启动入口，加载规范并定义首页路由：

import config from flask import render_template from models import Person app = config.connex_app app.add_api(config.basedir / "swagger.yml") @app.route("/") def home(): people = Person.query.all() return render_template("home.html", people=people) if __name__ == "__main__": app.run(host="0.0.0.0", port=8000, debug=True)

app.add_api(...)会解析swagger.yml中声明的全部路径，并根据operationId（如people.read_all）自动将 HTTP 请求路由到对应的处理函数——这是 Connexion 的核心机制：路由由规范声明，业务逻辑由模块实现，两者通过operationId字符串绑定。首页则直接查询Person并渲染模板。

数据模型：Person 与 Note 的一对多关联

Part 3 相对 Part 2 最重要的变化在 models.py：在Person基础上新增Note模型与关联关系。

Note 模型

class Note(db.Model): __tablename__ = "note" id = db.Column(db.Integer, primary_key=True) person_id = db.Column(db.Integer, db.ForeignKey("person.id")) content = db.Column(db.String, nullable=False) timestamp = db.Column( db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow )

person_id通过外键db.ForeignKey("person.id")指向person表，形成关联的物理基础；
timestamp在插入（default）与更新（onupdate）时都会自动写入当前 UTC 时间。

Person 模型与关系

class Person(db.Model): __tablename__ = "person" id = db.Column(db.Integer, primary_key=True) lname = db.Column(db.String(32), unique=True) fname = db.Column(db.String(32)) timestamp = db.Column( db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow ) notes = db.relationship( Note, backref="person", cascade="all, delete, delete-orphan", single_parent=True, order_by="desc(Note.timestamp)", )

db.relationship在 Python 对象层面建立一对多关系，要点包括：

backref="person"：让Note对象可以直接通过note.person反向访问所属人员；
cascade="all, delete, delete-orphan"：删除人员时级联删除其全部笔记，同时断开关联的孤儿笔记也会被删除，保证数据一致性；
order_by="desc(Note.timestamp)"：获取person.notes时默认按时间倒序排列。

Schema 与序列化

项目使用marshmallow-sqlalchemy的SQLAlchemyAutoSchema自动生成序列化 Schema：

class NoteSchema(ma.SQLAlchemyAutoSchema): class Meta: model = Note load_instance = True sqla_session = db.session include_fk = True class PersonSchema(ma.SQLAlchemyAutoSchema): class Meta: model = Person load_instance = True sqla_session = db.session include_relationships = True notes = fields.Nested(NoteSchema, many=True) note_schema = NoteSchema() person_schema = PersonSchema() people_schema = PersonSchema(many=True)

load_instance = True：反序列化时直接构造模型实例，配合session=db.session可写入数据库；
include_fk = True：序列化Note时包含外键person_id；
include_relationships = True+fields.Nested(NoteSchema, many=True)：序列化Person时嵌套输出其笔记列表，实现关联数据的完整呈现；
模块底部预创建了单条与多条（many=True）的 Schema 实例，供处理器复用。

API 处理器：people.py

people.py 实现Person的五个端点逻辑，并与swagger.yml中的operationId一一对应：

operationId	函数	HTTP 方法与路径	行为
`people.read_all`	`read_all()`	GET`/api/people`	返回全部人员及其笔记
`people.create`	`create(person)`	POST`/api/people`	新建人员，重复姓氏返回 406
`people.read_one`	`read_one(lname)`	GET`/api/people/{lname}`	按姓氏查询单个人，不存在返回 404
`people.update`	`update(lname, person)`	PUT`/api/people/{lname}`	更新人员名字，不存在返回 404
`people.delete`	`delete(lname)`	DELETE`/api/people/{lname}`	删除人员并级联删除笔记

创建：唯一性校验

def create(person): lname = person.get("lname") existing_person = Person.query.filter(Person.lname == lname).one_or_none() if existing_person is None: new_person = person_schema.load(person, session=db.session) db.session.add(new_person) db.session.commit() return person_schema.dump(new_person), 201 else: abort(406, f"Person with last name {lname} already exists")

由于lname在模型中声明为unique=True，业务层又通过查询预检重复，双保险保证姓氏唯一；冲突时返回 HTTP406 Not Acceptable。

更新：merge + commit 模式

def update(lname, person): existing_person = Person.query.filter(Person.lname == lname).one_or_none() if existing_person: update_person = person_schema.load(person, session=db.session) existing_person.fname = update_person.fname db.session.merge(existing_person) db.session.commit() return person_schema.dump(existing_person), 201 else: abort(404, f"Person with last name {lname} not found")

先按姓氏定位既有记录，再将新数据与现有对象合并并提交，更新成功返回201 Created。

删除与查询

delete直接调用db.session.delete(existing_person)，借助模型中配置的cascade="all, delete, delete-orphan"自动级联清理该人员的全部笔记；read_all与read_one分别使用Person.query.all()与按lname过滤的one_or_none()，并通过people_schema.dump()/person_schema.dump()输出嵌套笔记的 JSON。

API 处理器：notes.py

notes.py 是 Part 3 新增的笔记接口，包含四个端点：

operationId	函数	HTTP 方法与路径	行为
`notes.read_one`	`read_one(note_id)`	GET`/api/notes/{note_id}`	按 ID 读取笔记
`notes.update`	`update(note_id, note)`	PUT`/api/notes/{note_id}`	更新笔记内容
`notes.delete`	`delete(note_id)`	DELETE`/api/notes/{note_id}`	删除笔记
`notes.create`	`create(note)`	POST`/api/notes`	为指定人员创建笔记

创建笔记：先定位所属人员

def create(note): person_id = note.get("person_id") person = Person.query.get(person_id) if person: new_note = note_schema.load(note, session=db.session) person.notes.append(new_note) db.session.commit() return note_schema.dump(new_note), 201 else: abort(404, f"Person not found for ID: {person_id}")

笔记必须挂载到已存在的人员上：先校验person_id对应的人员，再通过person.notes.append(new_note)建立关联后提交，从源码可见关联关系的对象级写入方式。

更新与删除

update采用与人员相同的「定位 → 载入 → merge → commit」流程，但只更新content字段；delete按note_id删除单条记录。两者对不存在的 ID 均返回404并附带说明信息。

OpenAPI 规范：swagger.yml

swagger.yml 采用OpenAPI 3.0.0规范，是整个 API 的"契约"文件，Connexion 依据它生成路由、解析参数并注入 Swagger UI。

全局信息与 servers

openapi: 3.0.0 info: title: "RP Flask REST API" description: "An API about people and notes" version: "1.0.0" servers: - url: "/api"

servers.url声明/api，因此所有端点实际前缀为/api（如/api/people、/api/notes）。

可复用组件

components中定义了 Schema 与路径参数，供各端点以$ref引用：

components: schemas: Person: type: "object" required: - lname properties: fname: type: "string" lname: type: "string" parameters: lname: name: "lname" description: "Last name of the person to get" in: path required: True schema: type: "string" note_id: name: "note_id" description: "ID of the note" in: path required: true schema: type: "integer"

PersonSchema 声明lname为必填，Connexion 会自动校验请求体；
lname与note_id两个路径参数在parameters中统一定义，多个端点复用，避免重复。

端点声明与 operationId 绑定

以/people为例：

paths: /people: get: operationId: "people.read_all" tags: - "People" summary: "Read the list of people" responses: "200": description: "Successfully read people list" post: operationId: "people.create" ...

operationId: "people.read_all"即「模块名.函数名」，Connexion 据此将GET /api/people路由到 people.py 的read_all()。POST 请求体通过x-body-name: "person"指定处理器函数的参数名，实现请求 JSON 到函数参数的自动映射。

笔记端点声明

/notes的 POST 在规范中直接内联了请求体结构（person_id与content），而/notes/{note_id}的 GET/PUT/DELETE 复用note_id参数：

/notes: post: operationId: "notes.create" ... requestBody: content: application/json: schema: x-body-name: "note" type: "object" properties: person_id: type: "integer" content: type: "string"

可以看到note_id是整数类型、lname是字符串类型，路径参数的类型校验同样由 Connexion 依据规范自动完成。

首页模板：数据可视化

Part 3 的首页不再是空壳，templates/home.html 通过 Jinja2 遍历人员与其笔记：

{% for person in people %} <h2>{{ person.fname }} {{ person.lname }}</h2> <ul> {% for note in person.notes %} <li> {{ note.content }} </li> {% endfor %} </ul> {% endfor %}

这里直接消费了Person.notes的关联数据：控制器中Person.query.all()返回的人员对象，其notes属性经由 SQLAlchemy 懒加载填充，模板即可直接嵌套渲染——是关联模型在视图层的直观体现。

可选：构建演示数据库

项目提供了 build_database.py，一键生成带演示数据的people.db：

(venv) $ python build_database.py

脚本执行后会删除任何已存在的数据库并重建名为people.db的全新数据库。其实现要点：

在app.app_context()应用上下文中执行db.drop_all()与db.create_all()，按模型重建表结构；
内置PEOPLE_NOTES演示数据（Tooth Fairy、Knecht Ruprecht、Easter Bunny 三个人物及其多条笔记）；
对每个人员创建Person实例，通过new_person.notes.append(Note(...))挂载笔记，并使用datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S")解析固定格式时间戳；
最后统一db.session.commit()提交事务。

运行后即可通过首页或 API 查看到预置的关联数据，便于立即调试接口。

验证 API：Swagger UI 与命令行请求

启动应用后，可在 Swagger UI（http://127.0.0.1:8000/api/ui）中直接执行各端点，也可用curl验证。例如查询人员列表：

$ curl http://127.0.0.1:8000/api/people

创建笔记（需先存在对应人员）：

$ curl -X POST http://127.0.0.1:8000/api/notes \ -H "Content-Type: application/json" \ -d '{"person_id": 1, "content": "Remember to run build_database.py"}'

创建人员：

$ curl -X POST http://127.0.0.1:8000/api/people \ -H "Content-Type: application/json" \ -d '{"fname": "Example", "lname": "Smith"}'

可重点验证两类典型响应：创建成功返回201，查询不存在记录返回404，重复姓氏创建返回406；更新操作返回201，删除成功分别返回200（人员）与204（笔记）。

系列脉络与依赖清单

本仓库同时收录了该系列的其他阶段，可作为对照学习的资料：

flask-connexion-rest-part-1：Flask 基础骨架与启动方式；
flask-connexion-rest-part-2：引入 Connexion 与 OpenAPI，实现/people接口（可对比其 models.py 与 swagger.yml 与 Part 3 的差异，体会关联模型的演进）；
flask-connexion-rest-part-3：本文主题，新增笔记关联模型与 CRUD；
flask-connexion-rest-part-4：在 Part 3 基础上进一步引入静态资源与前端交互（该版本使用server.py作为入口，并新增static/目录）。

如需复现本项目，请严格按照 requirements.txt 锁定版本安装依赖，这是保证 Connexion 2.x 路由行为一致的前提。项目基于 MIT 许可证分发，详见仓库根目录 LICENSE。完成本文的启动与验证流程后，你就掌握了「OpenAPI 规范驱动路由 + ORM 关联模型 + Marshmallow 序列化」这一套可复用的 REST API 搭建范式，可直接将其迁移到自己的业务场景中。