示例工程教程【免费下载链接】materialsBonus 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/activate2. 安装锁定依赖项目的 requirements.txt 已固定全部依赖版本直接安装即可(venv) $ python -m pip install -r requirements.txt其中与本文主题直接相关的核心依赖包括依赖锁定版本作用connexion2.14.1依据 OpenAPI 规范自动路由与校验请求Flask2.2.2Web 应用框架Flask-SQLAlchemy3.0.2ORM管理 SQLite 数据模型flask-marshmallow0.14.0对象序列化/反序列化marshmallow-sqlalchemy0.28.1生成基于 SQLAlchemy 模型的 SchemaSQLAlchemy1.4.42ORM 底层引擎PyYAML6.0解析swagger.ymlswagger-ui-bundle0.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 与 Marshmallowimport 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_dirbasedir) app connex_app.app app.config[SQLALCHEMY_DATABASE_URI] fsqlite:///{basedir / people.db} app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app) ma Marshmallow(app)关键点说明specification_dirbasedir指定 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.pyapp.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, peoplepeople) if __name__ __main__: app.run(host0.0.0.0, port8000, debugTrue)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_keyTrue) person_id db.Column(db.Integer, db.ForeignKey(person.id)) content db.Column(db.String, nullableFalse) timestamp db.Column( db.DateTime, defaultdatetime.utcnow, onupdatedatetime.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_keyTrue) lname db.Column(db.String(32), uniqueTrue) fname db.Column(db.String(32)) timestamp db.Column( db.DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow ) notes db.relationship( Note, backrefperson, cascadeall, delete, delete-orphan, single_parentTrue, order_bydesc(Note.timestamp), )db.relationship在 Python 对象层面建立一对多关系要点包括backrefperson让Note对象可以直接通过note.person反向访问所属人员cascadeall, delete, delete-orphan删除人员时级联删除其全部笔记同时断开关联的孤儿笔记也会被删除保证数据一致性order_bydesc(Note.timestamp)获取person.notes时默认按时间倒序排列。Schema 与序列化项目使用marshmallow-sqlalchemy的SQLAlchemyAutoSchema自动生成序列化 Schemaclass 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, manyTrue) note_schema NoteSchema() person_schema PersonSchema() people_schema PersonSchema(manyTrue)load_instance True反序列化时直接构造模型实例配合sessiondb.session可写入数据库include_fk True序列化Note时包含外键person_idinclude_relationships Truefields.Nested(NoteSchema, manyTrue)序列化Person时嵌套输出其笔记列表实现关联数据的完整呈现模块底部预创建了单条与多条manyTrue的 Schema 实例供处理器复用。API 处理器people.pypeople.py 实现Person的五个端点逻辑并与swagger.yml中的operationId一一对应operationId函数HTTP 方法与路径行为people.read_allread_all()GET/api/people返回全部人员及其笔记people.createcreate(person)POST/api/people新建人员重复姓氏返回 406people.read_oneread_one(lname)GET/api/people/{lname}按姓氏查询单个人不存在返回 404people.updateupdate(lname, person)PUT/api/people/{lname}更新人员名字不存在返回 404people.deletedelete(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, sessiondb.session) db.session.add(new_person) db.session.commit() return person_schema.dump(new_person), 201 else: abort(406, fPerson with last name {lname} already exists)由于lname在模型中声明为uniqueTrue业务层又通过查询预检重复双保险保证姓氏唯一冲突时返回 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, sessiondb.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, fPerson with last name {lname} not found)先按姓氏定位既有记录再将新数据与现有对象合并并提交更新成功返回201 Created。删除与查询delete直接调用db.session.delete(existing_person)借助模型中配置的cascadeall, delete, delete-orphan自动级联清理该人员的全部笔记read_all与read_one分别使用Person.query.all()与按lname过滤的one_or_none()并通过people_schema.dump()/person_schema.dump()输出嵌套笔记的 JSON。API 处理器notes.pynotes.py 是 Part 3 新增的笔记接口包含四个端点operationId函数HTTP 方法与路径行为notes.read_oneread_one(note_id)GET/api/notes/{note_id}按 ID 读取笔记notes.updateupdate(note_id, note)PUT/api/notes/{note_id}更新笔记内容notes.deletedelete(note_id)DELETE/api/notes/{note_id}删除笔记notes.createcreate(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, sessiondb.session) person.notes.append(new_note) db.session.commit() return note_schema.dump(new_note), 201 else: abort(404, fPerson not found for ID: {person_id})笔记必须挂载到已存在的人员上先校验person_id对应的人员再通过person.notes.append(new_note)建立关联后提交从源码可见关联关系的对象级写入方式。更新与删除update采用与人员相同的「定位 → 载入 → merge → commit」流程但只更新content字段delete按note_id删除单条记录。两者对不存在的 ID 均返回404并附带说明信息。OpenAPI 规范swagger.ymlswagger.yml 采用OpenAPI 3.0.0规范是整个 API 的契约文件Connexion 依据它生成路由、解析参数并注入 Swagger UI。全局信息与 serversopenapi: 3.0.0 info: title: RP Flask REST API description: An API about people and notes version: 1.0.0 servers: - url: /apiservers.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: integerPersonSchema 声明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 查看到预置的关联数据便于立即调试接口。验证 APISwagger UI 与命令行请求启动应用后可在 Swagger UIhttp://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-1Flask 基础骨架与启动方式flask-connexion-rest-part-2引入 Connexion 与 OpenAPI实现/people接口可对比其 models.py 与 swagger.yml 与 Part 3 的差异体会关联模型的演进flask-connexion-rest-part-3本文主题新增笔记关联模型与 CRUDflask-connexion-rest-part-4在 Part 3 基础上进一步引入静态资源与前端交互该版本使用server.py作为入口并新增static/目录。如需复现本项目请严格按照 requirements.txt 锁定版本安装依赖这是保证 Connexion 2.x 路由行为一致的前提。项目基于 MIT 许可证分发详见仓库根目录 LICENSE。完成本文的启动与验证流程后你就掌握了「OpenAPI 规范驱动路由 ORM 关联模型 Marshmallow 序列化」这一套可复用的 REST API 搭建范式可直接将其迁移到自己的业务场景中。赞分享示例工程教程【免费下载链接】materialsBonus materials, exercises, and example projects for our Python tutorials项目地址https://gitcode.com/gh_mirrors/ma/materials点击查看免费下载相关推荐基于 Flask、Connexion 与 SQLAlchemy 构建 REST APIPart 2rp_flask_api 项目实战解析基于 Flask、Connexion 与 SQLAlchemy 构建 REST APIPart 2rp_flask_api 项目实战解析 本篇技术指南围绕示例工程教程使用 Flask Connexion 构建基于 OpenAPI 的 REST APIflask-connexion-rest-part-1 实战指南使用 Flask Connexion 构建基于 OpenAPI 的 REST APIflask connexion rest part 1 实战指南 本指示例工程教程使用 Gel 与 Flask 构建 REST API从数据模型设计到 CRUD 端点完整实战使用 Gel 与 Flask 构建 REST API从数据模型设计到 CRUD 端点完整实战 本篇教程将演示如何使用 Gel 官方 Python 客户端把数据库图数据库关系型数据库上一篇如何高效使用智慧教育平台电子课本解析工具专业用户的完整指南下一篇终极JavaScript设计模式指南clean-code-javascript教你编写专业级代码创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考