FastApi

FastApi 总结

前言&依赖

官方文档：https://fastapi.org.cn/learn/

在编写路由的时候请参考 Api规范

依赖：

1
2
python 3.10+
pip install fastapi, sqlalchemy[asyncio], aiomysql, uvicorn

项目构成

一个完整的项目应有以下配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
Project/
|├── config/			# 配置存放文件夹
||├── db_config.py		# 数据库相关配置文件
||└── ...
|├── crud/				# 存放操控数据库代码文件(增删改查)
||└── ...
|├── models/			# ORM 模型
||└── ...
|├── routers/			# 存放路由相关
||└── ...
|├── schemas/			# pydantic 模型类
||└── ...
|├── utils/ 			# 工具类
||└── ...
└── main.py 			# 主程序入口

确保各功能各司其职是完成一个项目的关键，同时模块化处理能够便于项目管理与优化

项目开发时应遵循：

​ 阅读了解 API 规范 → 创建路由 → 根据路由创建pydantic模型(可选) → 确认实现逻辑 → 创建对应的 ORM 模型 → 完成crud操作 → 完善路由 → 在交互式文档或前端页面调用api查看是否正常工作 → 修复bug/投入使用

遇见bug处理方式:

  • 检查是否挂载路由

  • 询问 AI

  • 参考官方文档

  • 搜索相似问题解决方法

主程序入口

主程序通常写在 main.py 文件中，启动程序的方法可以使用命令行

1
fastapi dev main.py

也可以在vscode中进行配置(视个人情况而定)

1
2
3
4
5
6
7
8
9
10
11
12
13
{
            "name": "Python Debugger: FastAPI",
            "type": "debugpy",
            "request": "launch",
            "module": "uvicorn",
            "args": [
                "main:app",
                "--reload"
            ],
            "jinja": true,
            "justMyCode": true,
            "cwd": "${fileDirname}"
        }

一个main.py通常含有以下配置：

  • lifespan 生命周期

    https://fastapi.org.cn/advanced/events/

  • app 运行实例

  • middleware 中间件

    https://fastapi.org.cn/tutorial/middleware/

  • include_router 挂载的路由

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from routers import users

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 开始前置
    yield
    # 结束后置

app = FastAPI(lifespan= lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],        # 允许的源，开发环境下使用 *  上线请使用具体域名
    allow_credentials= True,    # 允许携带 cookie
    allow_methods=["*"],        # 允许的请求方法
    allow_headers=["*"],        # 允许的请求头
)

@app.get("/")					# 此处是重定向到交互式文档
async def root():
    return RedirectResponse("/docs")

app.include_router(users.router)	# 挂载的路由

config相关

这是配置文件夹，可以同来配置数据库等基础配置，并将其封装为依赖

https://fastapi.org.cn/reference/dependencies/

数据库我们通常使用也同样具有异步的aiomysql驱动，搭配sqlalchemy使用，在构建依赖前，必须先配置数据库项

  • DB_URl 该项指定数据库类型和所用的驱动

    参考使用：DB_URL = “mysql+aiomysql://username:password@localhost:3306/test?charset=utf8mb4”

  • engine 构建数据库引擎

    参考文档：https://docs.sqlalchemy.org.cn/en/20/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.create_async_engine

  • async_session 会话项

    用于管理会话，需要绑定引擎

  • get_db 依赖项

    用于构建数据库依赖

    示例文件：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

    DB_URL = "mysql+aiomysql://username:password@localhost:3306/test?charset=utf8mb4"

    engine = create_async_engine(
        DB_URL,
        echo=True,				# 是否返回数据库操作信息
        pool_size=10,			# 连接池大小
        max_overflow=20			# 额外允许的溢出连接数
    )

    async_session = async_sessionmaker(
        bind= engine,
        class_= AsyncSession,
        expire_on_commit= False
    )

    # dependence 依赖项
    async def get_db():
        async with async_session() as session:
            try:
                yield session
                await session.commit()
            except Exception as e:
                await session.rollback()
                raise e
            finally:
                await session.close()

models相关

该文件夹管理ORM模型类，模型用来映射数据库表，是crud的基础

一个ORM模型类继承于父类 Base，而 Base 类则继承于 DeclarativeBase 类

  • Base 类常用来作为基类，即通用的表列结构，如创建时间和更新时间
  • ORM模型 类则具体规定映射表的各项结构

一个ORM模型类必须有对应的表名，即 __tablename__，对于具体的高频查询字段，可以设置额外的索引来提高效率，常定义在 __table_args__，接着通过 Mapped[] 和 mapped_column() 来映射字段

示例文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, DateTime, func, Integer, Index, Text, ForeignKey
import datetime
from typing_extensions import Annotated

ID_PRIMARY_KEY = Annotated[int, mapped_column(Integer, primary_key= True, autoincrement= True, comment= "id")]
# 此处用 Annotated 进行封装，可以减少重复操作，可选

class Base(DeclarativeBase):
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default= func.now(),
        default= func.now(),
        comment= "创建时间"
    )
    
    updated_at: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default= func.now(),
        default= func.now(),
        onupdate= func.now(),
        comment= "更新时间"
    )

class Category(Base):
    __tablename__ = "news_category"
    
    id: Mapped[ID_PRIMARY_KEY]			# 等效于 id: Mapped[int] = mapped_column(Integer, primary_key= True, autoincrement= True, comment= "id")
    
    name: Mapped[str] = mapped_column(String(50), nullable= False, unique= True, comment= "分类名称")
    sort_order : Mapped[int] = mapped_column(Integer, default= 0, nullable= False, comment= "排序")
    
class News(Base):
    __tablename__ = "news"

    # * 创建索引：提升查询效率
    __table_args__ = (
        Index('fk_news_category_idx', 'category_id'), # 高频查询字段
        Index('idx_publish_time', 'publish_time')
    )
    
    id: Mapped[ID_PRIMARY_KEY]	# 简化代码
    
    title: Mapped[str] = mapped_column(String(255), nullable= False, comment= "新闻标题")
    description: Mapped[Optional[str]] = mapped_column(String(500), nullable= False, comment= "新闻简介")
    content: Mapped[str] = mapped_column(Text, nullable= False, comment= "新闻内容")
    image: Mapped[str | None] = mapped_column(String(255), comment= "新闻图片URL")
    author: Mapped[str | None] = mapped_column(String(50), comment= "作者")
    category_id: Mapped[int] = mapped_column(Integer, ForeignKey('news_category.id'), nullable= False, comment= "新闻分类ID")
    views: Mapped[int] = mapped_column(Integer, default= 0, nullable= False, comment= "浏览量")
    publish_time: Mapped[datetime.datetime] = mapped_column(DateTime, insert_default= func.now(), default= func.now(), comment= "发布时间")

routers相关

这个文件夹用来管理路由

一个路由通常包含所有该模块相关的接口，每个路由文件都必须包含一个 router 用来挂载接口

  • router 指的是 fastapi 中的 APIRouter，其中包含的 prefix 和 tags 参数非常实用

    • prefix：前缀，用于规范接口
    • tags： 分类标签，用于交互式文档 docs 的渲染显示，用于区分其他路由，便于调试

    参考文档：https://fastapi.org.cn/reference/apirouter/

  • 每个接口的定义应遵循 api文档

  • 注意！别忘记在主程序挂载路由

示例文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated
from fastapi import APIRouter, Depends, Query, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from crud import favorite
from config.db_config import get_db
from schemas.favorite import FavoriteCheckResponse
from utils.response import success_response
from utils.auth import get_current_user
from models.users import User

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

@router.get("/check")
async def check_favorite(
    data : Annotated[int, Query(..., alias= "newsId")],		# 此处是从前端接收的请求数据
    														# 可在schemas自定义pydantic
    														# 类自定义接收的请求数据
    user : User = Depends(get_current_user),		# 此处为上文提及的依赖项，每次接口收到
    session: AsyncSession = Depends(get_db)			# 请求时会自动调用
):
    result = await favorite.is_news_favorite(data, user, session)
    response_data = FavoriteCheckResponse(is_favorite= result)
    return success_response(message= "获取收藏状态成功", data= response_data)
	# 此处的success_response为自定义响应，具体可参照api文档自行构建

schemas相关

schemas内定义着接收的请求类型和响应的数据类型，用于与前端的交互

每个最基础的pydantic类都必须继承于 BaseModel，可以调整设置 ConfigDict() 来调整命名兼容和 从orm对象中获取字段值

样例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from typing_extensions import Annotated
from pydantic import BaseModel, Field, ConfigDict

class UserRequest(BaseModel):
    username : str		# 此处可用 Annotated[str, Field(..., alias= "username")] 来
    password : str		# 进一步规定输入，也可以要求前端进行额外校验，如果使用此格式请务必设置
        				# populate_by_name= True 来避免可能的命名错误

class UserInfoBase(BaseModel):
    nickname : Annotated[str | None, Field(None, max_length= 50, description= "昵称")]
    avatar : Annotated[str | None, Field(None, max_length= 255, description= "头像URL")]
    bio : Annotated[str | None, Field(None, max_length= 500, description= "简介")]
    gender : Annotated[str | None, Field(None, max_length= 10, description= "性别")]

class UserInfoResponse(UserInfoBase):
    id : int
    username : str
    
    model_config = ConfigDict(
        populate_by_name= True, # alias / 字段名 兼容
        from_attributes= True   # 允许从 ORM 对象中获取字段值
    )

crud相关

crud内主要是数据库的增删改查操作，主要用途为被 routers 调用

如果一个接口没有正常运行，请优先检验数据库query是否正常被执行

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, update
from models.users import User
from schemas.users import UserRequest
from utils import security

async def get_user_by_username(session: AsyncSession, username: str):
    query = select(User).where(User.username == username)
    result = await session.execute(query)
    return result.scalar_one_or_none()

async def create_user(session: AsyncSession, user_data : UserRequest):
    # ! 密码加密
    hashed_pwd = security.get_hash_password(user_data.password)
    user = User(username= user_data.username, password= hashed_pwd)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

utils相关

该文件夹保存工具类函数，如上文提及的 auth(认证)，response(响应)文件，合理利用能够节省大量时间，提高代码复用性和可读性

工具类可以高度自定义，具体情况请按照实际开发为主，常用的工具类可以为：

  • security： 安全模块，用于加密密码等相关用途
  • response： 响应模块，用于响应数据，封装可提高可读性和复用性
  • exception： 异常捕获模块，用于捕获异常，快速分析问题
    • 异常可分为 业务层面异常，数据库完整性约束错误，数据库操作错误，其他未定义的异常
  • auth： 认证模块，用于检验用户令牌相关
#python #fastapi
FastApi
http://example.com/2026/01/30/fastapi/
作者
Suzuran
发布于
2026年1月30日
许可协议