# 一、简介   [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, ConfigDict 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") model_config = ConfigDict( from_attributes=True, populate_by_name=True, json_encoders={ datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S') } ) class BlogCategoryResponse(BaseModel): name: str = Field(..., description="分类名称") count: int = Field(..., description="博客数量") model_config = ConfigDict( 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") model_config = ConfigDict( 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="博客标题") model_config = ConfigDict( from_attributes=True, populate_by_name=True ) class BlogAdjacentResponse(BaseModel): id: int = Field(..., description="博客ID") title: str = Field(..., description="博客标题") model_config = ConfigDict( 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") model_config = ConfigDict( from_attributes=True, populate_by_name=True, json_encoders={ datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S') } ) ```   `@field_validator`为Pydantic V2的验证器装饰器。   `ConfigDict`为全局配置类,`from_attributes=True`表示允许从SQLAlchemy等ORM对象创建。`populate_by_name=True`表示允许通过字段的别名(alias)来赋值 `json_encoders`表示自定义特定类型的 JSON 序列化方式。 # 三、服务层 ## 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 ``` # 四、路由层 ```python router = APIRouter( prefix="/blog", tags=["博客管理"], responses={404: {"description": "Not found"}} ) @router.get("/page", summary="分页查询博客", response_model=PageResult[BlogResponse]) def query_blog_by_page( current_page: int = Query(1, ge=1, alias="currentPage", description="当前页码,从1开始"), page_size: int = Query(10, ge=1, le=100, alias="pageSize", description="每页显示数量,1-100之间"), db: Session = Depends(get_db) ): return blog_service.query_blog_by_page(db, current_page, page_size) @router.get("/condition", summary="条件查询博客", response_model=List[BlogResponse]) def query_blog_by_condition(query: BlogQuery = Depends(), db: Session = Depends(get_db)): return blog_service.query_blog_by_condition(db, query) @router.post("", summary="新增博客内容", response_model=bool) def add_blog( blog: BlogCreate = Body(..., description="博客创建数据"), db: Session = Depends(get_db), _=Depends(verify_token) ): return blog_service.add_blog(db, blog) @router.put("/{blog_id}", summary="更新博客内容", response_model=bool) def update_blog( blog_id: int = Path(..., ge=1, description="博客ID"), blog: BlogUpdate = Body(..., description="博客更新数据"), db: Session = Depends(get_db), _=Depends(verify_token) ): return blog_service.update_blog(db, blog_id, blog) @router.delete("/{blog_id}", summary="删除博客内容", response_model=bool) def delete_blog( blog_id: int = Path(..., ge=1, description="博客ID"), db: Session = Depends(get_db), _=Depends(verify_token) ): return blog_service.delete_blog(db, blog_id) @router.put("/{blog_id}/comment", summary="新增博客评论", response_model=bool) def add_blog_comment( request: Request, blog_id: int = Path(..., ge=1, description="博客ID"), blog_comment: BlogCommentCreate = Body(..., description="博客评论数据"), db: Session = Depends(get_db) ): return blog_service.add_blog_comment(db, request, blog_id, blog_comment) ```   路由装饰器参数中的`response_model`表示定义接口返回的数据模型。   参数注解中的`Query()`表示参数来自URL查询字符串,`Path/Query/Body`分别对应路径参数、查询参数、请求体参数。   `Depends()`表示依赖注入,自动解析参数或执行依赖函数。   `db: Session = Depends(get_db)`表示获取数据库连接,`_=Depends(verify_token)`表示验证用户身份。   `request: Request`可以获取到HTTP请求的完整上下文信息。 ::: tip 这里的`query: BlogQuery = Depends()`会从查询参数中自动实例化`BlogQuery`对象,并进行数据验证和类型转换。 :::   注册路由: ```python from fastapi import FastAPI from .blog import router as blog_router def register_routers(app: FastAPI): app.include_router(blog_router, prefix="") # main.py app = FastAPI(title="Blog Service") register_routers(app) ``` # 五、中间件 ## 5.1 全局异常处理器 ```python from fastapi import Request, HTTPException, status from fastapi.responses import JSONResponse from sqlalchemy.exc import SQLAlchemyError from config.logging import logger # 自定义异常类 class AppException(Exception): def __init__(self, message: str, details=None): self.message = message self.details = details # 全局异常处理中间件 async def global_exception_handler(request: Request, call_next): try: # 记录请求信息(可选) logger.info(f"请求: {request.method} {request.url}") if request.query_params: logger.info(f"查询参数: {dict(request.query_params)}") response = await call_next(request) # 记录响应信息(可选) if response.status_code >= 400: logger.warning(f"响应: {response.status_code}") return response except AppException as e: # 记录业务异常 logger.error(f"业务异常: {e.message} - 详情: {e.details}") return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content={"message": e.message, "details": e.details}) except SQLAlchemyError as e: # 记录数据库异常 logger.critical(f"数据库异常: {str(e)}") return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"message": "数据库操作失败", "details": str(e)}) except Exception as e: # 记录未知异常(带堆栈信息) logger.critical(f"未知异常: {str(e)}") return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"code": 500, "message": "服务器内部错误", "details": str(e)}) def get_credentials_exception() -> HTTPException: return HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) ``` ## 5.2 日志处理器 ```python import atexit import sys from fluent import sender from loguru import logger from config.setting import settings FLUENTD_HOST = settings.FLUENTD_HOST FLUENTD_PORT = 24224 TOPIC_TAG = 'blog-service' # 日志级别 LOG_LEVEL = settings.LOG_LEVEL.upper() # 日志格式 STDOUT_FORMAT = ( "{time:YYYY-MM-DD HH:mm:ss.SSS} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) FILE_FORMAT = ( "{time:YYYY-MM-DD HH:mm:ss.SSS} | " "{level: <8} | " "{name}:{function}:{line} - {message}" ) fluent_sender = sender.FluentSender( tag=TOPIC_TAG, host=FLUENTD_HOST, port=FLUENTD_PORT, buffer_max_size=8 * 1024 * 1024, timeout=3.0, retry_timeout=60 ) def log_to_fluent(message): try: record = message.record # 构建结构化日志数据 log_data = { 'topic': TOPIC_TAG, 'timestamp': record['time'].timestamp(), 'level': record['level'].name.lower(), 'message': record['message'], 'source': f"{record['file'].path}:{record['line']}", 'module': record['module'], 'function': record['function'], 'process_id': record['process'].id, 'thread_id': record['thread'].id, **record['extra'] } if not fluent_sender.emit(TOPIC_TAG, log_data): print(f"Fluentd 发送失败: {fluent_sender.last_error}") except Exception as e: print(f"日志处理异常: {str(e)}") # 移除默认处理器 logger.remove() # 添加控制台处理器 logger.add( sink=sys.stdout, level=LOG_LEVEL, format=STDOUT_FORMAT, colorize=True, backtrace=True, # 显示完整异常堆栈 diagnose=True, # 显示详细异常信息 ) if settings.ENVIRONMENT == 'docker': logger.add( log_to_fluent, level=LOG_LEVEL, # 处理 INFO 及以上级别 format="{message}", # 原始消息(实际使用结构化数据) backtrace=True, # 启用堆栈回溯 diagnose=True # 显示诊断信息 ) atexit.register(fluent_sender.close) # 导出配置好的logger __all__ = ["logger"] ```