企业官网建设流程全解析

官网

https://fastapi.tiangolo.com/zh/

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Web 框架，使用 Python 并基于标准的 Python 类型提示。

关键特性：

• 快速：极高性能，可与NodeJS和Go并肩（归功于 Starlette 和 Pydantic）。最快的 Python 框架之一。
• 高效编码：功能开发速度提升约 200% ～ 300%。*
• 更少 bug：人为（开发者）错误减少约 40%。*
• 直观：极佳的编辑器支持。处处皆可自动补全。更少的调试时间。
• 易用：为易用和易学而设计。更少的文档阅读时间。
• 简短：最小化代码重复。一次参数声明即可获得多种功能。更少的 bug。
• 健壮：生产可用级代码。并带有自动生成的交互式文档。
• 标准化：基于（并完全兼容）API 的开放标准：OpenAPI（以前称为 Swagger）和 JSON Schema。

创建项目

选用FastAPI模版创建

会自动下载Fastapi相关依赖

启动方式

直接使用右上角启动命令

看下配置

命令行启动

uvicorn main:app --reload

代码启动

if __name__ == '__main__': uvicorn.run('main:app', host="0.0.0.0", port=8000,reload=True)

接口文档

http://localhost:8000/docs

路由

纯路径

@app.get("/hi") async def hi(): return {"message": "hi!"}

路径参数

@app.get("/hello/{name}") async def say_hello(name: str): return {"message": f"Hello {name}"}

带校验的

@app.get("/book/{no}") async def book(no: str=Path(max_length=4)): return {"message": f"book {no}"}

参数校验

from fastapi import Path

Path( default=..., # 默认值，如果字段不是必填项则使用该值 *, default_factory=_Unset, # 默认值工厂函数，用于生成动态默认值（如当前时间） alias=None, # 字段别名，用于在请求/响应中使用的替代名称 alias_priority=_Unset, # 别名优先级，控制 alias 与 validation_alias 的优先顺序 validation_alias=None, # 校验时使用的别名，仅用于数据输入 serialization_alias=None, # 序列化时使用的别名，仅用于数据输出 title=None, # 字段标题，用于生成的 JSON Schema 文档 description=None, # 字段描述，用于生成的 JSON Schema 文档 gt=None, # 数值必须大于该值（greater than） ge=None, # 数值必须大于等于该值（greater than or equal） lt=None, # 数值必须小于该值（less than） le=None, # 数值必须小于等于该值（less than or equal） min_length=None, # 字符串或集合的最小长度 max_length=None, # 字符串或集合的最大长度 pattern=None, # 字符串必须匹配的正则表达式 regex=None, # 同 pattern，旧版参数名（已弃用） discriminator=None, # 联合类型鉴别器，用于多态模型解析 strict=_Unset, # 严格模式，禁止类型自动转换 multiple_of=_Unset, # 数值必须是该值的倍数 allow_inf_nan=_Unset, # 是否允许无穷大和 NaN 值 max_digits=_Unset, # 数字的最大总位数（用于 Decimal） decimal_places=_Unset, # 小数部分的最大位数（用于 Decimal） examples=None, # 字段示例值列表，用于文档展示 example=_Unset, # 单个示例值（已弃用，建议用 examples） openapi_examples=None, # OpenAPI 规范中的示例对象 deprecated=None, # 是否标记该字段为已弃用 include_in_schema=True, # 是否在生成的 JSON Schema 中包含该字段 json_schema_extra=None, # 额外添加到 JSON Schema 中的键值对 **extra # 其他额外参数，传递给底层校验逻辑 )

使用方法

book(no: str=Path(min_length=2,max_length=5,description='编号'))

查询参数

@app.get("/user/detail") async def detail(user_no: str): return {"message": f"用户 {user_no}信息"}

带校验

@app.get("/user/detail") async def detail(user_no: str=Query(min_length=2,max_length=5,description='用户编号')): return {"message": f"用户 {user_no}信息"}

参数校验

from fastapi import Query

Query( default=Undefined, # 默认值，如果查询参数未提供则使用该值 *, default_factory=_Unset, # 默认值工厂函数，用于生成动态默认值 alias=None, # 字段别名，用于在请求/响应中使用的替代名称 alias_priority=_Unset, # 别名优先级，控制 alias 与 validation_alias 的优先顺序 validation_alias=None, # 校验时使用的别名，仅用于数据输入 serialization_alias=None, # 序列化时使用的别名，仅用于数据输出 title=None, # 字段标题，用于生成的 JSON Schema 文档 description=None, # 字段描述，用于生成的 JSON Schema 文档 gt=None, # 数值必须大于该值（greater than） ge=None, # 数值必须大于等于该值（greater than or equal） lt=None, # 数值必须小于该值（less than） le=None, # 数值必须小于等于该值（less than or equal） min_length=None, # 字符串或集合的最小长度 max_length=None, # 字符串或集合的最大长度 pattern=None, # 字符串必须匹配的正则表达式 regex=None, # 同 pattern，旧版参数名（已弃用） discriminator=None, # 联合类型鉴别器，用于多态模型解析 strict=_Unset, # 严格模式，禁止类型自动转换 multiple_of=_Unset, # 数值必须是该值的倍数 allow_inf_nan=_Unset, # 是否允许无穷大和 NaN 值 max_digits=_Unset, # 数字的最大总位数（用于 Decimal） decimal_places=_Unset, # 小数部分的最大位数（用于 Decimal） examples=None, # 字段示例值列表，用于文档展示 example=_Unset, # 单个示例值（已弃用，建议用 examples） openapi_examples=None, # OpenAPI 规范中的示例对象 deprecated=None, # 是否标记该字段为已弃用 include_in_schema=True, # 是否在生成的 JSON Schema 中包含该字段 json_schema_extra=None, # 额外添加到 JSON Schema 中的键值对 **extra # 其他额外参数，传递给底层校验逻辑 )

请求体参数

1. 定义类型 (Define Type)

首先使用 Pydantic 的BaseModel定义一个数据模型，用于描述请求体中期望接收的数据结构。

from pydantic import BaseModel class User(BaseModel): username: str email: str password: str

2. 类型注解 (Type Annotation)

在路径操作函数（Path Operation Function）的参数中，直接使用上面定义的类作为类型注解。

@app.post("/user/add") async def add_user(user: User): return { "username": user.username, "email": user.email, "password": user.password }

带校验

class User(BaseModel): username: str=Field(min_length=2,max_length=5,description="用户名") email: str password: str

参数说明

def Field( # noqa: C901 default: Any = PydanticUndefined, # 字段默认值，未提供时使用该值 *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], Any] | None = _Unset, # 默认值工厂函数，用于生成动态默认值 alias: str | None = _Unset, # 字段别名，用于序列化/反序列化时的替代名称 alias_priority: int | None = _Unset, # 别名优先级，控制 alias 与 validation_alias 的优先顺序 validation_alias: str | AliasPath | AliasChoices | None = _Unset, # 校验时使用的别名，仅用于数据输入 serialization_alias: str | None = _Unset, # 序列化时使用的别名，仅用于数据输出 title: str | None = _Unset, # 字段标题，用于生成的 JSON Schema 文档 field_title_generator: Callable[[str, FieldInfo], str] | None = _Unset, # 自定义字段标题生成函数 description: str | None = _Unset, # 字段描述，用于生成的 JSON Schema 文档 examples: list[Any] | None = _Unset, # 字段示例值列表，用于文档展示 exclude: bool | None = _Unset, # 是否在序列化时排除该字段 exclude_if: Callable[[Any], bool] | None = _Unset, # 满足条件时排除该字段（返回 True 则排除） discriminator: str | types.Discriminator | None = _Unset, # 联合类型鉴别器，用于多态模型解析 deprecated: Deprecated | str | bool | None = _Unset, # 是否标记该字段为已弃用 json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = _Unset, # 额外添加到 JSON Schema 中的键值对或修改函数 frozen: bool | None = _Unset, # 是否冻结该字段（禁止修改） validate_default: bool | None = _Unset, # 是否对默认值也执行校验 repr: bool = _Unset, # 是否在 repr() 输出中包含该字段 init: bool | None = _Unset, # 是否在 __init__ 中作为参数（数据类场景） init_var: bool | None = _Unset, # 是否作为仅初始化变量（不保存为属性） kw_only: bool | None = _Unset, # 是否强制该字段为关键字参数（数据类场景） pattern: str | re.Pattern[str] | None = _Unset, # 字符串必须匹配的正则表达式 strict: bool | None = _Unset, # 严格模式，禁止类型自动转换 coerce_numbers_to_str: bool | None = _Unset, # 是否将数字自动转换为字符串 gt: annotated_types.SupportsGt | None = _Unset, # 数值必须大于该值（greater than） ge: annotated_types.SupportsGe | None = _Unset, # 数值必须大于等于该值（greater than or equal） lt: annotated_types.SupportsLt | None = _Unset, # 数值必须小于该值（less than） le: annotated_types.SupportsLe | None = _Unset, # 数值必须小于等于该值（less than or equal） multiple_of: float | None = _Unset, # 数值必须是该值的倍数 allow_inf_nan: bool | None = _Unset, # 是否允许无穷大和 NaN 值 max_digits: int | None = _Unset, # 数字的最大总位数（用于 Decimal） decimal_places: int | None = _Unset, # 小数部分的最大位数（用于 Decimal） min_length: int | None = _Unset, # 字符串或集合的最小长度 max_length: int | None = _Unset, # 字符串或集合的最大长度 union_mode: Literal['smart', 'left_to_right'] = _Unset, # 联合类型解析模式：smart 智能匹配，left_to_right 从左到右 fail_fast: bool | None = _Unset, # 是否在第一个校验错误时立即失败 **extra: Unpack[_EmptyKwargs], # 其他额外参数，传递给底层校验逻辑 ) -> Any:

响应类型

默认情况下，FastAPl会自动将路径操作函数返回的Python 对象(字典、列表、Pydantic 模型等)，经由jsonable_encoder 转换为JSON兼容格式，并包装为JSONResponse返回。这省去了手动序列化的步骤，让开发者能更专注于业务逻辑。如果需要返回非JSON数据(如HTML、文件流)，FastAPI提供了丰富的响应类型来返回不同数据

JSONResponse

@app.get("/hi") async def hi(): return {"message": "hi!"}

HTMLResponse

@app.get("/html",response_class=HTMLResponse) async def html(): return '<h1>Hello World</h1>'

PlainTextResponse

@app.get("/text",response_class=PlainTextResponse) async def text(): return PlainTextResponse('<h1>Hello World</h1>')

FileResponse

@app.get("/file",response_class=FileResponse) async def file(): return FileResponse('icon.png')

StreamingResponse

def generate(): for i in range(5): yield f"data chunk {i}\n" time.sleep(1) # 模拟耗时操作 @app.get("/stream") def stream(): return StreamingResponse(generate(), media_type="text/plain")

RedirectResponse

@app.get("/user/detail") async def detail(user_no: str=Query(min_length=2,max_length=5,description='用户编号')): return {"message": f"用户 {user_no}信息"} @app.get("/user/info",response_class=RedirectResponse) async def info(user_no: str): return RedirectResponse(url='/user/detail?user_no={}'.format(user_no))

自定义响应数据格式

文字说明：
response_model是路径操作装饰器（如@app.get或@app.post）的关键参数，它通过一个 Pydantic 模型来严格定义和约束 API 端点的输出格式。这一机制在提供自动数据验证和序列化的同时，更是保障数据安全性的第一道防线。

from pydantic import BaseModel class News(BaseModel): id: int title: str content: str @app.get("/news/{id}", response_model=News) async def get_news(id: int): return { "id": id, "title": f"这是第{id}本书", "content": "这是一本好书" }

异常处理

文字说明：
对于客户端引发的错误（4xx，如资源未找到、认证失败），应使用fastapi.HTTPException来中断正常处理流程，并返回标准错误响应。

from fastapi import FastAPI, HTTPException @app.get('/news/{id}') async def get_news(id: int): id_list = [1, 2, 3, 4, 5, 6] if id not in id_list: raise HTTPException(status_code=404, detail="当前id不存在") return {"id": id}

中间件(针对所有请求)

概述

中间件（Middleware）是一个在每次请求进入 FastAPI 应用时都会被执行的函数。
它在请求到达实际的路径操作（路由处理函数）之前运行，并且在响应返回给客户端之前再运行一次。

什么时候使用

使用中间件为每个请求前后添加统一的处理逻辑

• 01 多个接口：都需要验证用户身份

• 02 多个接口：都需要记录日志、性能数据

定义中间件

中间件:函数的顶部使用装饰器@app.middleware("http")

from fastapi import FastAPI from starlette.requests import Request app = FastAPI() @app.middleware("http") async def log(request: Request, call_next): url = request.url print(f"{request.method} {url} log enter") res = await call_next(request) print(f"{request.method} {url} log exit") return res @app.middleware("http") async def auth(request: Request, call_next): url = request.url print(f"{request.method} {url} auth enter") res = await call_next(request) print(f"{request.method} {url} auth exit") return res @app.get("/hi") async def hi(): return {"message": "hi!"}

依赖注入系统（针对指定请求）

概念

使用依赖注入系统来共享通用逻辑，减少代码重复

依赖项与注入

• 依赖项：可重用的组件（函数/类），负责提供某种功能或数据。

• 注入：FastAPI 自动帮你调用依赖项，并将结果“注入”到路径操作函数中。

优点：

• 代码复用：一次编写，多处使用。

• 解耦：业务逻辑与基础设施代码分离。

• 易于测试：轻松地用模拟依赖替换真实依赖进行测试。

依赖注入应用场景

写法

定义

from fastapi import FastAPI, Depends app = FastAPI() @app.get("/hi") async def hi(): return {"message": "hi!"} async def common_args(page_no: int, page_size: int): return { "page_no": page_no, "page_size": page_size, } @app.get("/page") def get_page_info(common=Depends(common_args)): page_no = common["page_no"] page_size = common["page_size"] return { "page_no": page_no, "page_size": page_size, }

ORM 简介

定义

ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式与数据库进行交互，而无需直接编写复杂的 SQL 语句。

优势：

• 减少重复的 SQL 代码

• 代码更简洁易读

• 自动处理数据库连接和事务

• 自动防止 SQL 注入攻击

ORM 分类

ORM使用流程

安装依赖

pip install "sqlalchemy[asyncio]" aiomysql

ORM建表

ORM – 创建数据库引擎

说明：使用create_async_engine创建异步引擎。

from sqlalchemy.ext.asyncio import create_async_engine ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_test?charset=utf8" # 创建异步引擎 async_engine = create_async_engine( ASYNC_DATABASE_URL, echo=True, # 可选：输出 SQL 日志 pool_size=10, # 设置连接池中保持的持久连接数 max_overflow=20 # 设置连接池允许创建的额外连接数 )

ORM – 定义模型类

步骤说明：

1. 基类，继承DeclarativeBase（包含通用属性和字段的映射）

2. 定义数据库表对应的模型类

class Base(DeclarativeBase): create_time: Mapped[datetime] = mapped_column( DateTime, default=datetime.now, comment="创建时间") update_time: Mapped[datetime] = mapped_column( DateTime, onupdate=func.now(), default=datetime.now, comment="修改时间") class Book(Base): __tablename__ = "book" id: Mapped[int] = mapped_column(primary_key=True) bookname: Mapped[str] = mapped_column(String(255)) author: Mapped[str] = mapped_column(String(255)) price: Mapped[float] = mapped_column(Float)

ORM – 创建数据库表

步骤说明：

1. 从连接池获取异步连接，开启事务，执行 ORM 操作

2. FastAPI 应用启动时，创建数据库表

async def create_tables(): async with async_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) @app.on_event("startup") async def startup_event(): await create_tables()

测试

ORM – 路由匹配中使用ORM

核心：创建依赖项，使用 Depends 注入到处理函数

# 创建异步会话工厂 AsyncSessionLocal = async_sessionmaker( bind=async_engine, # 绑定数据库引擎 class_=AsyncSession, # 指定会话类 expire_on_commit=False # 会话对象不过期，不重新查询数据库 ) # 依赖项，用于获取数据库会话 async def get_database(): async with AsyncSessionLocal() as session: try: yield session # 返回数据库会话给路由处理函数 await session.commit() # 无异常，提交事务 except Exception: await session.rollback() # 有异常则回滚 raise finally: await session.close() # 关闭会话

@app.get("/book/books") async def get_book_list( db: AsyncSession = Depends(get_database) ): # 查询所有书籍 result = await db.execute(select(Book)) # Book 模型类 user = result.scalars().all() return user

测试结果

数据库操作 – 查询

核心语句：await db.execute( select(模型类) )，返回一个 ORM 对象

➢ 获取所有数据
✓scalars().all()

➢ 获取单条数据
✓scalars().first()
✓get(模型类, 主键值)

列表查询

@app.get("/db/booklist") async def get_booklist(db: AsyncSession = Depends(get_db)): res = await db.execute(select(Book)) return res.scalars().all()

根据id查询

@app.get("/db/book/{book_id}") async def get_booklist(book_id: str,db: AsyncSession = Depends(get_db)): book = await db.get(Book,book_id) return book

数据库操作 – 查询条件

select(Book).where(条件, 条件2, ...)

条件：

✓ 比较判断：==;>;<;>=;<=等
✓ 模糊查询：like()
✓ 与非查询：&;|;~
✓ 包含查询：in_()

@app.get("/db/booklist") async def get_booklist(name:str,create_time:datetime,db: AsyncSession = Depends(get_db)): res = await db.execute(select(Book).where((Book.bookname.like('%'+name+'%') )&(Book.update_time !=None) , Book.create_time >=create_time)) return res.scalars().all()

数据库操作 – 聚合查询

聚合计算：func.方法(模型类.属性)

➢ count：统计行数量
➢ avg：求平均值
➢ max：求最大值
➢ min：求最小值
➢ sum：求和

@app.get("/book/count") async def get_count(db: AsyncSession = Depends(get_db)): # result = await db.execute(select(func.count(Book.id))) # result = await db.execute(select(func.max(Book.price))) # result = await db.execute(select(func.sum(Book.price))) result = await db.execute(select(func.avg(Book.price))) count = result.scalar() return count

数据库操作 – 分页查询

分页查询：select().offset().limit()

➢ offset：跳过的记录数
➢ limit：返回的记录数

@app.get("/book/page") async def get_page(page_no:int,page_size:int,db: AsyncSession = Depends(get_db)): result = await db.execute(select(Book).offset((page_no-1)*page_size).limit(page_size)) data = result.scalars().all() return data

ORM – 查询 – 总结

核心思路：

★select()→db.execute()→ 从 ORM 对象获取数据 → 响应结果
★db.get(模型类, 主键值)

从 ORM 对象获取数据的方式

➢ 获取所有数据
✓scalars().all()

➢ 获取单条数据
✓scalars().first(): 提取第一个数据
✓scalar_one_or_none(): 提取一个或 null
✓scalar(): 提取标量值（配合聚合查询使用）

数据库操作 – 新增

核心步骤：定义 ORM 对象 → 添加对象到事务：add(对象)→commit提交到数据库

class BookBase(BaseModel): id: int bookname: str author: str price: float @app.post("/book/add_book") async def add_book(book: BookBase, db: AsyncSession = Depends(get_db)): # 获取 book 参数，创建图书对象（__dict__ 返回 book 对象的属性字典） book_obj = Book(**book.__dict__) db.add(book_obj) await db.commit() return book

数据库操作 – 更新

核心步骤：查询get→ 属性重新赋值 →commit提交到数据库

@app.put("/book/update_book/{book_id}") async def update_book(book_id: int, data: BookBase, db: AsyncSession = Depends(get_db)): # 1. 查询 book = await db.get(Book, book_id) if book is None: raise HTTPException(status_code=404, detail="Book not found") # 2. 修改属性（重新赋值） book.bookname = data.bookname book.author = data.author book.price = data.price # 3. 提交 await db.commit() return book

数据库操作 – 删除

核心步骤：查询get→delete删除 →commit提交到数据库

@app.delete("/book/delete_book/{book_id}") async def delete_book(book_id: int, db: AsyncSession = Depends(get_database)): db_book = await db.get(Book, book_id) if db_book is None: raise HTTPException(status_code=404, detail="Book not found") await db.delete(db_book) await db.commit() return {"message": "Book deleted"}

总结