feat:增加FastAPI文档

docs/FastAPI/FastAPIGuide.md

@@ -1,2 +1,586 @@
-# 简介
+# 一、简介
   [FastAPI](https://fastapi.tiangolo.com/zh/) 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 并基于标准的 Python 类型提示。   
+  安装： 
+```cmd
+pip install fastapi
+pip install uvicorn[standard]  # ASGI服务器
+```
+
+# 二、数据层
+## 2.1 数据库ORM SQLAlchemy
+### 2.1.1 配置与连接
+```python
+# database.py - 数据库配置
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, Session
+from typing import Generator
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# 数据库配置
+DATABASE_URL = f"mysql+pymysql://{os.getenv("DATABASE_URL", "root:123456@127.0.0.1:3306/test")}"
+
+# 创建引擎
+engine = create_engine(
+    DATABASE_URL,
+    echo=True,  # 显示 SQL 语句（开发环境）
+    pool_size=20,  # 连接池大小
+    max_overflow=40,  # 最大溢出连接数
+    pool_pre_ping=True,  # 连接前 ping
+    pool_recycle=3600,  # 连接回收时间（秒）
+)
+
+# 创建会话工厂
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+    expire_on_commit=False,  # 提交后不使实例过期
+)
+
+# 声明基类
+Base = declarative_base()
+
+# 依赖注入：获取数据库会话
+def get_db() -> Generator[Session, None, None]:
+    """
+    获取数据库会话
+    使用 yield 确保会话正确关闭
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+```
+
+>[!WARNING]
+>需要把pool_pre_ping=True打开，否则会出现超过pool_recycle时间后，数据库断线。  
+
+### 2.1.2 数据库基类
+```python
+from sqlalchemy import Column, BigInteger, String, DateTime, event
+from sqlalchemy.ext.declarative import declared_attr
+
+from config.auth import context_sub
+from config.database import Base
+from datetime import datetime
+
+from utils.common import camel_to_snake
+from id_generator import options, generator
+
+# https://github.com/yitter/IdGenerator/tree/master/Python
+options = options.IdGeneratorOptions(worker_id=23)
+idgen = generator.DefaultIdGenerator()
+idgen.set_id_generator(options)
+
+
+# 第二层基类：包含ID
+class IdBase(Base):
+    __abstract__ = True
+
+    id = Column(BigInteger, primary_key=True, index=True)
+
+    @declared_attr
+    def __tablename__(cls):
+        # 自动把数据库实体类名驼峰转为数据库表名下划线
+        return camel_to_snake(cls.__name__)
+
+
+# 自动填充id
+@event.listens_for(IdBase, 'before_insert', propagate=True)
+def before_insert_listener(mapper, connection, target):
+    if target.id is None:
+        target.id = idgen.next_id()
+
+
+# 第二层基类：包含ID和审计字段
+class AuditBase(IdBase):
+    __abstract__ = True
+
+    create_time = Column(DateTime, nullable=True, default=datetime.now)
+    create_by = Column(String(255), nullable=True)
+    update_time = Column(DateTime, nullable=True, default=datetime.now, onupdate=datetime.now)
+    update_by = Column(String(255), nullable=True)
+
+
+@event.listens_for(AuditBase, 'before_insert', propagate=True)
+def before_insert(mapper, connection, target):
+    value = context_sub.get(None)
+    if value is not None:
+        target.create_by = value
+
+
+@event.listens_for(AuditBase, 'before_update', propagate=True)
+def before_update(mapper, connection, target):
+    value = context_sub.get(None)
+    if value is not None:
+        target.update_by = value
+```
+
+  id采用yitter雪花id。   
+  审计字段中的create_time和update_time通过SQLAlchemy参数配置。  
+  create_by和update_by字段为登录认证中存储的账号信息。   
+
+### 2.1.3 数据库模型
+```python
+from sqlalchemy import (BigInteger, Boolean, Column, Integer, DECIMAL, String, LargeBinary)
+from sqlalchemy.orm import relationship
+
+from models.base import AuditBase, IdBase
+
+
+class Blog(AuditBase):
+    title = Column(String(255), nullable=False, comment="博客标题")
+    top_value = Column(Integer, nullable=False, default=0, comment="置顶值 越大越靠前")
+    is_great = Column(Boolean, nullable=False, default=False, comment="是否是精品")
+    category_id = Column(BigInteger, nullable=False, comment="博客类别")
+    summary = Column(String(255), nullable=False, comment="博客内容概要")
+    content_id = Column(BigInteger, nullable=False, comment="博客内容")
+    word_count = Column(Integer, nullable=False, default=0, comment="字数统计")
+    read_duration = Column(DECIMAL(10, 2), nullable=False, default=0.00, comment="阅读时长")
+    is_approved = Column(Boolean, nullable=False, default=False, comment="是否发布")
+
+    # 分类关系（多对一）
+    category = relationship(
+        "BlogCategory",
+        # 与BlogCategory的blogs属性建立双向关系
+        back_populates="blogs",
+        # 只级联保存和合并操作，不级联删除（删除博客不应删除分类
+        cascade="save-update, merge",
+        # 明确指定连接条件
+        # 如果数据库设置了外键可以省略
+        primaryjoin="foreign(Blog.category_id) == BlogCategory.id"
+    )
+
+    # 内容关系（一对一）
+    content = relationship(
+        "BlogContent",
+        # 与BlogContent的blog属性建立双向关系
+        back_populates="blog",
+        # 完全级联操作：保存、合并、刷新、删除等所有操作都会级联
+        cascade="all, delete-orphan",
+        # 设置为False表示一对一关系，返回单个对象而不是列表
+        uselist=False,
+        # 确保内容只有一个父博客，与delete-orphan配合使用
+        single_parent=True,
+        # 明确指定连接条件
+        primaryjoin="foreign(Blog.content_id) == BlogContent.id"
+    )
+
+    # 评论（一对多）
+    comments = relationship(
+        "BlogComment",
+        # 与BlogComment的blog属性建立双向关系
+        back_populates="blog",
+        # 完全级联操作：博客删除时自动删除所有评论
+        cascade="all, delete-orphan",
+        # 明确指定连接条件
+        primaryjoin="Blog.id == foreign(BlogComment.blog_id)"
+    )
+
+
+class BlogCategory(AuditBase):
+    name = Column(String(45), nullable=False, comment="类别名称")
+
+    blogs = relationship(
+        "Blog",
+        # 与Blog的category属性建立双向关系
+        back_populates="category",
+        # 完全级联操作：分类删除时自动删除所有关联的博客
+        # 警告：这会级联删除分类下的所有博客，包括博客的内容、访问记录和评论
+        cascade="all, delete-orphan",
+        # 明确指定连接条件
+        primaryjoin="BlogCategory.id == foreign(Blog.category_id)"
+    )
+
+
+class BlogContent(IdBase):
+    content = Column(LargeBinary, nullable=False, comment="博客内容")
+
+    blog = relationship(
+        "Blog",
+        # 与Blog的content属性建立双向关系
+        back_populates="content",
+        # 设置为False表示一对一关系
+        uselist=False,
+        # 明确指定连接条件
+        primaryjoin="BlogContent.id == foreign(Blog.content_id)"
+    )
+
+
+class BlogComment(AuditBase):
+    blog_id = Column(BigInteger, nullable=False, comment="博客ID")
+    parent_id = Column(BigInteger, nullable=False, comment="父评论ID")
+    name = Column(String(255), nullable=False, comment="评论人昵称")
+    website = Column(String(255), nullable=True, comment="评论人网站")
+    ip_address = Column(String(45), nullable=False, comment="评论人IP")
+    user_agent = Column(String(255), nullable=False, comment="评论人浏览器信息")
+    content = Column(String(255), nullable=False, comment="评论内容")
+    is_approved = Column(Boolean, nullable=False, default=False, comment="是否通过")
+
+    blog = relationship(
+        "Blog",
+        # 与Blog的comments属性建立双向关系
+        back_populates="comments",
+        # 明确指定连接条件
+        primaryjoin="foreign(BlogComment.blog_id) == Blog.id"
+    )
+```
+
+  这里没有通过在数据库建立外键，而是通过SQLAlchemy中的relationship来建立。    
+  其中primaryjoin明确指定了连接条件。如果为一对一，需要设置uselist=False和single_parent=True，返回单个对象而不是列表。   
+  cascade表示级联操作：  
+```
+# cascade 的完整可选值列表：
+
+# 基本级联操作
+"save-update"      # 保存/更新时级联
+"merge"           # 合并会话时级联
+"refresh-expire"  # 刷新过期对象时级联
+"expunge"         # 从会话中移除时级联
+"delete"          # 删除时级联
+"delete-orphan"   # 成为孤儿时删除
+
+# 快捷组合
+"all"                     # 包含除 delete-orphan 外的所有操作
+"all, delete-orphan"      # 包含所有操作
+"none"                    # 禁用所有级联（默认）
+
+# 其他组合
+"save-update, merge"               # 常用组合
+"save-update, merge, delete"       # 包含删除
+"save-update, merge, refresh-expire"
+```
+
+## 2.2 数据验证pydantic
+```python
+from typing import Optional
+
+from pydantic import BaseModel, Field, field_validator
+from datetime import datetime
+
+
+class BlogQuery(BaseModel):
+    """博客查询参数"""
+    category: Optional[str] = Field(None, description="分类名称")
+    title: Optional[str] = Field(None, description="标题关键词")
+    year: Optional[int] = Field(None, description="发布年份", ge=2000, le=datetime.now().year)
+
+    @field_validator('year')
+    def validate_year(cls, v):
+        if v is not None and v > datetime.now().year:
+            raise ValueError('年份不能超过当前年份')
+        return v
+
+
+class BlogBase(BaseModel):
+    """博客基础模型"""
+    title: str = Field(..., min_length=1, max_length=255, description="博客标题")
+    top_value: int = Field(default=0, ge=0, description="置顶值，越大越靠前", alias="topValue")
+    is_great: bool = Field(default=False, description="是否是精品", alias="isGreat")
+    category: str = Field(..., min_length=1, max_length=45, description="分类名称")
+    content: Optional[str] = Field(None, description="博客内容")
+    is_approved: bool = Field(default=False, description="是否已发布", alias="isApproved")
+
+    @field_validator('title')
+    def title_not_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError('标题不能为空')
+        return v.strip()
+
+    @field_validator('category')
+    def category_not_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError('分类不能为空')
+        return v.strip()
+
+
+class BlogCreate(BlogBase):
+    pass
+
+
+class BlogUpdate(BlogBase):
+    pass
+
+
+class BlogResponse(BlogBase):
+    id: int = Field(..., description="博客ID")
+    summary: Optional[str] = Field(None, description="内容摘要")
+    word_count: Optional[int] = Field(None, description="字数统计", alias="wordCount")
+    read_duration: Optional[float] = Field(None, description="阅读时长", alias="readDuration")
+    visit_count: Optional[int] = Field(0, description="访问次数", alias="visitCount")
+    create_time: datetime = Field(..., description="创建时间", alias="createTime")
+    update_time: datetime = Field(..., description="更新时间", alias="updateTime")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+        json_encoders = {
+            # 自定义 datetime 类型的序列化格式
+            datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S')
+        }
+
+
+class BlogCategoryResponse(BaseModel):
+    name: str = Field(..., description="分类名称")
+    count: int = Field(..., description="博客数量")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+
+
+class BlogStatsResponse(BaseModel):
+    blog_count: int = Field(..., description="博客总数", alias="blogCount")
+    category_count: int = Field(..., description="分类总数", alias="categoryCount")
+    word_count: int = Field(..., description="总字数", alias="wordCount")
+
+
+class BlogVisitResponse(BaseModel):
+    ip: str = Field(..., description="IP地址")
+    os: str = Field(..., description="操作系统")
+    browser: str = Field(..., description="浏览器")
+    uri: str = Field(..., description="访问路径")
+    title: str = Field(None, description="博客标题")
+    visit_time: datetime = Field(..., description="访问时间", alias="visitTime")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+        json_encoders = {
+            datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S')
+        }
+
+
+class BlogLatestResponse(BaseModel):
+    id: int = Field(..., description="博客ID")
+    title: str = Field(..., description="博客标题")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+
+
+class BlogAdjacentResponse(BaseModel):
+    id: int = Field(..., description="博客ID")
+    title: str = Field(..., description="博客标题")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+
+
+class BlogCommentCreate(BaseModel):
+    parent_id: int = Field(default=0, ge=0, description="父评论ID，0表示顶级评论", alias="parentId")
+    name: str = Field(..., min_length=1, max_length=50, description="评论人昵称")
+    website: Optional[str] = Field(None, description="评论人网站")
+    content: str = Field(..., min_length=1, max_length=1000, description="评论内容")
+
+    @field_validator('name')
+    def name_not_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError('昵称不能为空')
+        return v.strip()
+
+    @field_validator('content')
+    def content_not_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError('评论内容不能为空')
+        return v.strip()
+
+
+class BlogCommentResponse(BlogCommentCreate):
+    id: int = Field(..., description="评论ID")
+    ip_address: str = Field(..., description="IP地址", alias="ipAddress")
+    user_agent: str = Field(..., description="浏览器信息", alias="userAgent")
+    create_time: datetime = Field(..., description="创建时间", alias="createTime")
+
+    class Config:
+        from_attributes = True
+        populate_by_name = True
+        json_encoders = {
+            # 自定义 datetime 类型的序列化格式
+            datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S')
+        }
+
+```
+
+  `@field_validator`为Pydantic V2的验证器装饰器。   
+  `class Config`为全局配置类，`from_attributes=True`表示允许从SQLAlchemy等ORM对象创建。  
+
+# 三、服务层
+## 3.1 查询操作
+```python
+def query_blog_by_id(db: Session, blog_id: int) -> BlogResponse:
+    check_blog_exist(db, blog_id)
+
+    stmt = (select(
+        Blog.id,
+        Blog.title,
+        Blog.top_value,
+        Blog.is_great,
+        BlogCategory.name.label("category"),
+        Blog.summary,
+        BlogContent.content,
+        Blog.word_count,
+        Blog.read_duration,
+        func.count(BlogVisit.id).label("visitCount"),
+        Blog.is_approved,
+        Blog.create_time,
+        Blog.update_time
+    ).where(Blog.id == blog_id)
+            .outerjoin(BlogCategory, Blog.category_id == BlogCategory.id)
+            .outerjoin(BlogContent, Blog.content_id == BlogContent.id)
+            .outerjoin(BlogVisit, Blog.id == BlogVisit.blog_id))
+
+    blog = db.execute(stmt).first()
+
+    return BlogResponse.model_validate(blog)
+```
+
+  条件和分页查询
+```python
+def query_blog_by_condition(db: Session, blog_query: BlogQuery) -> List[BlogResponse]:
+    stmt = get_query_blog_by_condition_stmt(blog_query)
+    results = db.execute(stmt).fetchall()
+    return [BlogResponse.model_validate(result) for result in results]
+
+
+def get_query_blog_by_condition_stmt(blog_query: BlogQuery):
+    stmt = (select(
+        Blog.id,
+        Blog.title,
+        Blog.top_value,
+        Blog.is_great,
+        BlogCategory.name.label("category"),
+        Blog.summary,
+        BlogContent.content,
+        Blog.word_count,
+        Blog.read_duration,
+        func.count(BlogVisit.id).label("visitCount"),
+        Blog.is_approved,
+        Blog.create_time,
+        Blog.update_time
+    ).where(Blog.is_approved == 1)
+            .outerjoin(BlogCategory, Blog.category_id == BlogCategory.id)
+            .outerjoin(BlogContent, Blog.content_id == BlogContent.id)
+            .outerjoin(BlogVisit, Blog.id == BlogVisit.blog_id)
+            .group_by(Blog.id))
+
+    conditions = []
+
+    if blog_query.category:
+        conditions.append(BlogCategory.name == blog_query.category)
+
+    if blog_query.title:
+        conditions.append(Blog.title.like(f"%{blog_query.title}%"))
+
+    if blog_query.year:
+        conditions.append(func.extract('year', Blog.create_time) == blog_query.year)
+
+    if conditions:
+        stmt = stmt.where(and_(*conditions))
+
+    stmt = stmt.order_by(desc(Blog.is_great), desc(Blog.update_time))
+
+    return stmt
+
+
+def query_blog_by_condition_page(db: Session, blog_query: BlogQuery,
+                                 current_page: int = 1, page_size: int = 10) -> PageResult[BlogResponse]:
+    stmt = get_query_blog_by_condition_stmt(blog_query)
+    return paginate_query(db, stmt, current_page, page_size)
+
+def paginate_query(db: Session, query, current_page: int = 1, page_size: int = 10) -> PageResult:
+    """通用分页查询函数"""
+    # 计算总记录数
+    total = db.execute(select(func.count()).select_from(query.subquery())).scalar_one_or_none() or 0
+
+    # 计算总页数
+    total_pages = (total + page_size - 1) // page_size if page_size != 0 else 0
+
+    # 执行分页查询
+    results = db.execute(query.offset((current_page - 1) * page_size).limit(page_size)).all()
+
+    # 转换为字典列表
+    records = [row._asdict() if hasattr(row, "_asdict") else dict(row) for row in results]
+
+    return PageResult(
+        current=current_page,
+        size=page_size,
+        total=total,
+        pages=total_pages,
+        records=records
+    )
+```
+
+## 3.2 新增操作
+```python
+def add_blog(db: Session, blog: BlogCreate) -> bool:
+    word_count = get_word_count(blog.content)
+
+    db_blog = Blog(
+        title=blog.title,
+        top_value=blog.top_value,
+        is_great=blog.is_great,
+        category_id=add_blog_category(db, blog.category),
+        summary=get_blog_summary(blog.content),
+        word_count=word_count,
+        read_duration=get_read_duration(word_count),
+        is_approved=blog.is_approved
+    )
+
+    # 级联新增
+    db_blog.content = BlogContent(content=blog.content.encode('utf-8'))
+
+    db.add(db_blog)
+    db.commit()
+    db.refresh(db_blog)
+
+    return True
+```
+
+  这里的`db_blog.content`可以直接赋值`BlogContent`对象实现级联新增，不用传`content_id`。    
+>[!WARNING]
+>这里的category和blog是多对一的关系，且需要一定的逻辑处理，不能直接赋值`BlogCategory`对象，否则每新增一条博客都会新增一个`category`。   
+
+## 3.3 更新操作
+```python
+def update_blog(db: Session, blog_id: int, blog: BlogUpdate) -> bool:
+    db_blog = check_blog_exist(db, blog_id)
+
+    db_blog.title = blog.title
+    db_blog.top_value = blog.topValue
+    db_blog.is_great = blog.isGreat
+    db_blog.category_id = update_blog_category(db, blog.category)
+
+    # 级联更新
+    db_blog.content = BlogContent(content=blog.content.encode('utf-8'))
+
+    word_count = get_word_count(blog.content)
+    db_blog.summary = get_blog_summary(blog.content),
+    db_blog.word_count = word_count
+    db_blog.read_duration = get_read_duration(word_count)
+    db_blog.is_approved = blog.isApproved
+
+    db.commit()
+    db.refresh(db_blog)
+
+    return True
+```
+
+## 3.4 删除操作
+```python
+def delete_blog(db: Session, blog_id: int) -> bool:
+    db_blog = check_blog_exist(db, blog_id)
+
+    # 级联删除
+    db.delete(db_blog)
+    db.commit()
+
+    return True
+```