当前位置：首页 > news >正文

FastAPI实战（第二部分）：用户注册接口开发详解

news 2026/5/12 0:00:24

一、项目结构解析

这是一个典型的 FastAPI 项目，采用分层架构，核心目录如下：

toutiao_backend/

├── config/ # 配置层

│ └── db_conf.py # 数据库配置（异步引擎、会话、依赖项）

├── crud/ # 业务逻辑层（CRUD 操作）

├── models/ # 数据模型层（SQLAlchemy ORM）

├── routers/ # 路由层（API 接口）

│ ├── news.py

│ └── users.py # 用户相关接口（当前编辑文件）

├── schemas/ # 数据校验层（Pydantic 模型）

├── utils/ # 工具函数层

├── main.py # 项目入口（FastAPI 实例、路由注册）

└── venv/ # 虚拟环境

二、核心代码结构化输出

1. 数据库配置（config/db_conf.py）

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from sqlalchemy.orm import DeclarativeBase

# 异步数据库连接 URL（示例，需根据实际环境修改）

ASYNC_DATABASE_URL = "mysql+aiomysql://root:password@localhost:3306/toutiao_backend?charset=utf8"

# 创建异步引擎

async_engine = create_async_engine(

ASYNC_DATABASE_URL,

echo=True,

pool_size=10,

max_overflow=20

)

# 创建异步会话工厂

AsyncSessionLocal = async_sessionmaker(

bind=async_engine,

class_=AsyncSession,

expire_on_commit=False

)

# 定义 ORM 模型基类

class Base(DeclarativeBase):

pass

# 数据库会话依赖项

async def get_db():

async with AsyncSessionLocal() as session:

try:

yield session

await session.commit()

except Exception:

await session.rollback()

raise

finally:

await session.close()

2. 用户路由（routers/users.py）

from fastapi import APIRouter, Depends

from sqlalchemy.ext.asyncio import AsyncSession

from config.db_conf import get_db

# 定义 APIRouter 实例

router = APIRouter(

prefix="/api/user", # 路由前缀

tags=["users"] # 接口文档标签

)

# 用户注册接口

@router.post("/register")

async def register(

db: AsyncSession = Depends(get_db) # 注入数据库会话

# 模拟返回注册成功的响应

return {

"code": 200,

"message": "注册成功",

"data": {

"token": "用户访问令牌",

"userInfo": {

"id": 1,

"username": "example_user",

"bio": "这个人很懒，什么都没留下",

"avatar": "https://fastly.jsdelivr.net/npm/@vant/assets/cat.jpeg"

}

3. 项目入口（main.py）

from fastapi import FastAPI

from routers import users, news # 导入路由模块

# 创建 FastAPI 实例

app = FastAPI(title="头条后端 API", version="1.0")

# 注册子路由

app.include_router(users.router)

app.include_router(news.router)

# 启动事件：自动创建数据库表

@app.on_event("startup")

async def startup_event():

from config.db_conf import async_engine, Base

async with async_engine.begin() as conn:

await conn.run_sync(Base.metadata.create_all)

三、核心代码实现如下：

1. 在model/user.py文件中，建立模型类User，通过用户表ORM模型实现：

from sqlalchemy import String, Integer, DateTime, Text, Boolean from sqlalchemy.orm import Mapped, mapped_column from datetime import datetime from config.db_conf import Base # 导入项目定义的 ORM 基类 class User(Base): """用户表 ORM 模型，映射数据库 user 表""" __tablename__ = "user" # 数据库表名（必须与实际表名一致） # 核心字段定义（Mapped 类型注解 + mapped_column 字段属性） 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(128), nullable=False, comment="加密后的密码（不可明文存储）" ) email: Mapped[str] = mapped_column( String(100), unique=True, nullable=False, comment="邮箱（唯一，用于登录/找回密码）" ) phone: Mapped[str | None] = mapped_column( String(20), unique=True, nullable=True, comment="手机号（可选，唯一）" ) avatar: Mapped[str | None] = mapped_column( String(255), nullable=True, default="https://fastly.jsdelivr.net/npm/@vant/assets/cat.jpeg", comment="用户头像URL" ) bio: Mapped[str | None] = mapped_column( Text, nullable=True, default="这个人很懒，什么都没留下", comment="个人简介" ) is_active: Mapped[bool] = mapped_column( Boolean, default=True, comment="账号是否激活（True=正常，False=禁用）" ) is_admin: Mapped[bool] = mapped_column( Boolean, default=False, comment="是否为管理员（True=管理员，False=普通用户）" ) create_time: Mapped[datetime] = mapped_column( DateTime, default=datetime.now, comment="账号创建时间" ) update_time: Mapped[datetime] = mapped_column( DateTime, default=datetime.now, onupdate=datetime.now, comment="账号更新时间" ) def __repr__(self): """实例打印格式化（便于调试）""" return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

2. 在crud/users.py中，建立两个函数：一是根据用户名查数据库；二是创建用户；

根据用户名查询数据库 async def get_user_by_username (db: AsyncSession, username: str) : 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: UserRequest): #先密码加密处理>add hashed_password = security.get_hash_password (user_data.password)user = User(username=user_data.username, password=hashed_password)db.add(user) await db.commit() await db.refresh(user) #从数据库读回最新的 user returnuser

3. 在routers/users.py文件中，编写register函数，实现创建用户和检查用户是否存在；

@router.post("/register") async def register(user data: UserRequest, db: AsyncSession = Depends(get_db)) : # 用户信息注册逻辑:验证用户是否存在->创建用户 >生成Token >响应结果 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="用户已存在") User = await users.create_user(db, User_data) return { "code":200 'message":"注册成功"， 'data":{ "token":"用户访问令牌”， "userInfo": { "id": user.id, "username": user.Username, "bio":user.bio, "avatar": [https://fastly.jsdeliyr.net/npm/@vant/assets/cat.jpeg" } } }

4. 在crud/users.py，创建create_token函数，生成访问令牌token，具体流程如下：

#生成Token asyncdef create_token(db: AsyncSession, user_id: int): #生成Token+设置过期时间查询数据库当前用户是否有Token有:更新;没有:添加# token = str(uuid.uuid4()) # timedelta(days=7, hours=2, minutes=30, seconds=10)expires_at = datetime.now() + timedelta(days=7)query = select(UserToken).where(UserToken.user_id == user_id)result = await db.execute(query)user_token = result.scalar_one_or_none() if user_token: user_token.token = token user_token.expires_at = expires_at else: user_token = UserToken(user_id=user_id, token=token,expires_at=expires_at) db.add(user_token) await db.commit() return token

5. 在routers/users.py文件中，编写register的create_token函数；

@router.post("/register") async def register(user _data: UserRequest, db: AsyncSession = Depends(get_db)): # 用户信息 和 db #注册逻辑:验证用户是否存在-> 创建用户 >生成Token >响应结果existing_user = await users.get_user_by_username(db, user_data.Username)if existing_user: HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户已存在")raise User = await Users.create_user(db, User_data) token = await users.create_token(db, user.id) return { "code": 200 "message":"注册成功"， "data":{ "token": token, "userInfo": { "id": user.id, "username": user.username "bio": user.bio, 'avatar": user.avatar }}}

6. 封装UserInfoResponse类；UserAuthResponse中包含UserInfoResponse类，

class UserRequest(BaseModel): username:str password:str #user_info 对应的类:基础类+Info类(id、用户名)# #userinfo第一个数据类型定义： class UserInfoBase(BaseModel): #用户信息基础数据模型 nickname:Qpt onal[str] = Field(None, max_length=50, description="昵称")avatar: Qeti0al[str] = Field(None, max_length=255, description="头像URL")gender: Qptional[str] = Field(None, max_length=10, description="性别!")bio: Qptional[str] = Field(None, max_length=500, description="个人简介") #userinfo第二个数据类型定义： class UserInfoResponse(UserInfoBase): id: int username: str # 模型类配置 model_config = ConfigDict( from_attributes=True # 允许从 ORM 对象属性中取值 #data数据类型要求如下： class UserAuthResponse(BaseModel): token: str User_info: UserInfoResponse = Field(..., alias="userInfo") #模型类配置 model_config = ConfigDict( populate_by_name=True,#alias/字段名兼容from_attributes }

7. 重新写register的返回数据类型response_data，调用封装好的类 UserAuthResponse。

@router.post("/register") async def register(user _data: UserRequest, db: AsyncSession = Depends(get_db)): # 用户信息 和 db #注册逻辑:验证用户是否存在-> 创建用户 >生成Token >响应结果existing_user = await users.get_user_by_username(db, user_data.Username)if existing_user: HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户已存在")raise User = await Users.create_user(db, User_data) token = await users.create_token(db, user.id) return { "code": 200 "message":"注册成功"， response_data = UserAuthResponse(token=token, user_info=UseripfoResponse.model validate(user)) return success_response(message="注册成功", data=response_data)