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

fastapi 博客系统模型分析

article 2025/1/23 10:54:57

博客系统模型分析

依赖版本要求

本项目使用的主要依赖版本：

# requirements.txt
sqlalchemy>=2.0.0
pydantic>=2.0.0
fastapi>=0.100.0

SQLAlchemy Base 与 Pydantic BaseModel 说明

SQLAlchemy Base

  1. 用途：

    • SQLAlchemy ORM的基类，用于定义数据库模型
    • 负责数据库表的映射和关系定义
    • 处理数据库层面的操作（CRUD）

  2. 特点：

    • 继承自declarative_base()
    • 定义数据库表结构和关系
    • 支持数据库迁移
    • 处理数据库事务
    • 管理数据库连接

  3. 示例：

from sqlalchemy.orm import declarative_base
Base = declarative_base()

class UserModel(Base):
    __tablename__ = "users"
    # 数据库字段定义

Pydantic BaseModel

  1. 用途：

    • 数据验证和序列化
    • API请求和响应模型定义
    • 数据模式文档生成

  2. 特点：

    • 运行时数据验证
    • 类型注解支持
    • 自动生成JSON Schema
    • 支持复杂的数据转换
    • 提供错误处理机制

  3. 示例：

from pydantic import BaseModel

class UserCreate(BaseModel):
    username: str
    email: str
    # API请求数据验证

两者的区别和联系

  1. 职责不同：

    • SQLAlchemy Base: 负责数据库层面的对象关系映射
    • Pydantic BaseModel: 负责API层面的数据验证和序列化

  2. 使用场景：

    • SQLAlchemy Base: 数据库操作、表结构定义
    • Pydantic BaseModel: API接口定义、请求响应验证

  3. 数据流转：

# API请求数据 -> Pydantic模型验证 -> SQLAlchemy模型存储 -> 数据库
# 数据库数据 -> SQLAlchemy模型查询 -> Pydantic模型序列化 -> API响应
  1. 配合使用：
# Pydantic模型
class UserCreate(BaseModel):
    username: str
    email: str

# SQLAlchemy模型
class UserModel(Base):
    __tablename__ = "users"
    username: Mapped[str]
    email: Mapped[str]

# 在API中使用
@router.post("/users/")
async def create_user(user: UserCreate, db: Session):
    db_user = UserModel(**user.dict())  # Pydantic -> SQLAlchemy
    db.add(db_user)
    db.commit()
    return user  # SQLAlchemy -> Pydantic

核心模型

1. 用户模型 (UserModel)

users
├── id (主键)
├── username (唯一)
├── email (唯一)
├── hashed_password (密码哈希)
├── is_active (是否激活)
├── is_admin (是否管理员)
└── created_at (创建时间)

模型实现：

class UserModel(Base):
    """用户表SQLAlchemy模型"""
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # 关联关系
    posts = relationship("PostModel", back_populates="user")
    images = relationship("ImageModel", back_populates="user")
    comments = relationship("CommentModel", back_populates="author")

class UserCreate(BaseModel):
    """用户创建Pydantic模型"""
    username: str = Field(..., description="用户名")
    email: EmailStr = Field(..., description="邮箱地址")
    password: constr(min_length=6, max_length=6, pattern=r'^\d{6}$') = Field(
        ...,
        description="密码必须是6位数字"
    )

class UserRead(BaseModel):
    """用户读取Pydantic模型"""
    id: int
    username: str
    email: EmailStr
    is_active: bool
    is_admin: bool
    created_at: datetime

    class Config:
        from_attributes = True

关联关系：

  • 与文章模型(PostModel)的一对多关系 (user.posts)
  • 与图片模型(ImageModel)的一对多关系 (user.images)
  • 与评论模型(CommentModel)的一对多关系 (user.comments)

2. 文章模型 (PostModel)

posts
├── id (主键)
├── title (标题)
├── raw_content (原始内容)
├── html_content (HTML内容)
├── slug (URL别名，唯一)
├── author_id (作者ID，外键 -> users.id)
├── category_id (分类ID，外键 -> categories.id)
├── created_at (创建时间)
└── updated_at (更新时间)

模型实现：

class PostModel(Base):
    """文章表SQLAlchemy模型"""
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    raw_content: Mapped[str] = mapped_column(Text, nullable=False)
    html_content: Mapped[str] = mapped_column(Text, nullable=True)
    slug: Mapped[str] = mapped_column(String(200), unique=True, nullable=True)
    author_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=True)
    category_id: Mapped[int] = mapped_column(Integer, ForeignKey("categories.id"), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 关联关系
    user = relationship("UserModel", back_populates="posts")
    category = relationship("CategoryModel", back_populates="posts")
    tags = relationship("TagModel", secondary="post_tags", back_populates="posts")
    images = relationship("ImageModel", secondary="post_images", back_populates="posts")
    comments = relationship("CommentModel", back_populates="post")

class PostCreate(BaseModel):
    """文章创建Pydantic模型"""
    title: str
    raw_content: str
    html_content: str | None = None
    slug: str | None = None
    category_id: int | None = None

class PostRead(BaseModel):
    """文章读取Pydantic模型"""
    id: int
    title: str
    raw_content: str
    html_content: str | None = None
    slug: str | None = None
    author_id: int
    category_id: int | None = None
    created_at: datetime
    updated_at: datetime
    user: UserRead | None = None
    category: CategoryRead | None = None

    class Config:
        from_attributes = True

关联关系：

  • 与用户模型(UserModel)的多对一关系 (post.user)
  • 与分类模型(CategoryModel)的多对一关系 (post.category)
  • 通过post_tags与标签模型(TagModel)的多对多关系
  • 通过post_images与图片模型(ImageModel)的多对多关系
  • 与评论模型(CommentModel)的一对多关系 (post.comments)

3. 分类模型 (CategoryModel)

categories
├── id (主键)
├── name (分类名，唯一)
└── description (描述)

模型实现：

class CategoryModel(Base):
    """分类表SQLAlchemy模型"""
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    description: Mapped[str] = mapped_column(String(200), nullable=True)

    # 关联关系
    posts = relationship("PostModel", back_populates="category")

class CategoryCreate(BaseModel):
    """分类创建Pydantic模型"""
    name: str
    description: str | None = None

class CategoryRead(BaseModel):
    """分类读取Pydantic模型"""
    id: int
    name: str
    description: str | None = None

    class Config:
        from_attributes = True

关联关系：

  • 与文章模型(PostModel)的一对多关系 (category.posts)

4. 标签模型 (TagModel)

tags
├── id (主键)
└── name (标签名，唯一)

模型实现：

class TagModel(Base):
    """标签表SQLAlchemy模型"""
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)

    # 关联关系
    posts = relationship("PostModel", secondary="post_tags", back_populates="tags")

class TagCreate(BaseModel):
    """标签创建Pydantic模型"""
    name: str

class TagRead(BaseModel):
    """标签读取Pydantic模型"""
    id: int
    name: str

    class Config:
        from_attributes = True

关联关系：

  • 通过post_tags与文章模型(PostModel)的多对多关系

5. 图片模型 (ImageModel)

images
├── id (主键)
├── filename (文件名)
├── original_name (原始文件名)
├── user_id (用户ID，外键 -> users.id)
└── upload_time (上传时间)

模型实现：

class ImageModel(Base):
    """图片表SQLAlchemy模型"""
    __tablename__ = "images"
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    original_name: Mapped[str] = mapped_column(String(255), nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=True)
    upload_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # 关联关系
    user = relationship("UserModel", back_populates="images")
    posts = relationship("PostModel", secondary="post_images", back_populates="images")

class ImageCreate(BaseModel):
    """图片创建Pydantic模型"""
    filename: str
    original_name: str | None = None
    user_id: int | None = None

class ImageRead(BaseModel):
    """图片读取Pydantic模型"""
    id: int
    filename: str
    original_name: str | None = None
    user_id: int | None = None
    upload_time: datetime

    class Config:
        from_attributes = True

关联关系：

  • 与用户模型(UserModel)的多对一关系 (image.user)
  • 通过post_images与文章模型(PostModel)的多对多关系

6. 评论模型 (CommentModel)

comments
├── id (主键)
├── post_id (文章ID，外键 -> posts.id)
├── author_id (作者ID，外键 -> users.id)
├── content (评论内容)
└── created_at (创建时间)

模型实现：

class CommentModel(Base):
    """评论表SQLAlchemy模型"""
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    post_id: Mapped[int] = mapped_column(Integer, ForeignKey("posts.id"))
    author_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
    content: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # 关联关系
    post = relationship("PostModel", back_populates="comments")
    author = relationship("UserModel", back_populates="comments")

class CommentCreate(BaseModel):
    """评论创建Pydantic模型"""
    post_id: int
    content: str

class CommentRead(BaseModel):
    """评论读取Pydantic模型"""
    id: int
    post_id: int
    author_id: int
    content: str
    created_at: datetime

    class Config:
        from_attributes = True

关联关系：

  • 与文章模型(PostModel)的多对一关系 (comment.post)
  • 与用户模型(UserModel)的多对一关系 (comment.author)

7. 网站配置模型 (WebConfigModel)

web_config
├── id (主键)
├── key (配置键，唯一)
├── value (配置值)
├── description (配置描述)
├── created_at (创建时间)
└── updated_at (更新时间)

模型实现：

class WebConfigModel(Base):
    """网站配置表SQLAlchemy模型"""
    __tablename__ = "web_config"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    key: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, comment="配置键")
    value: Mapped[str] = mapped_column(Text, nullable=True, comment="配置值")
    description: Mapped[str] = mapped_column(String(200), nullable=True, comment="配置描述")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, comment="创建时间")
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, 
        default=datetime.utcnow, 
        onupdate=datetime.utcnow,
        comment="更新时间"
    )

class WebConfigCreate(BaseModel):
    """网站配置创建Pydantic模型"""
    key: str
    value: str | None = None
    description: str | None = None

class WebConfigRead(BaseModel):
    """网站配置读取Pydantic模型"""
    id: int
    key: str
    value: str | None = None
    description: str | None = None
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

无其他模型关联关系

关联表

1. 文章标签关联表 (PostTagModel)

post_tags
├── post_id (主键，外键 -> posts.id)
└── tag_id (主键，外键 -> tags.id)

模型实现：

class PostTagModel(Base):
    """文章标签关联表SQLAlchemy模型"""
    __tablename__ = "post_tags"

    post_id: Mapped[int] = mapped_column(Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True)
    tag_id: Mapped[int] = mapped_column(Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True)

连接文章模型(PostModel)和标签模型(TagModel)的多对多关系

2. 文章图片关联表 (PostImageModel)

post_images
├── post_id (主键，外键 -> posts.id)
└── image_id (主键，外键 -> images.id)

模型实现：

class PostImageModel(Base):
    """文章图片关联表SQLAlchemy模型"""
    __tablename__ = "post_images"

    post_id: Mapped[int] = mapped_column(Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True)
    image_id: Mapped[int] = mapped_column(Integer, ForeignKey("images.id", ondelete="CASCADE"), primary_key=True)

连接文章模型(PostModel)和图片模型(ImageModel)的多对多关系

主要功能

  1. 用户管理

    • 用户认证和授权
    • 管理员/普通用户区分
    • 用户活动跟踪

  2. 内容管理

    • 支持HTML内容的博客文章
    • 文章分类和标签组织
    • 图片上传和管理
    • 评论系统

  3. 网站配置

    • 通过WebConfigModel实现动态网站设置
    • 可配置的站点参数

数据库关系概览

用户 ─┬─── 文章 ───┬─── 分类
      │           │
      │           ├─── 标签 (多对多)
      │           │
      ├─── 图片 ─┘    
      │
      └─── 评论

安全特性

  1. 用户账户密码哈希处理
  2. 用户活动状态控制
  3. 管理员权限控制
  4. 关键字段唯一性约束

数据验证

每个模型都有对应的Pydantic模型用于：

  • 创建 (如 UserCreate, PostCreate)
  • 读取 (如 UserRead, PostRead)
  • 更新 (适用时)

这确保了API层面的类型安全和数据验证。

模型使用示例

1. 基础依赖导入

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import select, and_, or_
from typing import List, Optional
from datetime import datetime

from dao.database import get_db
from models import UserModel, PostModel, CategoryModel, TagModel, CommentModel

2. 用户相关操作

# 创建用户
async def create_user(db: Session, user_create: UserCreate) -> UserModel:
    db_user = UserModel(
        username=user_create.username,
        email=user_create.email,
        hashed_password=get_password_hash(user_create.password)
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 获取用户信息
async def get_user(db: Session, user_id: int) -> Optional[UserModel]:
    return db.query(UserModel).filter(UserModel.id == user_id).first()

# 用户登录验证
async def authenticate_user(db: Session, username: str, password: str) -> Optional[UserModel]:
    user = db.query(UserModel).filter(UserModel.username == username).first()
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user

3. 文章相关操作

# 创建文章
async def create_post(
    db: Session, 
    post_create: PostCreate, 
    author_id: int,
    tags: List[int] = None
) -> PostModel:
    db_post = PostModel(
        title=post_create.title,
        raw_content=post_create.raw_content,
        html_content=post_create.html_content,
        slug=post_create.slug,
        category_id=post_create.category_id,
        author_id=author_id
    )
    
    # 添加标签
    if tags:
        db_tags = db.query(TagModel).filter(TagModel.id.in_(tags)).all()
        db_post.tags.extend(db_tags)
    
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

# 获取文章列表（带分页和过滤）
async def get_posts(
    db: Session,
    skip: int = 0,
    limit: int = 10,
    category_id: Optional[int] = None,
    tag_id: Optional[int] = None,
    author_id: Optional[int] = None
) -> List[PostModel]:
    query = select(PostModel)
    
    # 添加过滤条件
    filters = []
    if category_id:
        filters.append(PostModel.category_id == category_id)
    if author_id:
        filters.append(PostModel.author_id == author_id)
    if tag_id:
        query = query.join(PostModel.tags).filter(TagModel.id == tag_id)
    if filters:
        query = query.filter(and_(*filters))
    
    # 添加排序和分页
    query = query.order_by(PostModel.created_at.desc())
    query = query.offset(skip).limit(limit)
    
    return db.scalars(query).all()

# 更新文章
async def update_post(
    db: Session,
    post_id: int,
    post_update: PostCreate,
    current_user_id: int
) -> Optional[PostModel]:
    db_post = db.query(PostModel).filter(
        PostModel.id == post_id,
        PostModel.author_id == current_user_id
    ).first()
    
    if not db_post:
        return None
        
    for key, value in post_update.dict(exclude_unset=True).items():
        setattr(db_post, key, value)
    
    db.commit()
    db.refresh(db_post)
    return db_post

4. 评论相关操作

# 添加评论
async def create_comment(
    db: Session,
    comment_create: CommentCreate,
    author_id: int
) -> CommentModel:
    db_comment = CommentModel(
        post_id=comment_create.post_id,
        author_id=author_id,
        content=comment_create.content
    )
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)
    return db_comment

# 获取文章评论
async def get_post_comments(
    db: Session,
    post_id: int,
    skip: int = 0,
    limit: int = 10
) -> List[CommentModel]:
    return db.query(CommentModel)\
        .filter(CommentModel.post_id == post_id)\
        .order_by(CommentModel.created_at.desc())\
        .offset(skip)\
        .limit(limit)\
        .all()

5. 网站配置操作

# 获取所有配置
async def get_all_configs(db: Session) -> dict:
    configs = db.query(WebConfigModel).all()
    return {config.key: config.value for config in configs}

# 更新配置
async def update_config(
    db: Session,
    key: str,
    value: str,
    current_user: UserModel
) -> WebConfigModel:
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only administrators can modify configurations"
        )
    
    config = db.query(WebConfigModel).filter(WebConfigModel.key == key).first()
    if not config:
        config = WebConfigModel(key=key)
    
    config.value = value
    db.add(config)
    db.commit()
    db.refresh(config)
    return config

6. FastAPI路由示例

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List

router = APIRouter()

# 获取文章列表
@router.get("/posts/", response_model=List[PostRead])
async def read_posts(
    skip: int = 0,
    limit: int = 10,
    category_id: Optional[int] = None,
    tag_id: Optional[int] = None,
    db: Session = Depends(get_db)
):
    posts = await get_posts(db, skip, limit, category_id, tag_id)
    return posts

# 创建新文章
@router.post("/posts/", response_model=PostRead)
async def create_new_post(
    post: PostCreate,
    current_user: UserModel = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    return await create_post(db, post, current_user.id)

# 添加评论
@router.post("/comments/", response_model=CommentRead)
async def create_new_comment(
    comment: CommentCreate,
    current_user: UserModel = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    return await create_comment(db, comment, current_user.id)

7. 前端调用示例

// 获取文章列表
async function fetchPosts(page = 1, category = null, tag = null) {
    const limit = 10;
    const skip = (page - 1) * limit;
    let url = `/api/posts/?skip=${skip}&limit=${limit}`;
    
    if (category) url += `&category_id=${category}`;
    if (tag) url += `&tag_id=${tag}`;
    
    try {
        const response = await fetch(url);
        if (!response.ok) throw new Error('Failed to fetch posts');
        const posts = await response.json();
        return posts;
    } catch (error) {
        console.error('Error fetching posts:', error);
        throw error;
    }
}

// 创建新文章
async function createPost(postData) {
    try {
        const response = await fetch('/api/posts/', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': `Bearer ${localStorage.getItem('token')}`
            },
            body: JSON.stringify(postData)
        });
        
        if (!response.ok) throw new Error('Failed to create post');
        return await response.json();
    } catch (error) {
        console.error('Error creating post:', error);
        throw error;
    }
}

// 添加评论
async function addComment(postId, content) {
    try {
        const response = await fetch('/api/comments/', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': `Bearer ${localStorage.getItem('token')}`
            },
            body: JSON.stringify({
                post_id: postId,
                content: content
            })
        });
        
        if (!response.ok) throw new Error('Failed to add comment');
        return await response.json();
    } catch (error) {
        console.error('Error adding comment:', error);
        throw error;
    }
}

http://www.kler.cn/a/514795.html

相关文章：

  • 数据库事务详解
  • MyBatis 注解开发详解
  • simulink入门学习01
  • Web安全攻防入门教程——hvv行动详解
  • 基于 WPF 平台使用纯 C# 实现动态处理 json 字符串
  • 实施工程师：面试基础宝典
  • 考研408笔记之数据结构（六）——查找
  • go语言gui窗口应用之fyne框架-动态添加、删除一行控件（逐行注释）
  • Django的models.model如何使用
  • LoRA面试篇
  • AIGC浪潮下，图文内容社区数据指标体系如何构建？
  • nodeJS 系统学习（package-包-章节2）
  • 2025牛客寒假算法营1
  • C++并发编程之线程中断异常的捕捉与信息显示
  • Groovy语言的安全开发
  • PAT甲级-1014 Waiting in Line
  • 【软件】解决奥林巴斯生物显微镜软件OlyVIA提示“不支持您使用的操作系统”安装中止的问题
  • 【思科】NAT配置
  • macos app签名和公证
  • PHP教育系统小程序
  • Python网络自动化运维---用户交互模块
  • Vue3组件重构实战：从Geeker-Admin拆解DataTable的最佳实践
  • 场馆预定平台高并发时间段预定实现V2
  • 计算机组成原理(计算机系统3)--实验七：新增指令实验
  • [操作系统] 环境变量详解
  • vue项目动态div滚动条滑动到指定位置效果