Flask API开发如何快速上手？-杰瑞科技汇

Flask 简介：为什么选择 Flask？
环境准备：安装 Python、Flask 和虚拟环境。
第一个 "Hello World" API：创建最简单的 API 端点。
核心概念详解：
- 路由
- 请求
- 响应
- 视图函数
构建一个功能完整的 RESTful API：一个简单的待办事项列表 API。
- 设计 API 端点
- 使用 Flask-RESTful 扩展简化开发
- 数据存储（使用 Python 列表）
- 实现增删改查操作
数据验证：使用 Flask-Request 和 marshmallow。
数据库集成：使用 Flask-SQLAlchemy 与 SQLite。
进阶主题：
- 项目结构（蓝图 Blueprints）
- 部署（使用 Gunicorn）
总结与最佳实践

Flask 简介

Flask 是一个用 Python 编写的轻量级 Web 应用框架，它的核心非常简单，但通过强大的扩展生态系统，你可以轻松地为其添加各种复杂功能，如数据库集成、用户认证、表单处理等。

（图片来源网络，侵删）

选择 Flask 的理由：

轻量级：安装简单，核心代码少，学习曲线平缓。
灵活：你只使用你需要的部分，不会被框架束缚。
强大的扩展：社区活跃，有大量高质量的扩展（如 Flask-SQLAlchemy, Flask-Login, Flask-RESTful）。
文档完善：官方文档清晰易懂，是学习的最佳资源。

环境准备

在开始之前,请确保你已经安装了 Python (推荐 3.8+)。

1 虚拟环境

为了隔离项目依赖,强烈建议使用虚拟环境。

# 创建一个名为 'venv' 的虚拟环境
python -m venv venv
# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate

激活后,你的终端提示符前会出现 (venv)。

（图片来源网络，侵删）

2 安装 Flask

在激活的虚拟环境中,使用 pip 安装 Flask。

pip install Flask

第一个 "Hello World" API

让我们创建一个最简单的 API，它返回一个 JSON 响应。

创建一个名为 app.py 的文件。
写入以下代码：

# app.py
from flask import Flask, jsonify
# 创建一个 Flask 应用实例
app = Flask(__name__)
# 定义一个路由和对应的视图函数
# @app.route 是一个装饰器，它告诉 Flask 哪个 URL 应该触发我们的函数
@app.route('/hello', methods=['GET'])
def hello_world():
    # jsonify 是一个辅助函数，用于将 Python 字典转换为 JSON 响应
    # 并设置正确的 Content-Type 头部
    return jsonify({'message': 'Hello, World!'})
# 运行应用
if __name__ == '__main__':
    # debug=True 会在代码修改后自动重启服务器，并在出错时显示详细的调试信息
    app.run(debug=True)

运行 API

在终端中运行 app.py：

flask run

你会看到类似下面的输出：

（图片来源网络，侵删）

 * Serving Flask app 'app'
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment.
Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: ...

测试 API

打开你的浏览器或使用 API 客户端（如 Postman、Insomnia），访问 http://127.0.0.1:5000/hello。

你会得到如下的 JSON 响应：

{
  "message": "Hello, World!"
}

核心概念详解

1 路由

路由负责将 URL 映射到视图函数。

@app.route('/user/<username>')
def show_user_profile(username):
    # 显示用户名的页面
    # <username> 是一个动态部分，它会作为参数传递给函数
    return f'User: {username}'
@app.route('/post/<int:post_id>')
def show_post(post_id):
    # 显示 id 为 post_id 的帖子，id 是整数
    return f'Post: {post_id}'

2 请求

当客户端发送请求时,Flask 会创建一个全局的 request 对象，你可以从中获取所有请求数据。

from flask import request
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 获取 POST 请求中的表单数据
        username = request.form['username']
        password = request.form['password']
        return f'Processing login for {username}'
    else:
        # 这是一个 GET 请求，可能返回一个登录页面
        return 'This is a login page.'

3 响应

视图函数的返回值会自动转换为一个 HTTP 响应对象，最常用的方式是 jsonify。

from flask import jsonify, make_response
@app.route('/custom_response')
def custom_response():
    response_data = {'status': 'success', 'data': [1, 2, 3]}
    # 使用 jsonify 创建一个带有正确 JSON 头部的响应
    return jsonify(response_data)
@app.route('/manual_response')
def manual_response():
    # 你也可以手动创建响应对象
    response = make_response('This is a manual response.', 201)
    response.headers['X-Custom-Header'] = 'MyValue'
    return response

构建一个功能完整的 RESTful API

我们将创建一个简单的待办事项 API，支持对任务的增删改查操作。

1 设计 API 端点

方法	URL	描述
GET	`/tasks`	获取所有任务列表
GET	`/tasks/<int:id>`	获取单个任务详情
POST	`/tasks`	创建一个新任务
PUT	`/tasks/<int:id>`	更新一个现有任务
DELETE	`/tasks/<int:id>`	删除一个任务

2 使用 `Flask-RESTful` 扩展

为了更轻松地构建 RESTful API，我们使用 Flask-RESTful 扩展。

pip install flask-restful

3 完整代码 (`app.py`)

我们将使用一个 Python 列表作为临时的内存数据库。

# app.py
from flask import Flask
from flask_restful import Api, Resource, reqparse, abort
# 1. 创建 Flask 应用和 API 对象
app = Flask(__name__)
api = Api(app)
# 2. 模拟数据库
tasks = {}
task_id_counter = 1
# 3. 请求解析器
# 用于验证和解析请求数据
task_post_args = reqparse.RequestParser()
task_post_args.add_argument("content", type=str, help="Content of the task is required", required=True)
task_put_args = reqparse.RequestParser()
task_put_args.add_argument("content", type=str)
task_put_args.add_argument("completed", type=bool)
# 4. 定义资源
class Task(Resource):
    def get(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task not found")
        return tasks[task_id]
    def put(self, task_id):
        args = task_put_args.parse_args()
        if task_id not in tasks:
            abort(404, message="Task not found, cannot update")
        if args['content']:
            tasks[task_id]['content'] = args['content']
        if args['completed'] is not None:
            tasks[task_id]['completed'] = args['completed']
        return tasks[task_id]
    def delete(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task not found")
        del tasks[task_id]
        return '', 204 # No Content
class TaskList(Resource):
    def get(self):
        return tasks
    def post(self):
        global task_id_counter
        args = task_post_args.parse_args()
        task = {
            "id": task_id_counter,
            "content": args["content"],
            "completed": False
        }
        tasks[task_id_counter] = task
        task_id_counter += 1
        return task, 201 # Created
# 5. 将资源添加到 API
# <int:task_id> 表示 URL 中的 task_id 部分会被转换成整数
api.add_resource(TaskList, '/tasks')
api.add_resource(Task, '/tasks/<int:task_id>')
# 6. 运行应用
if __name__ == '__main__':
    app.run(debug=True)

4 测试 API

使用 curl 或 Postman 等工具来测试你的 API。

获取所有任务 (GET /tasks)

curl http://127.0.0.1:5000/tasks
# 返回: {} (空)

创建一个新任务 (POST /tasks)

curl -X POST -H "Content-Type: application/json" -d '{"content": "Learn Flask"}' http://127.0.0.1:5000/tasks
# 返回:
{
  "completed": false,
  "content": "Learn Flask",
  "id": 1
}

获取单个任务 (GET /tasks/1)

curl http://127.0.0.1:5000/tasks/1
# 返回:
{
  "completed": false,
  "content": "Learn Flask",
  "id": 1
}

更新任务 (PUT /tasks/1)

curl -X PUT -H "Content-Type: application/json" -d '{"completed": true}' http://127.0.0.1:5000/tasks/1
# 返回:
{
  "completed": true,
  "content": "Learn Flask",
  "id": 1
}

删除任务 (DELETE /tasks/1)

curl -X DELETE http://127.0.0.1:5000/tasks/1
# 返回: (空响应，状态码 204)

数据验证

在上面的例子中,reqparse 已经做了基本的验证，对于更复杂的验证，推荐使用 marshmallow 库。

pip install marshmallow

marshmallow 允许你定义数据模式，用于序列化（对象 -> JSON）和反序列化（JSON -> 对象）以及验证。

数据库集成

使用内存列表存储数据在应用重启后会丢失,对于真实项目，你需要一个数据库。Flask-SQLAlchemy 是一个流行的 ORM（对象关系映射）扩展，可以让你用 Python 类来操作数据库。

pip install Flask-SQLAlchemy

1 修改 `app.py` 以使用 SQLAlchemy

# app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource, reqparse, abort
app = Flask(__name__)
api = Api(app)
# --- 数据库配置 ---
# 使用 SQLite 数据库，文件名为 app.db
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
# --- 数据库模型 ---
class TaskModel(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(200), nullable=False)
    completed = db.Column(db.Boolean, default=False)
    def __repr__(self):
        return f"Task(id={self.id}, content='{self.content}', completed={self.completed})"
# --- 请求解析器 (保持不变) ---
task_post_args = reqparse.RequestParser()
task_post_args.add_argument("content", type=str, help="Content of the task is required", required=True)
# --- 资源类 (修改为使用数据库) ---
class Task(Resource):
    def get(self, task_id):
        task = TaskModel.query.get(task_id)
        if not task:
            abort(404, message="Task not found")
        return task.to_dict()
    def put(self, task_id):
        args = task_put_args.parse_args()
        task = TaskModel.query.get(task_id)
        if not task:
            abort(404, message="Task not found, cannot update")
        if args['content']:
            task.content = args['content']
        if args['completed'] is not None:
            task.completed = args['completed']
        db.session.commit()
        return task.to_dict()
    def delete(self, task_id):
        task = TaskModel.query.get(task_id)
        if not task:
            abort(404, message="Task not found")
        db.session.delete(task)
        db.session.commit()
        return '', 204
class TaskList(Resource):
    def get(self):
        tasks = TaskModel.query.all()
        return [task.to_dict() for task in tasks]
    def post(self):
        args = task_post_args.parse_args()
        new_task = TaskModel(content=args['content'])
        db.session.add(new_task)
        db.session.commit()
        return new_task.to_dict(), 201
# --- 添加资源到 API (保持不变) ---
api.add_resource(TaskList, '/tasks')
api.add_resource(Task, '/tasks/<int:task_id>')
# --- 初始化数据库 ---
@app.before_first_request
def create_tables():
    db.create_all()
if __name__ == '__main__':
    app.run(debug=True)

注意：to_dict() 方法是 db.Model 的一个辅助方法，可以方便地将模型实例转换为字典。

首次运行：你需要先创建数据库表，可以在 Python 交互式环境中运行：

from app import app, db
with app.app_context():
    db.create_all()

或者,在 app.py 中添加 @app.before_first_request 装饰器（如上所示），它会自动在第一个请求前创建表。

进阶主题

1 项目结构（蓝图 Blueprints）

当应用变大时,将所有代码放在 app.py 中会变得混乱，Flask 的蓝图功能允许你将应用拆分成多个模块。

/my_api
  /app
    /__init__.py      # 创建 Flask 应用实例
    /api
      /__init__.py    # 创建 API 实例
      /tasks.py       # 任务相关的路由和资源
    /models.py        # 数据库模型
  /venv
  /app.py            # 启动文件

2 部署（使用 Gunicorn）

Flask 自带的服务器仅用于开发，在生产环境中，你需要一个专业的 WSGI 服务器，如 Gunicorn。

安装 Gunicorn:
```
pip install gunicorn
```

运行应用:

# 格式: gunicorn -w [worker数] -b [地址:端口] [模块名:应用实例名]
gunicorn -w 4 -b 127.0.0.1:8000 app:app

这会启动 4 个 worker 进程，监听在 8000 端口上。

反向代理：Gunicorn 不应该直接暴露在公网上，你应该在它前面放一个 Nginx 或 Apache 作为反向代理，处理静态文件和 SSL 终止。

总结与最佳实践

使用虚拟环境：始终保持项目依赖隔离。
保持简单：Flask 的优势在于其简洁性，不要过度设计。
使用扩展：优先使用成熟的社区扩展，而不是自己造轮子。
编写清晰的代码：遵循 PEP 8 风格指南，使用有意义的变量名。
重视测试：学习使用 pytest 为你的 API 编写单元测试和集成测试。
安全性：在生产环境中，务必关闭 debug 模式，并考虑使用 Flask 的扩展来处理 CSRF、XSS 等安全问题。

这份指南为你提供了使用 Flask 开发 API 的坚实基础，从简单的端点开始，逐步构建更复杂的功能，并探索 Flask 丰富的扩展生态，你将能够构建出强大而可维护的 Web API。

Flask API开发如何快速上手？

目录

Flask 简介

环境准备

1 虚拟环境

2 安装 Flask

第一个 "Hello World" API

运行 API

测试 API

核心概念详解

1 路由

2 请求

3 响应

构建一个功能完整的 RESTful API

1 设计 API 端点

2 使用 Flask-RESTful 扩展

3 完整代码 (app.py)

4 测试 API

数据验证

数据库集成

1 修改 app.py 以使用 SQLAlchemy

进阶主题

1 项目结构（蓝图 Blueprints）

2 部署（使用 Gunicorn）

总结与最佳实践

相关推荐

Java Socket编程实例具体怎么实现？

2 使用 `Flask-RESTful` 扩展

3 完整代码 (`app.py`)

1 修改 `app.py` 以使用 SQLAlchemy