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

在前面我们分析了接口的设计，那么我们接下来做接口的开发。

首先，我们先设计下pydantic用户参数的校验：

"""
-*- encoding=utf-8 -*-
Time: 2024/7/19 16:48
Author: lc
Email: 15101006331@163.com
File: schemas.py
"""
from pydantic import BaseModel, Field
from typing import Optional

from common.constant import SexEnum, RoleEnum


class UserBase(BaseModel):
    username: str = Field(min_length=8, max_length=16)


class UserCreate(UserBase):
    """请求验证模型"""
    password: str = Field(min_length=8, max_length=16)
    role: RoleEnum
    job_num: Optional[str] = Field(None, min_length=8, max_length=8)
    student_num: Optional[str] = Field(None, min_length=16, max_length=16)
    sex: SexEnum
    age: int = Field(ge=18, le=65)

因为role和sex的值是固定的，所以我创建了两个枚举类来限制这两个字段

"""
-*- encoding=utf-8 -*-
Time: 2024/7/22 10:24
Author: lc
Email: 15101006331@163.com
File: constant.py
"""
from enum import Enum


class SexEnum(str, Enum):
    male = "男"
    female = "女"


class RoleEnum(str, Enum):
    student = "学生"
    teacher = "老师"

接着，我设计了注册/新增用户的主要逻辑

"""
-*- encoding=utf-8 -*-
Time: 2024/7/22 10:06
Author: lc
Email: 15101006331@163.com
File: user.py
"""

from sqlalchemy.orm import Session
from passlib.context import CryptContext
from traceback import format_exc

from models.models import *
from models.user_schema import *
from common.json_tools import *
from common.log import logger


def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id, User.status == 0).first()


def db_create_user(db: Session, user: UserCreate):
    logger.info("创建用户开始")
    db_user = get_by_username(db, user.username)
    if db_user:
        return response(code=100104, message="用户名已存在")
    if (user.role == "学生" and user.student_num is None) or (user.role == "老师" and user.job_num is None):
        return response(code=100102, message="身份和对应号不匹配")
    try:
        user.password = get_password_hash(user.password)
    except:
        logger.warning(f"method db_create_user error: {format_exc()}")
        return response(code=100105, message="密码加密失败")
    user_to_db = User(**user.dict())
    db_role = db.query(Role).filter(Role.name == user.role.value).first()
    user_to_db.role = db_role.id
    try:
        db.add(user_to_db)
        db.commit()  # 提交保存到数据库
        db.refresh(user_to_db)  # 刷新
        logger.info(f"创建用户:‘{user.username}’成功")
    except:
        logger.warning(f"method db_create_user error: {format_exc()}")
        return response(code=100101, message="注册失败，请重试")
    return response(code=200, data=user_to_db.username)



def get_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username, User.status == 0).first()


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)

接下来，我们看下注册接口的逻辑

1.校验参数是否合规（这里我使用的pydantic自身的校验，见user_schemas）

2.判断用户名是否已存在

3.加密密码

4.保存到数据库

其次，我在models中新增了初始化roles的方法，并将该方法加入到了startup中间件中，每次启动程序都会检查是否要初始化该数据

def init_roles():
    from middlewares.mysql.database import SessionLocal
    db = SessionLocal()
    exists = db.query(Role).all()
    if exists:
        print("roles already init")
        return
    for index, sex in enumerate(("老师", "学生")):
        role = Role(id=index+1, name=sex)
        db.add(role)
        db.commit()
        db.refresh(role)
    db.close()

@app.on_event("startup")
async def startup_event():
    app.state.redis = await create_redis()
    print("init redis success")
    create_tables()
    print("init database success")
    init_roles()
    print("init roles success")

然后我们要实现注册接口：

"""
-*- encoding=utf-8 -*-
Time: 2024/7/22 09:59
Author: lc
Email: 15101006331@163.com
File: user.py
"""
from fastapi import APIRouter
from fastapi import Depends


from methods.user_method import *
from middlewares.mysql.database import create_db
from methods.user_method import db_create_user

user_router = APIRouter()


@user_router.post("/", summary="创建用户")
def create_user(user: UserCreate, db: Session = Depends(create_db)):
    return db_create_user(db, user)

最后，我们要把新增的路由注册到main中的app。

from models.models import create_tables, init_roles
from routers.user import user_router


app.include_router(user_router, prefix="/user", tags=["users"])

测试：

到这里我们基本完成了注册接口的开发，使用 到了知识点passlib。