Flask框架RESTful API实战指南:从零构建可扩展接口

首先,我们需要准备开发环境——安装Flask和Flask-RESTful扩展。打开终端,运行以下命令:

pip install flask flask-restful

这两个库是实现RESTful API的核心:Flask负责搭建Web应用,Flask-RESTful提供了Resource类等RESTful开发工具,帮我们快速映射HTTP方法与资源操作。

Flask框架RESTful API实战指南:从零构建可扩展接口

理解核心逻辑: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"}
  1. 获取单个用户

    curl http://localhost:5000/users/1
    
  2. 更新用户

    curl -X PUT -H "Content-Type: application/json" -d '{"age":26}' http://localhost:5000/users/1
    
  3. 删除用户

    curl -X DELETE http://localhost:5000/users/1
    

进阶:添加JWT认证(保护敏感接口)

实际项目中,API需要身份验证。我们用flask-jwt-extended实现JWT(JSON Web Token)认证:

  1. 安装依赖:

    pip install flask-jwt-extended
    
  2. 初始化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)
    
  3. 添加登录接口(获取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')
    
  4. 保护敏感接口:
    在需要认证的方法上添加@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()没问题,但生产环境需要更稳定的部署方案:

  1. 用Gunicorn作为WSGI服务器

    pip install gunicorn
    gunicorn -w 4 -b 0.0.0.0:5000 app:app  # 4个工作进程,绑定5000端口
    
  2. 用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;
        }
    }
    

常见问题排查

  1. 接口返回400错误:检查请求参数是否符合reqparse的规则(比如类型错误、必填项缺失)。
  2. 接口返回401错误:JWT Token无效或过期,重新请求/login获取新Token。
  3. 接口返回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

(0)

相关推荐