Python 从入门到实战（十七）：Flask API 开发与多端集成（让学生系统支持 Web APP 小程序多端访问）

@[toc]
欢迎回到「Python 从入门到实战」系列专栏。上一篇咱们完成了学生成绩管理系统的优化与加固，系统已经能稳定支撑 Web 端的日常教学使用。但随着场景扩展，你可能会遇到新需求：比如想开发一个手机 APP 让学生随时查成绩，或者对接学校的教务系统自动同步数据，甚至想做一个小程序方便家长查看孩子的成绩 —— 这些需求都需要API 接口来实现。

今天咱们聚焦「Flask API 开发与多端集成」，从 API 基础概念入手，实战开发一套符合 RESTful 规范的 API 接口，包括用户认证、学生成绩 CRUD（增删改查）、成绩变动通知等核心功能。同时，我们会集成 JWT 认证保障 API 安全，生成自动 API 文档方便调用，并演示如何通过 APP、小程序、Python 脚本调用这些 API，让学生系统真正具备 “多端访问” 能力。所有开发都基于前面的系统，复用已有模型和权限逻辑，确保技术栈的连贯性。

一、为什么需要 API？从 “单端” 到 “多端” 的升级

在讲具体开发前，先明确 API 的价值 —— 为什么要在 Web 系统之外再做 API？

1. 现有 Web 系统的局限

前面的系统是 “Web 端独占” 模式，存在以下局限：

• 访问场景单一：只能通过浏览器访问，手机上体验差，无法对接 APP、小程序；
• 系统耦合紧密：Web 界面和业务逻辑绑定，想对接其他系统（如学校教务系统）只能手动导入导出数据；
• 扩展性差：无法实现 “成绩变动自动通知”“批量数据同步” 等自动化场景。

2. API 能解决什么问题？

API（应用程序编程接口）是不同系统之间的 “通信桥梁”，就像两个系统之间的 “普通话”。基于 API，我们的学生系统可以：

• 支持多端访问：Web 端、手机 APP、小程序、桌面软件共用一套业务逻辑，数据实时同步；
• 对接第三方系统：自动从教务系统同步学生信息，或把成绩同步到家长群、邮件；
• 实现自动化：定时任务调用 API 统计成绩，成绩变动时通过 API 触发通知（邮件 / 短信）。

3. 我们要开发的 API 范围

基于学生成绩系统，重点开发以下三类 API，覆盖核心业务：

二、API 基础：RESTful 规范与核心概念

开发 API 前，需要先掌握一套通用的规范 ——RESTful 规范，它能让 API 接口更易读、易维护，不同开发者调用时不用反复沟通格式。

1. 什么是 RESTful 规范？

RESTful 是一种 API 设计风格，核心是 “用 HTTP 协议的特性来表达业务逻辑”，比如：

• 用HTTP 方法表示操作类型（GET 查、POST 增、PUT 改、DELETE 删）；
• 用URL 路径表示操作的资源（如/api/students表示学生资源）；
• 用HTTP 状态码表示请求结果（200 成功、201 创建成功、401 未认证、403 无权限）；
• 用JSON 格式传输数据（前后端 / 多端通用的数据格式）。

2. 核心概念速查表

3. 技术选型：Flask-RESTX

为了快速开发符合 RESTful 规范的 API，并自动生成 API 文档，我们选择Flask-RESTX（Flask-RESTful 的升级版），它的核心优势：

• 支持命名空间（按功能分类 API，如auth_api、student_api）；
• 自动生成 Swagger 风格的 API 文档（不用手动写文档）；
• 内置请求参数验证（避免手动判断参数）；
• 与 Flask 生态无缝兼容（复用已有模型、权限逻辑）。

三、实战 1：API 开发环境准备

基于前面的学生系统，新增 API 相关依赖和初始化配置。

1. 安装依赖

在服务器或本地的虚拟环境中安装 Flask-RESTX 和 JWT（API 认证用 JWT，无状态，适合多端）：

bash

运行

# 激活虚拟环境
source ~/student_system/venv/bin/activate  # 服务器
# 或本地（Windows）：venv\Scripts\activate

# 安装依赖
pip install flask-restx pyjwt

• flask-restx：API 开发核心库；
• pyjwt：生成和验证 JWT 令牌（API 认证用）。

2. 初始化 API（app.py改造）

在app.py中新增 API 初始化代码，复用已有 Flask 应用和数据库：

python

运行

# app.py（新增API相关代码，放在Flask-Login配置后面）
from flask_restx import Api, Resource, fields, Namespace
import jwt
from datetime import datetime, timedelta
from functools import wraps

# 1. 初始化API（设置API文档标题和版本）
api = Api(
    app,
    version='1.0',
    title='学生成绩管理系统API',
    description='支持多端访问的RESTful API接口，包含认证、学生管理、成绩管理',
    doc='/api/docs/'  # API文档访问路径（浏览器打开http://域名/api/docs/查看）
)

# 2. 定义JWT配置（密钥要和SECRET_KEY一致，过期时间2小时）
JWT_SECRET_KEY = app.config['SECRET_KEY']
JWT_EXPIRATION_DELTA = timedelta(hours=2)

# 3. 生成JWT令牌的工具函数
def generate_jwt_token(user_id, username, role):
    """生成JWT令牌：包含用户ID、用户名、角色，过期时间2小时"""
    payload = {
        'exp': datetime.utcnow() + JWT_EXPIRATION_DELTA,  # 过期时间
        'iat': datetime.utcnow(),  # 签发时间
        'sub': user_id,  # 主题（用户ID）
        'username': username,
        'role': role
    }
    # 用密钥签名令牌
    token = jwt.encode(payload, JWT_SECRET_KEY, algorithm='HS256')
    return token

# 4. API认证装饰器（验证JWT令牌）
def token_required(f):
    """API接口的JWT认证装饰器：未登录返回401，令牌无效返回401"""
    @wraps(f)
    def decorated(*args, **kwargs):
        token = None
        # 从请求头获取令牌（格式：Authorization: Bearer <token>）
        auth_header = request.headers.get('Authorization')
        if auth_header and auth_header.startswith('Bearer '):
            token = auth_header.split(' ')[1]
        
        if not token:
            # 返回401：未提供令牌
            return {'code': 401, 'msg': '未提供认证令牌，请先登录'}, 401
        
        try:
            # 验证令牌
            payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=['HS256'])
            # 从令牌中获取用户信息，传递给视图函数
            current_user_id = payload['sub']
            current_username = payload['username']
            current_role = payload['role']
        except jwt.ExpiredSignatureError:
            # 令牌过期
            return {'code': 401, 'msg': '认证令牌已过期，请重新登录'}, 401
        except jwt.InvalidTokenError:
            # 令牌无效
            return {'code': 401, 'msg': '无效的认证令牌'}, 401
        
        # 将用户信息传递给视图函数（通过kwargs）
        kwargs['current_user_id'] = current_user_id
        kwargs['current_username'] = current_username
        kwargs['current_role'] = current_role
        return f(*args, **kwargs)
    return decorated

# 5. API权限装饰器（管理员验证）
def api_admin_required(f):
    """API接口的管理员权限装饰器：非管理员返回403"""
    @wraps(f)
    def decorated(*args, **kwargs):
        # 从kwargs获取角色（token_required已传递）
        current_role = kwargs.get('current_role')
        if current_role != 'admin':
            return {'code': 403, 'msg': '无管理员权限，无法执行此操作'}, 403
        return f(*args, **kwargs)
    return decorated

# 6. 导入后续定义的命名空间（先占位，后面编写具体API）
# 注意：命名空间定义要放在此处后面，避免循环导入
# 后续会定义auth_ns（认证API）、student_ns（学生API）、course_ns（成绩API）

四、实战 2：开发核心 API 接口

按功能拆分 API 为 3 个命名空间（Namespace），每个命名空间对应一类 API，结构更清晰。

1. 命名空间 1：认证 API（auth_ns）

负责用户登录获取 JWT 令牌、刷新令牌，是所有 API 的入口。

python

运行

# app.py（新增认证API命名空间）
# 1. 定义认证命名空间（路径前缀：/api/auth）
auth_ns = api.namespace('auth', description='用户认证相关API', path='/api/auth')

# 2. 定义请求参数模型（API文档自动识别，验证参数）
login_model = auth_ns.model('LoginModel', {
    'username': fields.String(required=True, description='用户名', example='admin'),
    'password': fields.String(required=True, description='密码', example='Admin123!')
})

# 3. 认证API视图类（继承Resource）
@auth_ns.route('/login')
class AuthLogin(Resource):
    """用户登录：获取JWT令牌"""
    @auth_ns.expect(login_model)  # 关联请求参数模型
    @auth_ns.doc(responses={
        200: '登录成功，返回令牌',
        400: '参数错误',
        401: '用户名或密码错误'
    })
    def post(self):
        """用户登录，返回JWT令牌（有效期2小时）"""
        # 获取请求JSON数据
        data = request.get_json()
        username = data.get('username', '').strip()
        password = data.get('password', '').strip()
        
        # 验证参数
        if not username or not password:
            return {'code': 400, 'msg': '用户名和密码不能为空'}, 400
        
        # 查询用户并验证密码
        with app.app_context():
            user = User.query.filter_by(username=username).first()
            if not user or not user.check_password(password):
                # 记录登录失败日志
                log_operation(username, f'API登录失败：用户名或密码错误')
                return {'code': 401, 'msg': '用户名或密码错误'}, 401
        
        # 生成JWT令牌
        token = generate_jwt_token(user.id, user.username, user.role)
        # 记录登录成功日志
        log_operation(user.username, f'API登录成功，生成令牌')
        
        # 返回结果（包含令牌和用户信息）
        return {
            'code': 200,
            'msg': '登录成功',
            'data': {
                'token': token,
                'username': user.username,
                'role': user.role,
                'expires_in': int(JWT_EXPIRATION_DELTA.total_seconds())  # 过期时间（秒）
            }
        }, 200

@auth_ns.route('/refresh')
class AuthRefresh(Resource):
    """刷新JWT令牌（避免频繁登录）"""
    @token_required  # 需登录（验证旧令牌）
    @auth_ns.doc(responses={
        200: '刷新令牌成功',
        401: '令牌无效或过期'
    })
    def post(self, **kwargs):
        """用旧令牌刷新新令牌（新令牌有效期2小时）"""
        # 从kwargs获取用户信息（token_required传递）
        user_id = kwargs['current_user_id']
        username = kwargs['current_username']
        role = kwargs['current_role']
        
        # 生成新令牌
        new_token = generate_jwt_token(user_id, username, role)
        log_operation(username, f'API刷新令牌成功')
        
        return {
            'code': 200,
            'msg': '令牌刷新成功',
            'data': {
                'token': new_token,
                'expires_in': int(JWT_EXPIRATION_DELTA.total_seconds())
            }
        }, 200

2. 命名空间 2：学生管理 API（student_ns）

负责学生的增删改查，只有管理员能执行新增 / 编辑 / 删除，普通用户只能查询。

python

运行

# app.py（新增学生管理API命名空间）
# 1. 定义学生命名空间（路径前缀：/api/students）
student_ns = api.namespace('students', description='学生管理相关API', path='/api/students')

# 2. 定义请求/响应模型
student_model = student_ns.model('StudentModel', {
    'name': fields.String(required=True, description='学生姓名', example='小明'),
    'age': fields.Integer(required=True, description='学生年龄（10-30）', example=15)
})

# 3. 学生API视图类
@student_ns.route('')  # 路径：/api/students
class StudentList(Resource):
    """学生列表：查询所有学生（所有登录用户）、新增学生（管理员）"""
    @token_required  # 需登录
    @student_ns.doc(responses={
        200: '查询学生列表成功',
        401: '未登录'
    })
    def get(self, **kwargs):
        """查询所有学生列表（普通用户和管理员都可访问）"""
        with app.app_context():
            students = Student.query.all()
            # 整理响应数据
            student_list = []
            for student in students:
                student_list.append({
                    'id': student.id,
                    'name': student.name,
                    'age': student.age,
                    'created_at': student.created_at.strftime('%Y-%m-%d %H:%M:%S')
                })
        return {
            'code': 200,
            'msg': '查询成功',
            'data': student_list
        }, 200
    
    @token_required  # 需登录
    @api_admin_required  # 需管理员
    @student_ns.expect(student_model)  # 接收学生信息
    @student_ns.doc(responses={
        201: '新增学生成功',
        401: '未登录',
        403: '无管理员权限',
        400: '参数错误或学生已存在'
    })
    def post(self, **kwargs):
        """新增学生（仅管理员可访问）"""
        data = request.get_json()
        name = data.get('name', '').strip()
        age = data.get('age')
        username = kwargs['current_username']
        
        # 验证参数
        if not name or not isinstance(age, int) or not (10 <= age <= 30):
            return {'code': 400, 'msg': '参数错误：姓名不能为空，年龄需为10-30的整数'}, 400
        
        with app.app_context():
            # 检查学生是否已存在
            if Student.query.filter_by(name=name).first():
                return {'code': 400, 'msg': f'学生“{name}”已存在'}, 400
            
            # 新增学生
            new_student = Student(name=name, age=age)
            db.session.add(new_student)
            db.session.commit()
            
            # 记录日志
            log_operation(username, f'API新增学生：{name}（ID：{new_student.id}）')
        
        return {
            'code': 201,
            'msg': '新增学生成功',
            'data': {
                'id': new_student.id,
                'name': new_student.name,
                'age': new_student.age
            }
        }, 201

@student_ns.route('/<int:student_id>')  # 路径：/api/students/1
class StudentDetail(Resource):
    """学生详情：查询单个学生、编辑学生、删除学生"""
    @token_required
    @student_ns.doc(responses={
        200: '查询学生详情成功',
        401: '未登录',
        404: '学生不存在'
    })
    def get(self, student_id, **kwargs):
        """查询单个学生的详情（包含课程成绩）"""
        with app.app_context():
            student = Student.query.get(student_id)
            if not student:
                return {'code': 404, 'msg': f'ID为{student_id}的学生不存在'}, 404
            
            # 整理课程成绩
            courses = []
            total_score = 0
            for course in student.courses:
                courses.append({
                    'id': course.id,
                    'name': course.name,
                    'score': course.score,
                    'grade': 'A级(90+)' if course.score >=90 else 'B级(80-89)' if course.score>=80 else 'C级(<80)'
                })
                total_score += course.score
            avg_score = total_score / len(courses) if courses else 0
        
        return {
            'code': 200,
            'msg': '查询成功',
            'data': {
                'id': student.id,
                'name': student.name,
                'age': student.age,
                'courses': courses,
                'total_score': total_score,
                'avg_score': round(avg_score, 1),
                'created_at': student.created_at.strftime('%Y-%m-%d %H:%M:%S')
            }
        }, 200
    
    @token_required
    @api_admin_required
    @student_ns.expect(student_model)
    @student_ns.doc(responses={
        200: '编辑学生成功',
        401: '未登录',
        403: '无管理员权限',
        404: '学生不存在'
    })
    def put(self, student_id, **kwargs):
        """编辑学生信息（仅管理员可访问）"""
        data = request.get_json()
        name = data.get('name', '').strip()
        age = data.get('age')
        username = kwargs['current_username']
        
        with app.app_context():
            student = Student.query.get(student_id)
            if not student:
                return {'code': 404, 'msg': f'ID为{student_id}的学生不存在'}, 404
            
            # 验证参数并更新
            if name:
                # 检查新姓名是否已被其他学生使用
                if Student.query.filter_by(name=name).filter(Student.id != student_id).first():
                    return {'code': 400, 'msg': f'姓名“{name}”已被使用'}, 400
                student.name = name
            if isinstance(age, int) and 10 <= age <= 30:
                student.age = age
            
            db.session.commit()
            log_operation(username, f'API编辑学生：ID{student_id}（新姓名：{student.name}）')
        
        return {
            'code': 200,
            'msg': '编辑学生成功',
            'data': {
                'id': student.id,
                'name': student.name,
                'age': student.age
            }
        }, 200
    
    @token_required
    @api_admin_required
    @student_ns.doc(responses={
        200: '删除学生成功',
        401: '未登录',
        403: '无管理员权限',
        404: '学生不存在'
    })
    def delete(self, student_id, **kwargs):
        """删除学生（含关联课程，仅管理员可访问）"""
        username = kwargs['current_username']
        with app.app_context():
            student = Student.query.get(student_id)
            if not student:
                return {'code': 404, 'msg': f'ID为{student_id}的学生不存在'}, 404
            
            student_name = student.name
            # 删除学生（级联删除课程）
            db.session.delete(student)
            db.session.commit()
            log_operation(username, f'API删除学生：{student_name}（ID：{student_id}）')
        
        return {
            'code': 200,
            'msg': f'删除学生“{student_name}”成功'
        }, 200

3. 命名空间 3：成绩管理 API（course_ns）

负责课程成绩的增删改查，与学生 API 关联。

python

运行

# app.py（新增成绩管理API命名空间）
# 1. 定义成绩命名空间（路径前缀：/api/courses）
course_ns = api.namespace('courses', description='课程成绩管理相关API', path='/api/courses')

# 2. 定义请求模型
course_model = course_ns.model('CourseModel', {
    'student_id': fields.Integer(required=True, description='学生ID', example=1),
    'name': fields.String(required=True, description='课程名称', example='数学'),
    'score': fields.Integer(required=True, description='成绩（0-100）', example=85)
})

# 3. 成绩API视图类
@course_ns.route('')  # 路径：/api/courses
class CourseList(Resource):
    """成绩列表：查询所有成绩、新增成绩"""
    @token_required
    @course_ns.doc(responses={
        200: '查询成绩列表成功',
        401: '未登录'
    })
    def get(self, **kwargs):
        """查询所有课程成绩（普通用户可查，管理员可查所有）"""
        with app.app_context():
            courses = Course.query.all()
            course_list = []
            for course in courses:
                course_list.append({
                    'id': course.id,
                    'student_id': course.student_id,
                    'student_name': course.student.name,  # 关联学生姓名
                    'course_name': course.name,
                    'score': course.score,
                    'grade': 'A级(90+)' if course.score >=90 else 'B级(80-89)' if course.score>=80 else 'C级(<80)'
                })
        return {
            'code': 200,
            'msg': '查询成功',
            'data': course_list
        }, 200
    
    @token_required
    @api_admin_required
    @course_ns.expect(course_model)
    @course_ns.doc(responses={
        201: '新增成绩成功',
        401: '未登录',
        403: '无管理员权限',
        404: '学生不存在'
    })
    def post(self, **kwargs):
        """新增课程成绩（仅管理员可访问）"""
        data = request.get_json()
        student_id = data.get('student_id')
        course_name = data.get('name', '').strip()
        score = data.get('score')
        username = kwargs['current_username']
        
        # 验证参数
        if not student_id or not course_name or not isinstance(score, int) or not (0 <= score <= 100):
            return {'code': 400, 'msg': '参数错误：学生ID、课程名不能为空，成绩需为0-100的整数'}, 400
        
        with app.app_context():
            # 检查学生是否存在
            student = Student.query.get(student_id)
            if not student:
                return {'code': 404, 'msg': f'ID为{student_id}的学生不存在'}, 404
            
            # 检查该学生是否已存在该课程
            if Course.query.filter_by(student_id=student_id, name=course_name).first():
                return {'code': 400, 'msg': f'学生“{student.name}”已存在“{course_name}”课程'}, 400
            
            # 新增成绩
            new_course = Course(
                student_id=student_id,
                name=course_name,
                score=score
            )
            db.session.add(new_course)
            db.session.commit()
            log_operation(username, f'API新增成绩：学生{student.name}（ID：{student_id}）- {course_name}：{score}分')
        
        return {
            'code': 201,
            'msg': '新增成绩成功',
            'data': {
                'id': new_course.id,
                'student_id': student_id,
                'student_name': student.name,
                'course_name': course_name,
                'score': score
            }
        }, 201

@course_ns.route('/<int:course_id>')  # 路径：/api/courses/1
class CourseDetail(Resource):
    """成绩详情：编辑、删除成绩"""
    @token_required
    @api_admin_required
    @course_ns.expect(course_model)
    @course_ns.doc(responses={
        200: '编辑成绩成功',
        401: '未登录',
        403: '无管理员权限',
        404: '成绩不存在'
    })
    def put(self, course_id, **kwargs):
        """编辑课程成绩（仅管理员可访问）"""
        data = request.get_json()
        score = data.get('score')
        username = kwargs['current_username']
        
        with app.app_context():
            course = Course.query.get(course_id)
            if not course:
                return {'code': 404, 'msg': f'ID为{course_id}的成绩不存在'}, 404
            
            # 验证并更新成绩
            if isinstance(score, int) and 0 <= score <= 100:
                old_score = course.score
                course.score = score
                db.session.commit()
                log_operation(username, f'API编辑成绩：ID{course_id}（{course.name}）- 从{old_score}分改为{score}分')
            else:
                return {'code': 400, 'msg': '成绩需为0-100的整数'}, 400
        
        return {
            'code': 200,
            'msg': '编辑成绩成功',
            'data': {
                'id': course.id,
                'student_name': course.student.name,
                'course_name': course.name,
                'score': course.score
            }
        }, 200
    
    @token_required
    @api_admin_required
    @course_ns.doc(responses={
        200: '删除成绩成功',
        401: '未登录',
        403: '无管理员权限',
        404: '成绩不存在'
    })
    def delete(self, course_id, **kwargs):
        """删除课程成绩（仅管理员可访问）"""
        username = kwargs['current_username']
        with app.app_context():
            course = Course.query.get(course_id)
            if not course:
                return {'code': 404, 'msg': f'ID为{course_id}的成绩不存在'}, 404
            
            course_info = f'{course.student.name}-{course.name}（{course.score}分）'
            db.session.delete(course)
            db.session.commit()
            log_operation(username, f'API删除成绩：ID{course_id}（{course_info}）')
        
        return {
            'code': 200,
            'msg': f'删除成绩“{course_info}”成功'
        }, 200

五、实战 3：API 文档与测试

Flask-RESTX 会自动生成 API 文档，我们只需启动应用，即可在浏览器查看和测试 API。

1. 访问 API 文档

1. 启动 Flask 应用（或确保 Gunicorn 服务已运行）：

   bash

   运行

   # 本地测试（开发模式）
   flask run --host=0.0.0.0
   # 服务器（生产模式，已配置systemd）
   sudo systemctl restart student-system
   

2. 打开浏览器，访问http://你的域名/api/docs/（或http://localhost:5000/api/docs/），即可看到 Swagger 风格的 API 文档：

   • 左侧是 API 分类（auth、students、courses）；
   • 中间是 API 详情（请求方法、参数、响应示例）；
   • 右侧是 “Try it out” 按钮，可直接在浏览器测试 API。

2. 测试 API 步骤（以登录和查询学生为例）

步骤 1：测试登录 API（获取令牌）

1. 在 API 文档中找到「auth → /api/auth/login」，点击 “Try it out”；

2. 在请求体中输入管理员账号密码：

   json

   {
     "username": "admin",
     "password": "Admin123!"
   }
   

3. 点击 “Execute”，响应会返回 JWT 令牌：

   json

   {
     "code": 200,
     "msg": "登录成功",
     "data": {
       "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
       "username": "admin",
       "role": "admin",
       "expires_in": 7200
     }
   }
   

4. 复制token值，后续测试其他 API 需要用到。

步骤 2：测试查询学生 API（需令牌）

1. 找到「students → /api/students」，点击 “Try it out”；

2. 在 “Authorization” 请求头中输入Bearer <你的token>（注意 Bearer 后有空格）：

   plaintext

   Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
   

3. 点击 “Execute”，会返回所有学生列表：

   json

   {
     "code": 200,
     "msg": "查询成功",
     "data": [
       {
         "id": 1,
         "name": "小明",
         "age": 15,
         "created_at": "2024-06-10 10:00:00"
       }
     ]
   }
   

3. 用 Postman 测试 API（更灵活的工具）

浏览器文档适合快速测试，Postman 适合复杂场景（如批量测试、保存请求）：

1. 下载 Postman，新建 “请求集合”（如 “学生系统 API”）；
2. 新建 “登录请求”：
   • 方法：POST；
   • URL：http://你的域名/api/auth/login；
   • Body：选择 “raw → JSON”，输入登录参数；
   • 发送后，在响应中复制 token。
3. 新建 “查询学生请求”：
   • 方法：GET；
   • URL：http://你的域名/api/students；
   • Headers：添加Authorization: Bearer <token>；
   • 发送后，查看学生列表响应。

四、实战 4：API 多端集成示例

API 开发完成后，可对接 APP、小程序、Python 脚本等多端，这里演示两种常见场景。

1. 场景 1：Python 脚本调用 API（批量同步数据）

比如学校教务系统用 Python 脚本定期同步学生数据到我们的系统：

python

运行

# sync_students.py（从教务系统同步学生到本系统）
import requests
import json

# 配置
API_BASE_URL = "http://你的域名/api"
ADMIN_USERNAME = "admin"
ADMIN_PASSWORD = "Admin123!"

def get_jwt_token():
    """获取JWT令牌"""
    login_url = f"{API_BASE_URL}/auth/login"
    data = {
        "username": ADMIN_USERNAME,
        "password": ADMIN_PASSWORD
    }
    response = requests.post(login_url, json=data)
    result = response.json()
    if result["code"] == 200:
        return result["data"]["token"]
    else:
        raise Exception(f"登录失败：{result['msg']}")

def add_student(token, name, age):
    """调用API新增学生"""
    student_url = f"{API_BASE_URL}/students"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    data = {
        "name": name,
        "age": age
    }
    response = requests.post(student_url, headers=headers, json=data)
    result = response.json()
    if result["code"] in [200, 201]:
        print(f"新增学生成功：{name}")
    else:
        print(f"新增学生失败：{result['msg']}")

if __name__ == "__main__":
    # 1. 获取令牌
    try:
        token = get_jwt_token()
        print(f"获取令牌成功：{token[:20]}...")
    except Exception as e:
        print(e)
        exit()
    
    # 2. 模拟从教务系统获取的学生数据
    edu_students = [
        {"name": "小李", "age": 15},
        {"name": "小张", "age": 14},
        {"name": "小王", "age": 15}
    ]
    
    # 3. 批量同步学生
    for student in edu_students:
        add_student(token, student["name"], student["age"])
    print("批量同步完成！")

2. 场景 2：集成邮件通知 API（成绩变动通知）

当管理员修改学生成绩时，自动调用邮件 API 通知家长：

python

运行

# app.py（在编辑成绩的API中新增邮件通知）
def send_score_notify(student_name, course_name, old_score, new_score, parent_email):
    """发送成绩变动邮件通知"""
    subject = f"【成绩通知】{student_name}的{course_name}成绩已更新"
    content = f"""
    尊敬的家长：
    您好！{student_name}的{course_name}成绩已更新：
    旧成绩：{old_score}分
    新成绩：{new_score}分
    请及时关注孩子的学习情况。
    """
    # 复用之前的邮件发送函数（见第16篇的monitor_system.py）
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = SMTP_USER
    msg['To'] = parent_email
    
    try:
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASS)
            server.send_message(msg)
        return True
    except Exception as e:
        app.logger.error(f"发送成绩通知失败：{str(e)}")
        return False

# 在CourseDetail的put方法中调用（编辑成绩后）
@course_ns.route('/<int:course_id>')
class CourseDetail(Resource):
    def put(self, course_id, **kwargs):
        # 原有代码...
        if isinstance(score, int) and 0 <= score <= 100:
            old_score = course.score
            course.score = score
            db.session.commit()
            log_operation(username, f'API编辑成绩：ID{course_id}（{course.name}）- 从{old_score}分改为{score}分')
            
            # 新增：发送邮件通知（假设家长邮箱存在student的扩展字段中）
            # 实际场景中，可给Student模型新增parent_email字段
            parent_email = "parent@example.com"  # 实际应从数据库获取
            if parent_email:
                send_score_notify(
                    course.student.name,
                    course.name,
                    old_score,
                    score,
                    parent_email
                )
        # 原有代码...

五、API 安全与优化

API 暴露在公网，需要额外加固，避免被滥用。

1. API 限流（防止恶意请求）

复用前面的 Flask-Limiter，给 API 添加限流：

python

运行

# app.py（给API命名空间添加限流）
from flask_limiter.util import get_remote_address

# 给所有API添加限流：每小时最多100次请求（按IP）
api_limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["100 per hour"],
    storage_uri="memory://"
)

# 给认证API加更严格的限流（防止暴力破解）
@auth_ns.route('/login')
@api_limiter.limit("5 per 10 minute")  # 10分钟最多5次登录请求
class AuthLogin(Resource):
    # 原有代码...

2. 敏感数据脱敏

返回用户信息时，隐藏敏感字段（如管理员密码哈希，虽然 API 中不返回，但需确保）：

python

运行

# 在用户相关API中，绝对不返回password_hash
# 错误示例：return {'data': {'username': user.username, 'password_hash': user.password_hash}}
# 正确示例：只返回必要字段
return {'data': {'username': user.username, 'role': user.role}}

六、小结与后续方向

本篇核心收获

1. API 基础：理解 RESTful 规范（HTTP 方法、URL、状态码、JSON），掌握 Flask-RESTX 的使用；
2. 实战开发：基于已有系统开发认证、学生、成绩三类 API，复用权限逻辑和模型；
3. API 测试与集成：通过自动文档和 Postman 测试 API，实现 Python 脚本同步数据、邮件通知等多端集成；
4. API 安全：用 JWT 认证、限流、敏感数据脱敏保障 API 安全。

后续进阶方向

1. 对接移动端：基于 API 开发 React Native 或 Flutter APP，实现手机查成绩；
2. 小程序集成：开发微信小程序，调用 API 实现家长端成绩查询；
3. API 版本控制：当 API 需要升级时，通过版本（如/api/v2/students）兼容旧端；
4. API 监控：集成 Prometheus+Grafana 监控 API 调用量、响应时间，及时发现异常。

通过本篇的 API 开发，学生成绩系统从 “Web 单端” 升级为 “多端兼容” 的系统，具备了对接其他应用的能力，实用性和扩展性大幅提升。如果你跟着完成了所有步骤，不仅掌握了 Flask API 开发，还理解了 “后端服务化” 的核心思想，为后续开发分布式系统打下基础～