文章
文章用户商铺专辑快讯圈子网址导航问答程序定制

Python 项目实战：用 Flask 实现 MySQL 数据库增删改查 API

一、项目概述与环境准备

1.1 项目目标

本项目将使用 Flask 框架构建一个完整的 RESTful API，实现对 MySQL 数据库的增删改查（CRUD）操作。我们将创建一个学生信息管理系统，包含学生信息的添加、查询、更新和删除功能。

1.2 技术栈

  • 后端框架：Flask 2.3.3
  • 数据库：MySQL 8.0
  • ORM：Flask-SQLAlchemy 3.0.1
  • 序列化：Flask-Marshmallow 0.15.0
  • 请求处理：Flask-RESTful 0.3.10
  • 跨域支持：Flask-CORS 4.0.0

1.3 环境准备

安装必要的包：

pip install flask flask-sqlalchemy flask-marshmallow flask-restful flask-cors mysql-connector-python

创建项目结构：

student-api/
├── app.py              # 主应用文件
├── config.py           # 配置文件
├── models.py          # 数据模型
├── schemas.py         # 序列化模式
├── resources.py       # API资源
└── requirements.txt   # 依赖包列表

二、数据库配置与模型定义

2.1 数据库配置

在 config.py中配置数据库连接：

class Config:
    # 数据库配置
    SQLALCHEMY_DATABASE_URI = 'mysql+mysqlconnector://username:password@localhost/student_db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_ECHO = True  # 打印SQL语句，调试用

2.2 创建数据库

在 MySQL 中创建数据库：

CREATE DATABASE student_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

2.3 定义数据模型

在 models.py中定义学生模型：

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Student(db.Model):
    """学生信息模型"""
    __tablename__ = 'students'
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    gender = db.Column(db.String(10), nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    phone = db.Column(db.String(20), unique=True, nullable=False)
    address = db.Column(db.String(200))
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    updated_at = db.Column(db.DateTime, default=db.func.current_timestamp(), 
                          onupdate=db.func.current_timestamp())
    
    def __repr__(self):
        return f'<Student {self.name}>'

三、序列化模式定义

在 schemas.py中定义序列化模式：

from flask_marshmallow import Marshmallow
from models import Student

ma = Marshmallow()

class StudentSchema(ma.SQLAlchemySchema):
    """学生信息序列化模式"""
    class Meta:
        model = Student
    
    id = ma.auto_field()
    name = ma.auto_field()
    age = ma.auto_field()
    gender = ma.auto_field()
    email = ma.auto_field()
    phone = ma.auto_field()
    address = ma.auto_field()
    created_at = ma.auto_field()
    updated_at = ma.auto_field()

# 创建实例
student_schema = StudentSchema()
students_schema = StudentSchema(many=True)

四、API资源实现

4.1 学生列表和创建接口

在 resources.py中实现：

from flask import request
from flask_restful import Resource
from models import db, Student
from schemas import student_schema, students_schema

class StudentListResource(Resource):
    """学生列表和创建资源"""
    
    def get(self):
        """获取所有学生列表"""
        students = Student.query.all()
        return students_schema.dump(students), 200
    
    def post(self):
        """创建新学生"""
        try:
            # 验证请求数据
            data = request.get_json()
            errors = student_schema.validate(data)
            if errors:
                return {'message': '数据验证失败', 'errors': errors}, 400
            
            # 检查邮箱和手机号是否已存在
            if Student.query.filter_by(email=data['email']).first():
                return {'message': '邮箱已存在'}, 400
            if Student.query.filter_by(phone=data['phone']).first():
                return {'message': '手机号已存在'}, 400
            
            # 创建学生
            student = Student(
                name=data['name'],
                age=data['age'],
                gender=data['gender'],
                email=data['email'],
                phone=data['phone'],
                address=data.get('address')
            )
            
            db.session.add(student)
            db.session.commit()
            
            return student_schema.dump(student), 201
            
        except Exception as e:
            db.session.rollback()
            return {'message': f'创建失败: {str(e)}'}, 500

4.2 学生详情、更新和删除接口

继续在 resources.py中添加：

class StudentResource(Resource):
    """学生详情、更新和删除资源"""
    
    def get(self, student_id):
        """根据ID获取学生详情"""
        student = Student.query.get_or_404(student_id)
        return student_schema.dump(student), 200
    
    def put(self, student_id):
        """更新学生信息"""
        try:
            student = Student.query.get_or_404(student_id)
            data = request.get_json()
            
            # 验证数据
            errors = student_schema.validate(data, partial=True)
            if errors:
                return {'message': '数据验证失败', 'errors': errors}, 400
            
            # 检查邮箱和手机号是否被其他学生使用
            if data.get('email') and data['email'] != student.email:
                if Student.query.filter(Student.email == data['email'], Student.id != student_id).first():
                    return {'message': '邮箱已被其他学生使用'}, 400
            
            if data.get('phone') and data['phone'] != student.phone:
                if Student.query.filter(Student.phone == data['phone'], Student.id != student_id).first():
                    return {'message': '手机号已被其他学生使用'}, 400
            
            # 更新学生信息
            if 'name' in data:
                student.name = data['name']
            if 'age' in data:
                student.age = data['age']
            if 'gender' in data:
                student.gender = data['gender']
            if 'email' in data:
                student.email = data['email']
            if 'phone' in data:
                student.phone = data['phone']
            if 'address' in data:
                student.address = data['address']
            
            db.session.commit()
            return student_schema.dump(student), 200
            
        except Exception as e:
            db.session.rollback()
            return {'message': f'更新失败: {str(e)}'}, 500
    
    def delete(self, student_id):
        """删除学生"""
        try:
            student = Student.query.get_or_404(student_id)
            db.session.delete(student)
            db.session.commit()
            return {'message': '学生删除成功'}, 200
            
        except Exception as e:
            db.session.rollback()
            return {'message': f'删除失败: {str(e)}'}, 500

4.3 学生查询接口

添加查询功能：

class StudentSearchResource(Resource):
    """学生查询资源"""
    
    def get(self):
        """根据条件查询学生"""
        try:
            name = request.args.get('name')
            gender = request.args.get('gender')
            min_age = request.args.get('min_age')
            max_age = request.args.get('max_age')
            
            query = Student.query
            
            if name:
                query = query.filter(Student.name.like(f'%{name}%'))
            if gender:
                query = query.filter_by(gender=gender)
            if min_age:
                query = query.filter(Student.age >= int(min_age))
            if max_age:
                query = query.filter(Student.age <= int(max_age))
            
            students = query.all()
            return students_schema.dump(students), 200
            
        except Exception as e:
            return {'message': f'查询失败: {str(e)}'}, 500

五、主应用文件

在 app.py中配置应用：

from flask import Flask
from flask_restful import Api
from flask_cors import CORS
from config import Config
from models import db
from schemas import ma
from resources import StudentListResource, StudentResource, StudentSearchResource

def create_app():
    """创建Flask应用"""
    app = Flask(__name__)
    app.config.from_object(Config)
    
    # 初始化扩展
    db.init_app(app)
    ma.init_app(app)
    
    # 启用跨域
    CORS(app)
    
    # 创建API
    api = Api(app)
    
    # 注册路由
    api.add_resource(StudentListResource, '/api/students')
    api.add_resource(StudentResource, '/api/students/<int:student_id>')
    api.add_resource(StudentSearchResource, '/api/students/search')
    
    # 创建数据库表
    with app.app_context():
        db.create_all()
    
    return app

if __name__ == '__main__':
    app = create_app()
    app.run(debug=True, host='0.0.0.0', port=5000)

六、API接口测试

6.1 创建学生

请求：

curl -X POST http://localhost:5000/api/students \
  -H "Content-Type: application/json" \
  -d '{
    "name": "张三",
    "age": 20,
    "gender": "男",
    "email": "zhangsan@example.com",
    "phone": "13800138000",
    "address": "北京市海淀区"
  }'

响应：

{
  "id": 1,
  "name": "张三",
  "age": 20,
  "gender": "男",
  "email": "zhangsan@example.com",
  "phone": "13800138000",
  "address": "北京市海淀区",
  "created_at": "2024-01-15T10:30:00",
  "updated_at": "2024-01-15T10:30:00"
}

6.2 获取学生列表

请求：

curl http://localhost:5000/api/students

响应：

[
  {
    "id": 1,
    "name": "张三",
    "age": 20,
    "gender": "男",
    "email": "zhangsan@example.com",
    "phone": "13800138000",
    "address": "北京市海淀区",
    "created_at": "2024-01-15T10:30:00",
    "updated_at": "2024-01-15T10:30:00"
  }
]

6.3 获取学生详情

请求：

curl http://localhost:5000/api/students/1

响应：

{
  "id": 1,
  "name": "张三",
  "age": 20,
  "gender": "男",
  "email": "zhangsan@example.com",
  "phone": "13800138000",
  "address": "北京市海淀区",
  "created_at": "2024-01-15T10:30:00",
  "updated_at": "2024-01-15T10:30:00"
}

6.4 更新学生信息

请求：

curl -X PUT http://localhost:5000/api/students/1 \
  -H "Content-Type: application/json" \
  -d '{
    "age": 21,
    "address": "北京市朝阳区"
  }'

响应：

{
  "id": 1,
  "name": "张三",
  "age": 21,
  "gender": "男",
  "email": "zhangsan@example.com",
  "phone": "13800138000",
  "address": "北京市朝阳区",
  "created_at": "2024-01-15T10:30:00",
  "updated_at": "2024-01-15T10:35:00"
}

6.5 删除学生

请求：

curl -X DELETE http://localhost:5000/api/students/1

响应：

{
  "message": "学生删除成功"
}

6.6 查询学生

请求：

curl "http://localhost:5000/api/students/search?name=张&gender=男&min_age=18&max_age=25"

响应：

[
  {
    "id": 1,
    "name": "张三",
    "age": 21,
    "gender": "男",
    "email": "zhangsan@example.com",
    "phone": "13800138000",
    "address": "北京市朝阳区",
    "created_at": "2024-01-15T10:30:00",
    "updated_at": "2024-01-15T10:35:00"
  }
]

七、错误处理

7.1 数据验证错误

请求（缺少必填字段）：

curl -X POST http://localhost:5000/api/students \
  -H "Content-Type: application/json" \
  -d '{
    "name": "李四",
    "age": 22
  }'

响应：

{
  "message": "数据验证失败",
  "errors": {
    "email": ["缺少数据"],
    "gender": ["缺少数据"],
    "phone": ["缺少数据"]
  }
}

7.2 邮箱或手机号重复

请求：

curl -X POST http://localhost:5000/api/students \
  -H "Content-Type: application/json" \
  -d '{
    "name": "王五",
    "age": 23,
    "gender": "男",
    "email": "zhangsan@example.com",
    "phone": "13800138001",
    "address": "北京市海淀区"
  }'

响应：

{
  "message": "邮箱已存在"
}

7.3 学生不存在

请求：

curl http://localhost:5000/api/students/999

响应：

{
  "message": "The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again."
}

八、高级功能扩展

8.1 分页查询

在 StudentListResource中添加分页功能：

def get(self):
    """获取学生列表（分页）"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    
    students = Student.query.paginate(
        page=page, 
        per_page=per_page, 
        error_out=False
    )
    
    return {
        'items': students_schema.dump(students.items),
        'total': students.total,
        'pages': students.pages,
        'current_page': page,
        'per_page': per_page
    }, 200

8.2 数据验证增强

在 schemas.py中增强验证规则：

from marshmallow import validate

class StudentSchema(ma.SQLAlchemySchema):
    class Meta:
        model = Student
    
    id = ma.auto_field()
    name = ma.auto_field(validate=validate.Length(min=1, max=100))
    age = ma.auto_field(validate=validate.Range(min=0, max=150))
    gender = ma.auto_field(validate=validate.OneOf(['男', '女']))
    email = ma.auto_field(validate=validate.Email())
    phone = ma.auto_field(validate=validate.Length(min=11, max=20))
    address = ma.auto_field(validate=validate.Length(max=200))
    created_at = ma.auto_field()
    updated_at = ma.auto_field()

8.3 文件上传

添加头像上传功能：

from flask import request, current_app
import os
from werkzeug.utils import secure_filename

class StudentAvatarResource(Resource):
    """学生头像上传资源"""
    
    def post(self, student_id):
        """上传头像"""
        try:
            student = Student.query.get_or_404(student_id)
            
            if 'avatar' not in request.files:
                return {'message': '请选择头像文件'}, 400
            
            file = request.files['avatar']
            if file.filename == '':
                return {'message': '请选择头像文件'}, 400
            
            if file and allowed_file(file.filename):
                filename = secure_filename(f"student_{student_id}_{file.filename}")
                filepath = os.path.join(current_app.config['UPLOAD_FOLDER'], filename)
                file.save(filepath)
                
                student.avatar = filename
                db.session.commit()
                
                return {'message': '头像上传成功', 'avatar': filename}, 200
            else:
                return {'message': '文件类型不支持'}, 400
                
        except Exception as e:
            db.session.rollback()
            return {'message': f'上传失败: {str(e)}'}, 500

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in {'png', 'jpg', 'jpeg', 'gif'}

8.4 用户认证

添加简单的JWT认证：

from flask_jwt_extended import JWTManager, jwt_required, create_access_token

# 配置JWT
app.config['JWT_SECRET_KEY'] = 'your-secret-key'
jwt = JWTManager(app)

class LoginResource(Resource):
    """登录接口"""
    
    def post(self):
        data = request.get_json()
        username = data.get('username')
        password = data.get('password')
        
        # 这里简化验证，实际应查询数据库
        if username == 'admin' and password == 'password':
            access_token = create_access_token(identity=username)
            return {'access_token': access_token}, 200
        else:
            return {'message': '用户名或密码错误'}, 401

# 在需要认证的资源上添加装饰器
class ProtectedStudentResource(Resource):
    @jwt_required()
    def get(self, student_id):
        student = Student.query.get_or_404(student_id)
        return student_schema.dump(student), 200

九、项目部署

9.1 创建 requirements.txt

pip freeze > requirements.txt

9.2 使用 Gunicorn 部署

pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:5000 app:app

9.3 使用 Nginx 反向代理

Nginx 配置：

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;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

9.4 使用 Supervisor 管理进程

创建 supervisor 配置文件：

[program:student-api]
command=/path/to/venv/bin/gunicorn -w 4 -b 127.0.0.1:5000 app:app
directory=/path/to/student-api
user=www-data
autostart=true
autorestart=true
redirect_stderr=true

十、总结

通过本项目的实践，我们完整地实现了一个基于 Flask 的 RESTful API，具备以下特点：

  1. 完整的 CRUD 操作：支持学生信息的增删改查
  2. 数据验证：使用 Marshmallow 进行数据验证和序列化
  3. 错误处理：完善的错误处理机制
  4. 分页查询：支持分页查询学生列表
  5. 条件查询：支持按姓名、性别、年龄范围查询
  6. 扩展功能：支持文件上传、用户认证等高级功能
  7. 部署方案：提供生产环境部署方案

这个项目可以作为学习 Flask 和 RESTful API 开发的入门项目，也可以作为实际项目的基础模板进行扩展。

版权声明：本文为JienDa博主的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
若内容若侵犯到您的权益，请发送邮件至：platform_service@jienda.com我们将第一时间处理！
所有资源仅限于参考和学习，版权归JienDa作者所有，更多请访问JienDa首页。

给TA赞助
共{{data.count}}人
人已赞助
后端

企业级部署升级：Nginx反向代理+ELK日志监控，让成绩预测平台稳定可追溯

2025-12-19 21:28:38

后端

【Java 开发日记】我们来说一说 Redis 主从复制的原理及作用

2025-12-19 21:34:54

0 条回复 A文章作者 M管理员
    暂无讨论，说说你的看法吧
❯

解锁会员权限

开通会员

解锁海量优质VIP资源

立刻开通
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索
客服

  • 扫码打开当前页

  • 微信小程序

  • App下载

  • 百度小程序

  • 抖音小程序

  • 微信客服

返回顶部