网络运营是干什么的信息流优化师证书

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

在之前的文章：FastAPI（六十九)实战开发《在线课程学习系统》接口开发--修改密码，这次分享留言功能开发

我们梳理下对应的逻辑

1.校验用户是否登录

2.校验留言的用户是否存在

3.校验用户是否和留言的用户一致

4.校验留言长度是否符合要求

5.增加留言

从这里开始是留言相关的接口，因此我新建了以下几个文件来单独维护留言相关接口逻辑：

首先，看下对应的pydantic参数模型

"""
-*- encoding=utf-8 -*-
Time: 2024/7/23 15:31
Author: lc
Email: 15101006331@163.com
File: message_method.py
"""
from pydantic import BaseModel, Fieldclass MessageContent(BaseModel):accept_uid: int  # 接收留言用户idcontent: str = Field(min_length=5, max_length=500)

接下来实现接口逻辑，message_method.py

"""
-*- encoding=utf-8 -*-
Time: 2024/7/23 15:31
Author: lc
Email: 15101006331@163.com
File: message_method.py
"""
import traceback
from datetime import datetime
from sqlalchemy.orm import Sessionfrom methods.user_method import get_by_username, get_by_uid
from models.message_schemas import MessageContent
from models.models import Message
from models.user_schema import UsernameRole
from common.log import logger
from common.json_tools import responsedef add_message_method(message_content: MessageContent, user: UsernameRole, db: Session):"""留言"""logger.info(f"用户:{user.username}来留言了")accept_user = get_by_uid(db, message_content.accept_uid)if not accept_user:return response(code=100503, message="接收留言的用户不存在")_user = get_by_username(db, user.username)if accept_user.id == _user.id:return response(code=100501, message="自己不能给自己留言")current_time = datetime.strftime(datetime.now(), "%Y-%m_%d %H:%M:%S")try:message = Message(send_user=_user.id,accept_user=accept_user.id,context=message_content.content,send_time=current_time,add_time=current_time)db.add(message)db.commit()db.refresh(message)except:logger.warning(f"method add_message_method error：{traceback.format_exc()}")return response(code=100502, message="留言失败")logger.info(f"用户：{user.username}留言成功")return response()

在message.py中实现接口

"""
-*- encoding=utf-8 -*-
Time: 2024/7/23 15:31
Author: lc
Email: 15101006331@163.com
File: message_router.py
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Sessionfrom middlewares.mysql.database import create_db
from models.message_schema import *
from models.user_schema import UsernameRole
from methods.user_method import get_current_user
from methods.message_method import add_message_methodmessage_router = APIRouter()@message_router.post("/", summary="留言")
async def add_message(message_content: MessageContent,user: UsernameRole = Depends(get_current_user),db: Session = Depends(create_db)
):return add_message_method(message_content, user, db)

最后将留言相关路由注册到main.py中的app下

from routers.message import message_routerapp.include_router(message_router, prefix="/message", tags=["messages"])

测试：

至此，留言接口完成。