源码见："fastapi_study_road-learning_system_online_courses: fastapi框架实战之--在线课程学习系统"

上次我们分享了，FastAPI（七十四）实战开发《在线课程学习系统》接口开发-- 删除留言

从本篇文章开始，我们分享课程相关的api

还是按照惯例，梳理下对应的逻辑

1.判断用户是否登录

2.校验是否有权限

3.校验参数是否完整（本步骤我们使用pydantic自校验）

4.查询课程是否存在

5.创建课程

课程我们也需要单独创建一个新的路由来管理

我们在routers创建course.py

from fastapi import APIRouter

course_router = APIRouter()

接着我们在main.py中引入并注册给app

from routers.course import course_router

app.include_router(course_router, prefix="/course", tags=["courses"])

这样我们的新路由就生效了

接下来，我们看下如何实现创建课程？

首先，创建课程的pydantic参数类，同样的我们新建一个course_schema.py来单独管理

from pydantic import BaseModel, Field
from typing import Optional

from common.constant import ShelfEnum


class Courses(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    icon: Optional[str] = None
    desc: Optional[str] = None
    catalog: Optional[str] = None
    onsale: Optional[SaleEnum] = None
    owner: Optional[int] = None
    like_num: Optional[int] = 0

其中 SaleEnum 如下：

class SaleEnum(int, Enum):
    ZERO = 0
    ONE = 1

接下来，我们创建 course_method.py，在这里我们实现有关课程的所有逻辑

"""
-*- encoding=utf-8 -*-
Time: 2024/7/25 09:53
Author: lc
Email: 15101006331@163.com
File: course_method.py
"""
from sqlalchemy.orm import Session

from common.json_tools import response
from methods.user_method import get_by_username
from models.course_schema import Courses
from models.models import Course
from models.user_schema import UsernameRole
from common.log import logger


def get_course_by_name(db: Session, name: str):
    """根据名称获取课程"""
    return db.query(Course).filter(Course.name == name, Course.status == False).first()


def create_course(db: Session, course: Courses, user: UsernameRole):
    """创建课程"""
    user_role = user.role
    if not user_role or user_role == "学生":
        return response(code=101004, message="只有老师才可以创建课程")
    db_course = get_course_by_name(db, course.name)
    if db_course:
        return response(code=101002, message="该课程名称已存在，不允许重复")
    try:
        db_user = get_by_username(db, user.username)
        course.owner = db_user.id
        course = Course(**course.dict())
        db.add(course)
        db.commit()
        db.refresh(course)
    except:
        logger.warning(f"创建课程：{course.name} 失败")
        return response(code=101001, message="创建课程失败")
    logger.info(f"创建课程：{course.name} 成功")
    return response()

最后，我们在新建course.py，在其中实现接口api

"""
-*- encoding=utf-8 -*-
Time: 2024/7/25 09:37
Author: lc
Email: 15101006331@163.com
File: course.py
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from methods.course_method import create_course
from middlewares.mysql.database import create_db
from models.course_schema import Courses
from models.user_schema import UsernameRole
from methods.user_method import get_current_user

course_router = APIRouter()


@course_router.post("/", summary="创建课程")
def create(course: Courses, db: Session = Depends(create_db), user: UsernameRole = Depends(get_current_user)):
    return create_course(db, course, user)

测试：

以上就是创建课程的逻辑