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

首先，我们先设计下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 Optionalfrom common.constant import SexEnum, RoleEnumclass UserBase(BaseModel):username: str = Field(min_length=8, max_length=16)class UserCreate(UserBase):"""请求验证模型"""password: str = Field(min_length=8, max_length=16)role: RoleEnumjob_num: Optional[str] = Field(None, min_length=8, max_length=8)student_num: Optional[str] = Field(None, min_length=16, max_length=16)sex: SexEnumage: 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 Enumclass 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_excfrom models.models import *
from models.user_schema import *
from common.json_tools import *
from common.log import loggerdef 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.idtry: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 SessionLocaldb = SessionLocal()exists = db.query(Role).all()if exists:print("roles already init")returnfor 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 Dependsfrom methods.user_method import *
from middlewares.mysql.database import create_db
from methods.user_method import db_create_useruser_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_routerapp.include_router(user_router, prefix="/user", tags=["users"])

测试：

 

 

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