课程管理平台后端开发

需求

开发一个具有课程管理模块的后端系统，模块功能包含增删改查。

使用 MySQL + SQLAlchemy + Flask 完成整个后端流程，实现课程管理功能。

实现思路

uml diagram

接口设计

接口名称	接口入参	接口返回	请求 url	请求方法
课程新增	课程相关信息	创建成功	/course	post
课程删除	课程 id	删除成功	/course	delete
课程修改	课程相关信息	修改成功	/course	put
获取课程	课程 id	课程相关信息	/course	get
获取课程列表		课程列表相关信息	/course/list	get

开发

安装相关依赖

pip install SQLAlchemy
pip install pymysql
pip install pydantic
pip install flask

声明 SQLAlchemy 的 ORM 模型

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base


# 声明基类
Base = declarative_base()


# 定义表结构
class Course(Base):

    __tablename__ = "t_course"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(30))
    detail = Column(String(255), default="")

为了更为规范地定义和使用数据接口，可以使用 pydantic 库。pydantic 库是一种常用的用于数据接口 schema 定义与检查的库，通过这种方式，数据的传递将会更为安全。

Pydantic 在运行时强制执行类型提示，并在数据无效时提供用户友好的错误信息。

from typing import Optional
from pydantic import BaseModel, ConfigDict, constr


# 定义 pydantic 对象，用于数据验证与解析
class CourseModel(BaseModel):

    # 允许通过其他类实例来生成并初始化一个 Pydantic 实例
    model_config = ConfigDict(from_attributes=True)

    # id 属性为整型
    id: int
    # name 属性为字符串，最大长度为 30
    name: constr(max_length=30)
    # Optional 的作用是可选类型，告诉 IDE 或者框架这个参数除了给定的默认值外还可以是 None
    detail: Optional[constr(max_length=255)] = ""

使用 pydantic 对象进行数据验证与解析，更多的实现可以参考 pydantic 官方文档。

创建数据库

进入 mysql

执行以下命令创建数据库 course。

CREATE DATABASE course;

配置 MySQL 创建数据库表

在当前文件添加 python 代码并执行，注意替换正确的连接信息。

# 连接 mysql
# mysql 登录用户名
username = "root"
# mysql 登录密码
pwd = "hogwarts"
# mysql 域名
host = "127.0.0.1"
# mysql 端口
port = 3306
# 要操作的数据库名
database = "course"
engine = create_engine(f"mysql+pymysql://{username}:{pwd}@{host}:{port}/{database}")


if __name__ == '__main__':
    # 创建表
    Base.metadata.create_all(engine)

进入数据库，查看表是否成功创建。

use course;
show tables;

可以看到 t_course 表已经成功创建了。

定义 app 和创建 session 对象

# 定义 app
app = Flask(__name__)

# 创建会话
DBSession = sessionmaker(bind=engine)
# 获取会话实例
db_session: Session = DBSession()

编写课程管理增删改查接口

# 使用蓝图管理路由
course_manager = Blueprint("course", __name__, url_prefix="/course")

# 新增课程接口
# 蓝图对象可以直接调用对应请求方法，等价于 methods=["POST"]
@course_manager.post("")
def create_course():
    course_json = request.get_json()
    print(f"请求体为 {course_json}")
    # 验证传输过来的请求体是否正确
    try:
        # 使用 pydantic 把请求参数转为 pydantic 对象
        CourseModel.model_validate(course_json)
        # 如果没有错误则表示请求参数符合要求
    except ValidationError as e:
        # 如果有错误，则捕捉对应错误
        print(e)
        return {
            "errcode": 10002,
            # 通过 e.errors() 获取错误列表
            # repr 为内置函数，返回一个对象的 string 格式
            "errmsg": repr(e.errors())
        }
    # 通过课程名称和详情进行创建
    db_session.add(Course(**course_json))
    # 提交操作
    db_session.commit()
    # 关闭连接
    db_session.close()
    return {
        "errcode": 0,
        "errmsg": "创建课程成功！"
    }

# 获取课程
@course_manager.get(f"/<int:course_id>")
def get_course(course_id: int):
    print(f"要获取的课程 id 为 {course_id}")
    # 通过课程 id 在数据库中查询
    course_data = db_session.query(Course).filter_by(id=course_id).first()

    # 如果查询结果为空
    if course_data is None:
        return {
            "errcode": 10001,
            "errmsg": "课程不存在"
        }
    # 通过 pydantic 把查询结果对象转为字典格式
    datas = CourseModel.model_validate(course_data).model_dump()
    print(datas, type(datas))
    # 关闭连接
    db_session.close()
    return {
        "errcode": 0,
        "errmsg": f"获取 id 为 {course_id} 的课程信息成功！",
        "datas": datas
    }



# 获取全部课程列表
@course_manager.get("/list")
def get_courses():
    # 查询数据库获取全部课程列表
    course_datas = db_session.query(Course).all()
    print(f"获取到的课程列表为 {course_datas}")
    # 使用列表推导式获取全部数据列表，列表中元素为每个查询结果的字典格式
    datas = [CourseModel.model_validate(course).model_dump() for course in course_datas]
    # 关闭连接
    db_session.close()
    return {
        "errcode": 0,
        "errmsg": "获取全部课程列表成功！",
        "datas": datas
    }

# 修改课程
@course_manager.put("")
def update_course():
    # 获取 json 格式请求体
    course_json = request.get_json()
    print(f"请求体为 {course_json}")
    # 验证传输过来的请求体是否正确
    try:
        # 使用 pydantic 把请求参数转为 pydantic 对象
        CourseModel.model_validate(course_json)
    except ValidationError as e:
        print(e)
        return {
            "errcode": 10002,
            "errmsg": repr(e.errors())
        }
    # 构造课程对象
    course_data = Course(**course_json)
    # 根据课程 id 进行查询，如果查找不到结果则返回课程不存在
    if db_session.query(Course).filter_by(id=course_data.id).first() is None:
        return {
            "errcode": 10001,
            "errmsg": "课程不存在"
        }
    # 使用 pydantic 把请求数据转为字典
    data = CourseModel.model_validate(course_json).model_dump()
    print(data ,type(data))
    # 更新数据
    db_session.query(Course).filter_by(id=course_data.id).update(data)
    # 提交操作
    db_session.commit()
    # 关闭连接
    db_session.close()
    return {
        "errcode": 0,
        "errmsg": "课程修改成功"
    }

# 删除课程
@course_manager.delete(f"/<int:course_id>")
def delete_course(course_id: int):
    print(f"要删除的课程 id 为 {course_id}")
    # 根据id进行查询
    course_data = db_session.query(Course).filter(Course.id == course_id).first()
    # 如果查询结果为空，则返回课程不存在
    if course_data is None:
        return {
            "errcode": 10001,
            "errmsg": "课程不存在"
        }
    # 删除查找到的课程
    db_session.delete(course_data)
    # 提交操作
    db_session.commit()
    # 关闭连接
    db_session.close()
    return {
        "errcode": 0,
        "errmsg": "课程删除成功"
    }

运行程序

if __name__ == '__main__':
    app.register_blueprint(course_manager)
    app.run(debug=True, port=5055)

测试接口

可以通过 postman 或者 requests 进行接口测试，查看接口功能是否正常实现。

总结

环境准备
需求分析
接口设计
接口实现
- 声明 SQLAlchemy 的 ORM 模型
- 创建数据库
- 定义 app 创建 session 对象
- 编写课程管理增删改查接口