首先,我们需要准备开发环境——安装Flask和Flask-RESTful扩展。打开终端,运行以下命令:
pip install flask flask-restful
这两个库是实现RESTful API的核心:Flask负责搭建Web应用,Flask-RESTful提供了Resource
类等RESTful开发工具,帮我们快速映射HTTP方法与资源操作。

理解核心逻辑:Resource类与HTTP方法映射
Flask-RESTful的核心设计是将资源(比如用户、商品)封装为Resource
子类,再将HTTP方法(GET/POST/PUT/DELETE)对应到类中的同名方法。比如一个“用户”资源的操作对应关系如下:
HTTP方法 | 资源操作 | 示例URL | 描述 |
---|---|---|---|
GET | 获取所有用户 | /users | 返回用户列表 |
GET | 获取单个用户 | /users/ |
根据ID返回单个用户 |
POST | 创建用户 | /users | 新增一个用户 |
PUT | 更新用户 | /users/ |
根据ID修改用户信息 |
DELETE | 删除用户 | /users/ |
根据ID删除用户 |
记住这个对应关系,接下来的代码会围绕它展开。
第一步:初始化Flask应用与API对象
创建app.py
文件,写入基础代码:
from flask import Flask
from flask_restful import Api
# 初始化Flask应用
app = Flask(__name__)
# 初始化API对象(绑定Flask应用与RESTful功能)
api = Api(app)
第二步:定义资源类(实现CRUD)
我们以“用户资源(User)”为例,实现增删改查(CRUD)。为了简化,先用字典模拟数据库(实际项目请用SQLAlchemy连接MySQL/PostgreSQL):
from datetime import datetime
from flask_restful import Resource, reqparse, abort
from flask import jsonify
# 模拟数据库(键:用户ID,值:用户信息)
users = {}
next_user_id = 1 # 自增ID计数器
class UserResource(Resource):
# 1. 获取用户(支持单个/全部)
def get(self, user_id=None):
if user_id:
# 查找单个用户
user = users.get(user_id)
if not user:
abort(404, message=f"用户{user_id}不存在") # 返回404错误
return jsonify(user)
else:
# 返回所有用户列表
return jsonify(list(users.values()))
# 2. 创建用户(POST请求)
def post(self):
# 定义请求参数规则(验证必填项)
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, required=True, help="姓名不能为空")
parser.add_argument('email', type=str, required=True, help="邮箱不能为空")
parser.add_argument('age', type=int, default=18, help="年龄默认18")
args = parser.parse_args() # 解析并验证参数
# 生成新用户
global next_user_id
new_user = {
"id": next_user_id,
"name": args['name'],
"email": args['email'],
"age": args['age'],
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
users[next_user_id] = new_user
next_user_id += 1
return jsonify(new_user), 201 # 201表示资源创建成功
# 3. 更新用户(PUT请求)
def put(self, user_id):
# 检查用户是否存在
if user_id not in users:
abort(404, message=f"用户{user_id}不存在")
# 解析更新参数(非必填,只修改提供的字段)
parser = reqparse.RequestParser()
parser.add_argument('name', type=str)
parser.add_argument('email', type=str)
parser.add_argument('age', type=int)
args = parser.parse_args()
# 更新用户信息
if args['name']:
users[user_id]['name'] = args['name']
if args['email']:
users[user_id]['email'] = args['email']
if args['age'] is not None:
users[user_id]['age'] = args['age']
return jsonify(users[user_id])
# 4. 删除用户(DELETE请求)
def delete(self, user_id):
if user_id not in users:
abort(404, message=f"用户{user_id}不存在")
del users[user_id]
return '', 204 # 204表示无内容(删除成功)
第三步:绑定资源路由
将资源类与URL路径关联,让Flask知道如何分发请求:
# 添加用户资源路由(支持带ID和不带ID的URL)
api.add_resource(UserResource, '/users', '/users/<int:user_id>')
第四步:测试接口(用curl或Postman)
启动Flask应用:
if __name__ == '__main__':
app.run(debug=True) # debug模式自动重启
用curl测试几个关键接口:
1. 创建用户:
curl -X POST -H "Content-Type: application/json" -d '{"name":"张三","email":"zhangsan@example.com","age":25}' http://localhost:5000/users
响应(状态码201):
{"id":1,"name":"张三","email":"zhangsan@example.com","age":25,"created_at":"2025-08-24 14:30:00"}
-
获取单个用户:
curl http://localhost:5000/users/1
-
更新用户:
curl -X PUT -H "Content-Type: application/json" -d '{"age":26}' http://localhost:5000/users/1
-
删除用户:
curl -X DELETE http://localhost:5000/users/1
进阶:添加JWT认证(保护敏感接口)
实际项目中,API需要身份验证。我们用flask-jwt-extended
实现JWT(JSON Web Token)认证:
-
安装依赖:
pip install flask-jwt-extended
-
初始化JWT配置:
from flask_jwt_extended import JWTManager, create_access_token, jwt_required # JWT配置(生产环境请用环境变量存储密钥) app.config['JWT_SECRET_KEY'] = 'your-strong-secret-key' # 密钥要复杂 app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1) # Token有效期1小时 jwt = JWTManager(app)
-
添加登录接口(获取Token):
class LoginResource(Resource): def post(self): parser = reqparse.RequestParser() parser.add_argument('username', type=str, required=True) parser.add_argument('password', type=str, required=True) args = parser.parse_args() # 模拟验证(实际查数据库) if args['username'] == 'admin' and args['password'] == 'admin123': # 创建Token(包含用户身份信息) access_token = create_access_token(identity=args['username']) return jsonify(access_token=access_token) else: abort(401, message="用户名或密码错误") # 绑定登录路由 api.add_resource(LoginResource, '/login')
-
保护敏感接口:
在需要认证的方法上添加@jwt_required()
装饰器,比如“获取用户列表”:class UserResource(Resource): @jwt_required() # 需要Token才能访问 def get(self, user_id=None): # 原逻辑不变
测试认证流程:
– 先请求/login
获取Token:
curl -X POST -H "Content-Type: application/json" -d '{"username":"admin","password":"admin123"}' http://localhost:5000/login
– 用Token访问保护的接口:
curl -H "Authorization: Bearer your-token-here" http://localhost:5000/users
生产环境部署技巧
开发环境用app.run()
没问题,但生产环境需要更稳定的部署方案:
-
用Gunicorn作为WSGI服务器:
pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 app:app # 4个工作进程,绑定5000端口
-
用Nginx做反向代理:
配置Nginx转发请求到Gunicorn,同时处理静态资源和HTTPS:server { listen 80; server_name your-domain.com; location / { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }
常见问题排查
- 接口返回400错误:检查请求参数是否符合
reqparse
的规则(比如类型错误、必填项缺失)。 - 接口返回401错误:JWT Token无效或过期,重新请求
/login
获取新Token。 - 接口返回404错误:URL路径错误,比如
/user/1
应为/users/1
(注意复数)。
最佳实践总结
- 用SQLAlchemy连接数据库:代替字典模拟,支持事务和复杂查询。
- 统一响应格式:比如所有接口返回
{"code":0,"data":...,"message":"成功"}
,方便前端处理。 - 添加日志:用Flask的
app.logger
记录请求日志,比如:from flask import request @app.before_request def log_request(): app.logger.info(f"Request: {request.method} {request.url} - {request.remote_addr}")
- 使用Marshmallow验证数据:比
reqparse
更灵活,支持嵌套数据验证。
原创文章,作者:,如若转载,请注明出处:https://zube.cn/archives/187