使用 Flask + Connexion 构建基于 OpenAPI 的 REST API：flask-connexion-rest-part-1 实战指南

本指南以仓库中 flask-connexion-rest-part-1/README.md 为核心脉络，围绕 "Python REST APIs With Flask, Connexion, and SQLAlchemy" 系列的第一部分展开。你将学会用 Connexion 将 Flask 与 OpenAPI（Swagger）规范无缝结合：编写 swagger.yml 描述接口、通过 operationId 自动绑定 Python 处理函数、以内存数据实现完整的 RESTful CRUD，并在浏览器中直接使用 Swagger UI 调试接口。读完本文，你将掌握一套"规范驱动（Specification-First）"的 REST API 开发工作流，为后续接入 SQLAlchemy 持久层打好基础。

项目概览：一个"规范驱动"的最小完整示例

flask-connexion-rest-part-1/ 目录包含两个可直接运行的工程，分别对应教程的两层目标：

flask-connexion-rest-part-1/
├── README.md            # 官方使用说明（本篇文章的依据）
├── requirements.txt     # 锁定版本的依赖清单
├── flask_starter/       # 纯 Flask 起步模板（未引入 Connexion）
│   ├── app.py
│   └── templates/home.html
└── rp_flask_api/        # 核心示例：Flask + Connexion REST API
    ├── app.py
    ├── people.py
    ├── swagger.yml
    └── templates/home.html

其中 rp_flask_api/ 是本部分的重点：它只用一个 OpenAPI 文件（swagger.yml）和一个业务处理模块（people.py），就搭建出一个包含 GET /api/people、POST /api/people、GET /api/people/{lname}、PUT /api/people/{lname}、DELETE /api/people/{lname} 五个端点的完整 REST 服务。数据暂存于进程内存中的 PEOPLE 字典，正好适合作为 REST 概念与 Connexion 机制的演示载体。

环境准备与快速启动

按照 README 的指引，完成以下三步即可把服务跑起来。

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 路由的核心库
Flask	2.2.2	底层 Web 框架
swagger-ui-bundle	0.0.9	为 API 提供浏览器端 Swagger UI 文档页
PyYAML	6.0	解析 swagger.yml
Werkzeug	2.2.2	Flask 依赖的 WSGI 工具库
requests	2.28.1	可用于后续接口自测

3. 启动开发服务器

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

启动成功后：

• 首页（"Hello, World!" 页面）：访问 http://127.0.0.1:8000
• Swagger UI 接口文档：访问 http://127.0.0.1:8000/api/ui

Swagger UI 是本方案最直观的亮点：它会自动读取 swagger.yml 生成可交互的文档页，你可以在页面上直接对每个端点发起真实请求，无需额外编写测试脚本即可验证接口行为。

Flask Starter：不带 Connexion 的纯 Flask 起步模板

flask_starter/ 是教程为读者准备的"零依赖 Connexion"起点，用来先建立纯 Flask 的基础认知。其核心代码位于 flask_starter/app.py：

from flask import Flask, render_template

app = Flask(__name__)


@app.route("/")
def home():
    return render_template("home.html")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

使用方式也很简单：复制整个 flask_starter/ 文件夹并改名为你的项目名，然后在项目目录内创建虚拟环境并安装 Flask：

$ python -m venv venv
$ source venv/bin/activate
(venv) $ python -m pip install flask
(venv) $ python app.py

访问 http://127.0.0.1:8000 即可看到由 templates/home.html 渲染的 "Hello, World!" 页面。这个模板展示了 Flask 最朴素的形态——路由、视图函数、模板渲染——当你把它与下文 rp_flask_api/app.py 对比时，就能清晰看到 Connexion 带来的架构变化：路由定义不再散落在视图函数装饰器里，而是集中迁移到了 swagger.yml 中。

核心入口：Connexion 如何把 OpenAPI 变成 Flask 路由

rp_flask_api 中的应用入口只有十几行代码（见 rp_flask_api/app.py），却同时承载了页面路由与 API 路由：

import connexion
from flask import render_template

app = connexion.App(__name__, specification_dir="./")
app.add_api("swagger.yml")


@app.route("/")
def home():
    return render_template("home.html")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

逐行解读这段代码，可以拆出 Connexion 的三个关键机制：

1. connexion.App(__name__, specification_dir="./")：创建一个 Connexion 应用实例。__name__ 用于定位模板目录等资源；specification_dir="./" 告诉 Connexion 在应用所在目录下查找规范文件（即当前目录下的 swagger.yml）。从实现细节看，connexion.App 内部包装了一个 Flask 应用，因此它同时拥有 Flask 的全部能力——这正是下文 @app.route("/") 可以继续使用的根本原因。

2. app.add_api("swagger.yml")：加载并解析 OpenAPI 文件，把其中的每个 path 条目编译为真实的 Flask 路由，路由前缀由规范中的 servers.url（/api）决定。这一步是"规范驱动"开发模式的核心：API 的形状以规范文件为准，业务实现只负责被 operationId 点名。

3. app.run(host="0.0.0.0", port=8000, debug=True)：启动开发服务器，监听所有网卡地址、8000 端口并开启调试模式。

从源码结构可以推断，一旦 add_api("swagger.yml") 执行完毕，所有 API 路由就绪，而 home() 这样的普通 Flask 路由则与 API 路由并存于同一应用中——这正是教程第一部分"先跑通 REST + 页面共存"的设计意图。

OpenAPI 规范详解：swagger.yml 是 API 的"唯一事实来源"

swagger.yml 是整个项目的核心配置，它用 YAML 描述了 API 的版本、服务器前缀、数据模型、参数与全部端点。逐段拆解如下。

顶层信息与服务器前缀

openapi: 3.0.0
info:
  title: "RP Flask REST API"
  description: "An API about people and notes"
  version: "1.0.0"

servers:
  - url: "/api"

openapi: 3.0.0 声明遵循 OpenAPI 3.0 规范；info 提供 API 元信息（会显示在 Swagger UI 顶部）；servers.url: "/api" 为所有端点加上 /api 前缀，因此规范中定义的路径 GET /people 实际对外是 GET /api/people。

组件：数据模型与路径参数

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"

• Person 是 API 的请求/响应模型：一个对象，包含字符串字段 fname 与 lname，其中 lname 必填。
• lname 是可复用的路径参数定义，声明它出现在 URL 路径（in: path）中且必须提供。后续多个端点通过 $ref 引用它，避免重复定义。

路径与 operationId 的绑定关系

规范共定义了五个端点，每个端点都通过 operationId 指向 people.py 中对应的函数（operationId 的格式为"模块名.函数名"）：

方法	路径	operationId	绑定的处理函数
GET	/people	people.read_all	read_all()
POST	/people	people.create	create(person)
GET	/people/{lname}	people.read_one	read_one(lname)
PUT	/people/{lname}	people.update	update(lname, person)
DELETE	/people/{lname}	people.delete	delete(lname)

以列表接口与新增接口为例，规范中的定义如下：

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"
      tags:
        - People
      summary: "Create a person"
      requestBody:
          description: "Person to create"
          required: True
          content:
            application/json:
              schema:
                x-body-name: "person"
                $ref: "#/components/schemas/Person"
      responses:
        "201":
          description: "Successfully created person"

这里有两个值得注意的细节：

• operationId 是连接规范与代码的唯一桥梁。Connexion 根据 people.create 自动导入 people 模块并调用其 create 函数，开发者不需要手写任何路由注册代码。
• x-body-name: "person" 决定了函数形参名。x-body-name 是 Connexion 的扩展字段，它告诉 Connexion"把请求体作为名为 person 的参数传入"，因此 create(person) 与 update(lname, person) 的第二个参数名必须与之对应，否则无法正确接收请求体。

单条资源端点 GET /people/{lname} 则演示了路径参数与规范的结合：

  /people/{lname}:
    get:
      operationId: "people.read_one"
      tags:
        - People
      summary: "Read one person"
      parameters:
        - $ref: "#/components/parameters/lname"
      responses:
        "200":
          description: "Successfully read person"

Connexion 会从 URL 中提取 lname 并作为关键字参数传给 read_one(lname)。

业务处理层：people.py 中的内存版 CRUD

people.py 实现了全部业务逻辑。数据以模块级字典 PEOPLE 保存，键是姓氏（lname），值是一条包含 fname、lname 和 timestamp 的记录：

from datetime import datetime

from flask import abort, make_response


def get_timestamp():
    return datetime.now().strftime(("%Y-%m-%d %H:%M:%S"))


PEOPLE = {
    "Fairy": {
        "fname": "Tooth",
        "lname": "Fairy",
        "timestamp": get_timestamp(),
    },
    "Ruprecht": {
        "fname": "Knecht",
        "lname": "Ruprecht",
        "timestamp": get_timestamp(),
    },
    "Bunny": {
        "fname": "Easter",
        "lname": "Bunny",
        "timestamp": get_timestamp(),
    },
}

初始数据包含三个"人物"（Tooth Fairy、Knecht Ruprecht、Easter Bunny），每条记录带一个写入时生成的时间戳。

查询列表与单条记录

def read_all():
    return list(PEOPLE.values())

read_all() 直接返回字典的值列表，Connexion 会将其序列化为 JSON 数组返回；GET /api/people 即返回三个初始对象。

def read_one(lname):
    if lname in PEOPLE:
        return PEOPLE[lname]
    else:
        abort(404, f"Person with last name {lname} not found")

read_one(lname) 在命中时返回记录，未命中时调用 Flask 的 abort(404, ...)，返回 404 状态码和描述信息。

新增记录

def create(person):
    lname = person.get("lname")
    fname = person.get("fname", "")

    if lname and lname not in PEOPLE:
        PEOPLE[lname] = {
            "lname": lname,
            "fname": fname,
            "timestamp": get_timestamp(),
        }
        return PEOPLE[lname], 201
    else:
        abort(406, f"Person with last name {lname} already exists")

create(person) 展示了 REST 语义与 HTTP 状态码的配合：

• 校验逻辑：请求体必须包含 lname（与 Person 模型的 required: [lname] 呼应），且该姓氏不能已存在；
• 成功时返回新记录并以元组 (数据, 201) 的形式明确指定状态码为 201 Created；
• 失败（姓氏缺失或已存在）时返回 406 Not Acceptable。

这里注意一个小细节：person.get("fname", "") 说明 fname 是可选字段，缺省值为空字符串——这与规范中 Person 只强制要求 lname 完全一致。

更新与删除

def update(lname, person):
    if lname in PEOPLE:
        PEOPLE[lname]["fname"] = person.get("fname", PEOPLE[lname]["fname"])
        PEOPLE[lname]["timestamp"] = get_timestamp()
        return PEOPLE[lname]
    else:
        abort(404, f"Person with last name {lname} not found")


def delete(lname):
    if lname in PEOPLE:
        del PEOPLE[lname]
        return make_response(f"{lname} successfully deleted", 200)
    else:
        abort(404, f"Person with last name {lname} not found")

update 采用"部分更新"策略：只覆盖请求体中提供的 fname，未提供时保留原值，并刷新 timestamp。delete 从字典中移除记录，成功时用 make_response(消息, 200) 返回一条文本消息；二者对不存在的 lname 均返回 404。

一个值得注意的细节是：规范中 DELETE /people/{lname} 声明的成功响应码是 204（见 swagger.yml），而实现中实际返回的是 200 加文本消息——这提示了 Connexion 下"规范描述"与"函数返回值"之间存在一定自由度，实践时建议让二者保持一致，避免客户端与文档产生歧义。

用 curl 验证五个端点

启动服务后，可以按下面顺序用 curl 完整走一遍 CRUD 流程（注意 URL 前缀 /api）：

1. 读取列表

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

应返回包含 Fairy、Ruprecht、Bunny 三条记录的 JSON 数组。

2. 新增记录

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

成功返回 201 与新建记录；若重复提交同一个 lname，则返回 406 及 "Person with last name Claus already exists" 错误。

3. 读取单条记录

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

命中返回该记录，不存在的姓氏（如 curl http://127.0.0.1:8000/api/people/Missing）返回 404。

4. 更新记录

$ curl -X PUT http://127.0.0.1:8000/api/people/Claus \
       -H "Content-Type: application/json" \
       -d '{"fname": "Santa Claus"}'

更新 fname 并刷新 timestamp。

5. 删除记录

$ curl -X DELETE http://127.0.0.1:8000/api/people/Claus

返回 "Claus successfully deleted" 文本消息。

也可以直接在 http://127.0.0.1:8000/api/ui 的 Swagger UI 页面中依次展开各端点、填入参数并点击 "Execute" 发起相同请求，图形化地验证上述行为。

从 Part 1 到完整系列：内存数据只是起点

按照 README 的标题定位，本仓库是 "Python REST APIs With Flask, Connexion, and SQLAlchemy" 系列的第一部分。从当前代码结构可以推断，Part 1 刻意选择了内存字典 PEOPLE 作为数据存储，目的是把注意力集中在"Connexion + OpenAPI"这一层机制上：规范如何描述接口、operationId 如何绑定函数、状态码如何表达 REST 语义。系列后续部分将会引入 SQLAlchemy 将数据落到数据库，届时 PEOPLE 字典会被 ORM 模型与数据库会话取代，而本文所讲的 swagger.yml 规范与 operationId 绑定机制将保持不变——这正是"规范驱动"开发模式的最大收益：接口契约与业务实现解耦，无论底层存储如何演进，API 层都稳定可复用。

许可与致谢

本仓库基于 MIT 许可证分发，详见仓库根目录的 LICENSE；README 作者为 Philipp Acsany（Real Python 团队）。你可以自由使用、修改与再分发本示例代码，用于学习或作为自己 Flask REST 项目的起点。