用户注册

I.请求与响应过程

flowchart  LR

    subgraph B [后端]
        U[FastAPI服务]
    end

    subgraph A [用户界面]
        direction LR
        F1[用户注册]
    end

   A -- ① 请求分类<br> /api/user/register --> B
   B -- ② 返回响应 <br> {'code': 200,'message': 'success','token':用户令牌','data': [{'id': 1,'username': 用户名}]}  -->  A

II. 数据校验模型

在进行用户注册时，前端会传递用户注册信息，为了保证用户数据的准确性，需要对应传递的数据进行校验

在schemas目录下创建一个users.py文件

from pydantic import BaseModel

class UserRegisterRequest(BaseModel):
    """
    用户注册请求数据模型
    """
    username: str  # 用户名，必填
    password: str  # 密码，必填

    class Config:
        schema_extra = {
            "example": {
                "username": "example_user",
                "password": "example_password"
            }
        }

III. 路由定义

在routers目录下创建一个users.py文件，编写路由解析代码

注意：注册时保存数据，需要使用post

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from db_conf import get_db
from schemas.users import UserRegisterRequest

router = APIRouter(
    prefix="/api/user",
    tags=["user"]
)


@router.post("/register")
async def register_user(user_data: UserRegisterRequest,db: AsyncSession = Depends(get_db)):
    """
    用户注册接口

    Args:
        user_data: 用户注册请求数据
        db: 数据库会话

    Returns:
        UserRegisterResponse: 注册成功响应
    """

    # 返回成功响应（包含token）
    return {
        "code": 200,
        "message": "注册成功",
        "data": '用户注册'
    }

在main.py文件中进行路由注册

from routers import news,users

# ....其他代码....

app.include_router(users.router)

重启FastAPI服务，运行测试，检查定义的路由接口是否能够处理前端发送的请求

post请求可以借助http工具

POST http://127.0.0.1:8000/api/user/register
Content-Type: application/json

{
  "username": "test",
  "password": "password"
}

IV. 定义模型类

定义对应的模型类，方便对应数据库中的数据进行ORM操作

在models目录下创建users.py文件，编写模型类

from sqlalchemy.orm import Mapped, mapped_column,DeclarativeBase

from sqlalchemy import String, Integer, DateTime, ForeignKey, Enum, Index
from datetime import datetime
from typing import Optional


class Base(DeclarativeBase):
    pass

class User(Base):
    """
    用户信息表ORM模型
    """
    __tablename__ = 'user'

    # 创建索引
    __table_args__ = (
        Index('username_UNIQUE', 'username'),
        Index('phone_UNIQUE', 'phone'),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="用户ID")
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, comment="用户名")
    password: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码（加密存储）")
    nickname: Mapped[Optional[str]] = mapped_column(String(50), comment="昵称")
    avatar: Mapped[Optional[str]] = mapped_column(String(255), comment="头像URL",
                                                  default='https://fastly.jsdelivr.net/npm/@vant/assets/cat.jpeg')
    gender: Mapped[Optional[str]] = mapped_column(Enum('male', 'female', 'unknown'), comment="性别", default='unknown')
    bio: Mapped[Optional[str]] = mapped_column(String(500), comment="个人简介", default='这个人很懒，什么都没留下')
    phone: Mapped[Optional[str]] = mapped_column(String(20), unique=True, comment="手机号")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now(), comment="创建时间")
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now(), onupdate=datetime.now(),
                                                 comment="更新时间")


    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}', nickname='{self.nickname}')>"


class UserToken(Base):
    """
    用户令牌表ORM模型
    """
    __tablename__ = 'user_token'

    # 创建索引
    __table_args__ = (
        Index('token_UNIQUE', 'token'),
        Index('fk_user_token_user_idx', 'user_id'),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="令牌ID")
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey(User.id), nullable=False, comment="用户ID")
    token: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, comment="令牌值")
    expires_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="过期时间")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now(), comment="创建时间")


    def __repr__(self):
        return f"<UserToken(id={self.id}, user_id={self.user_id}, token='{self.token}')>"

UserToken表的作用时记录用户的状态信息，相当于用户的身份标识。

如果用户登录或注册成功，则给用户生成一个身份标识（token），然后token会返回给前端，前端会存储该token值

当用户我的页面中进行收藏，浏览历史操作时，前端会携带token，后端会对token信息进行验证，确认用户身份。

如果token失效则需要用户重新登录。

如果token错误也会让用户重新登录。

如果没有携带token也需要用户重新登录。

V. 编写ORM数据操作

在curd目录下创建users.py文件，将用户模块的数据操作按照对应的逻辑编写代码

主要实现以下功能:

密码混淆处理，避免铭文保存密码
根据用户查询用户是否存在
创建用户
创建token

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from passlib.context import CryptContext
import uuid
from datetime import datetime, timedelta

from models.users import User
from schemas.users import UserRegisterRequest
from models.users import UserToken


# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def get_password_hash(password: str) -> str:
    """
    对密码进行哈希处理

    Args:
        password: 明文密码

    Returns:
        str: 哈希后的密码
    """
    return pwd_context.hash(password)


async def get_user_by_username(db: AsyncSession, username: str) -> User:
    """
    根据用户名获取用户

    Args:
        db: 数据库会话
        username: 用户名

    Returns:
        User: 用户对象，如果不存在返回None
    """
    query = select(User).where(User.username == username)
    result = await db.execute(query)
    return result.scalar_one_or_none()


async def create_user(db: AsyncSession, user_data: UserRegisterRequest) -> User:
    """
    创建新用户

    Args:
        db: 数据库会话
        user_data: 用户注册数据

    Returns:
        User: 创建的用户对象
    """
    # 对密码进行哈希处理
    hashed_password = get_password_hash(user_data.password)

    # 创建用户对象
    db_user = User(
        username=user_data.username,
        password=hashed_password

    )

    # 添加到数据库
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)

    return db_user

async def create_user_token(db: AsyncSession, user_id: int) -> str:
    """
    为用户创建访问令牌

    Args:
        db: 数据库会话
        user_id: 用户ID

    Returns:
        str: 访问令牌
    """
    # 生成随机token
    token = str(uuid.uuid4())

    # 设置过期时间（7天）
    expires_at = datetime.now() + timedelta(days=7)

    # 查询用户是否已有令牌记录
    query = select(UserToken).where(UserToken.user_id == user_id)
    result = await db.execute(query)
    existing_token = result.scalar_one_or_none()

    if existing_token:
        # 如果用户已有令牌，则更新令牌信息
        existing_token.token = token
        existing_token.expires_at = expires_at
        await db.commit()
    else:
        # 如果用户没有令牌，则创建新的令牌记录
        db_token = UserToken(
            user_id=user_id,
            token=token,
            expires_at=expires_at
        )
        db.add(db_token)
        await db.commit()

    return token

VI. 路由中实现数据操作

回到对应的路由中，调用刚才编写好的数据处理函数。

@router.post("/register")
async def register_user(
        user_data: UserRegisterRequest,
        db: AsyncSession = Depends(get_db)
):
    """
    用户注册接口

    Args:
        user_data: 用户注册请求数据
        db: 数据库会话

    Returns:
        UserRegisterResponse: 注册成功响应
    """
    try:
        # 检查用户名是否已存在
        existing_user = await users.get_user_by_username(db, user_data.username)
        if existing_user:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="用户名已存在"
            )

        # 调用CRUD函数创建用户
        user = await users.create_user(db, user_data)

        # 创建访问令牌
        token = await users.create_user_token(db, user.id)

        # 返回成功响应（包含token）
        return {
            "code": 200,
            "message": "注册成功",
            "data": {
                "token": token,
                "userInfo": {
                    "id": user.id,
                    "username": user.username,
                    "bio": user.bio,
                    "avatar": user.avatar
                }
            }
        }
    except IntegrityError as e:
        # 处理数据库约束违反错误
        await db.rollback()
        if "username_UNIQUE" in str(e):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="用户名已存在"
            )
        else:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="注册失败，请稍后重试"
            )
    except HTTPException:
        # 重新抛出自定义异常
        raise
    except Exception as e:
        # 处理其他异常
        await db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="注册失败，请稍后重试"
        )

重启FastAPI服务，测试数据是否正常返回

启动前端服务检查数据是否正常加载，注册成功后会自动跳转首页