基于FastAPI的AI智能体Web系统构建（四）
·
目录
一.FastAPI中间件
中间件 是位于客户端和应用程序核心逻辑之间的软件层，用于：
- 拦截请求和响应
- 在请求到达路由处理程序之前进行处理
- 在响应返回给客户端之前进行处理
# 中间件的使用
from fastapi import FastAPI
from fastapi.responses import Response
app = FastAPI()
@app.middleware('http')
async def middleware1(request, call_next):
print('处理业务：之前')
print(request.method, request.url)
# if request.url.path == '/middle':
# print('这个用户访问了middle接口')
#中间件必须返回一个response对象，否则会报错
# return Response(content="你没有权限访问该接口")
response = await call_next(request)
# 设置响应头信息
response.headers['X-Token'] = '123456'
print('处理业务：之后')
return response
@app.get('/middle')
async def get_middle():
print('处理业务：处理中')
return '业务逻辑处理结果'
if __name__ == "__main__":
import uvicorn
uvicorn.run('main25:app', host="127.0.0.1", port=8000, reload=True)
# 中间件的使用2
from fastapi import FastAPI
app = FastAPI()
@app.middleware('http')
async def middleware1(request, call_next):
print('中间件1:请求前')
response = await call_next(request)
print('中间件1:请求后')
return response
# 这种注解写法是中间件3写法的封装后的写法，写起来更方便
@app.middleware('http')
async def middleware2(request, call_next):
print('中间件2:请求前')
response = await call_next(request)
print('中间件2:请求后')
return response
# 创建一个中间件记录日志信息
# 下面的写法更底层，是服务器端写法
#中间件从下往上执行，所以中间件3会先执行，然后中间件2再执行，最后中间件1执行
class LogMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
print('中间件3:请求前')
await self.app(scope, receive, send)
print('中间件3:请求后')
app.add_middleware(LogMiddleware)
@app.get('/middle')
async def get_middle():
print('逻辑处理完成！')
return '这是中间件'
if __name__ == "__main__":
import uvicorn
uvicorn.run('main26:app', host="127.0.0.1", port=8000, reload=True)
二.FastAPI跨域资源共享
同源策略（ SOP ）
同源策略 是浏览器的一种安全机制，限制了一个源（ origin ）的网页如何与另一个源的资源进行交互。
源（ origin ）由 协议 （如 HTTP/HTTPS ）、 域名 （如 example.com ）和 端口 （如 80 或 443 ）组成。
例如， https://example.com:443 和 http://example.com:80 是不同源，因为协议和端口不同。
同源策略防止恶意网站通过脚本（如 JavaScript ）未经授权访问其他网站的数据，例如窃取用户的敏感信息。
但是：现代Web应用经常需要跨源请求！
CORS 的工作原理
CORS （ Cross-Origin Resource Sharing ，跨源资源共享）是一种基于HTTP 的机制，它允许服务器指示哪些其他源（域名、协议或端口）可以访问其资源，从而绕过浏览器的同源策略（Same-Origin Policy， SOP ）限制。
# CORS使用
from fastapi import FastAPI
from fastapi.responses import Response
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
#fastapi的CORS中间件,直接将CORS中间件添加到应用中
app.add_middleware(
CORSMiddleware,
allow_origins=['*'], # 允许的域名 ：如 ['http://127.0.0.1:8080']
allow_credentials=True, # 允许携带cookie
allow_methods=['*'], # 允许的请求方法
allow_headers=['*'], # 允许的请求头
)
# 手动添加CORS头，需要在中间件中处理option请求，写起来麻烦
# @app.middleware('http')
# async def add_cors_headers(request, call_next):
# # 处理option 请求
# if request.method == 'OPTIONS':
# headers = {
# "Access-Control-Allow-Origin": '*',
# "Access-Control-Allow-Methods": 'GET,POST,PUT,DELETE,OPTIONS',
# "Access-Control-Allow-Headers": 'Content-Type,Authorization'
# }
# return Response(status_code=200, headers=headers)
# response = await call_next(request)
# response.headers['Access-Control-Allow-Origin'] = '*' # 允许所有源访问
# return response
@app.get('/info')
async def get_info():
return '内容成功获取!'
if __name__ == "__main__":
import uvicorn
uvicorn.run('main27:app', host="127.0.0.1", port=8000, reload=True)<!DOCTYPE html>
<html>
<head>
<title>CORS Test</title>
</head>
<body>
<h1>CORS 测试</h1>
<button onclick="testCors()">测试CORS</button>
<p id="result">这块儿显示响应</p>
<script>
async function testCors() {
try {
const response = await fetch('http://127.0.0.1:8000/info', {
method: 'GET',
headers: {
'Content-Type': 'application/json'
}
});
const data = await response.json();
document.getElementById('result').textContent = 'Success: ' + JSON.stringify(data);
} catch (error) {
document.getElementById('result').textContent = 'Error: ' + error.message;
}
}
</script>
</body>
</html>三.FastAPI业务划分
APIRouter 核心作用
模块化架构 ：将大型应用拆分为独立功能模块
1
路由分组 ：统一管理相关端点
2
组织优化 ：解耦业务逻辑，提升可维护性
有了路由之后，就可以将业务细分到各种路由当中，编写具体业务逻辑，然后通过导包的方式来相互联系，实现解耦。
# APIRouter的使用
from fastapi import FastAPI, APIRouter
app = FastAPI()
# tags路由注释，prefix路由前缀（第一种写法，生成路由时指定前缀）
# main_router = APIRouter(tags=['主应用'], prefix='/main')
# user_router = APIRouter(tags=['用户应用'], prefix='/user')
# item_router = APIRouter(tags=['商品应用'])
v1_router = APIRouter(prefix='/api/v1')
v2_router = APIRouter(prefix='/api/v2')
user_router = APIRouter(tags=['用户应用'], prefix='/user')
item_router = APIRouter(tags=['商品应用'])
# @main_router.get('/info')
# async def get_info1():
# return 'main:内容成功获取!'
@user_router.get('/info')
async def get_info2():
return 'user:内容成功获取!'
@user_router.get('/create')
async def create_info2():
return 'user:内容成功获取!'
@item_router.get('/info')
async def get_info3():
return 'item:内容成功获取!'
#将路由添加到app中
# app.include_router(main_router)
# app.include_router(user_router)
# 前缀第二种写法，添加路由到app时指定前缀
# app.include_router(item_router, prefix='/item')
# 让v1_router包含user_router和item_router，成为主路由
v1_router.include_router(user_router)
v1_router.include_router(item_router)
# app只需要包含主路由即可
# 实际开发时一般只保留一个版本，当然两个也不是不行
app.include_router(v1_router)
app.include_router(v2_router)
if __name__ == "__main__":
import uvicorn
uvicorn.run('main28:app', host="127.0.0.1", port=8000, reload=True)
四.FastAPI项目结构和搭建
FastAPI 项目中，代码全部集中在单个文件中，虽然这种方式 适合 小型项目 或 原型开发 ，但在企业级应用中，将代码拆分为 多个文件 和目录 是更优的实践，主要原因：
职责分离：
将应用的各个组成部分（如数据库配置、模型、模式、路由、中间件）分别放置在单独的文件或目录中，使代码结构更清晰，便于理解和维护
提高代码可读性和可维护性：
将代码拆分为小而专一的文件，开发者可以快速定位和修改特定功能，而无需在单一长文件中搜索
增强可扩展性：
模块化的结构便于扩展，当需要添加新功能（如新的模型或路由）时，只需在相应的目录中创建新文件，而不影响现有代码
清晰的依赖管理：
通过在每个文件中显式导入依赖，可以清楚地看到模块之间的关系，便于调试和理解代码逻辑
便于团队协作：
在企业开发中，多个开发者可能同时开发不同功能，拆分后的文件结构允许团队成员并行工作。例如，一个开发者修改用户路由，另一个开发者可以同时处理商品路由，互不干扰。
符合企业级项目规范：
企业级项目通常需要遵循标准化的代码组织规范，拆分文件是行业最佳实践之一。这种结构也便于集成自动化测试、CI/CD 流程和代码审查。
项目结构示例
按软件功能目录结构示例
按业务模块的目录结构示例
project/
├── main.py # 主入口文件，初始
化 FastAPI 应用
├── config/
│ └── database.py # 数据库配置和
Tortoise-ORM 初始化
├── modules/ # 业务模块目录
│ ├── user/ # 用户管理模块
│ │ ├── __init__.py # 标记 user 为
Python 包
│ │ ├── models.py # 用户相关的数据模型
│ │ ├── schemas.py # 用户相关的
Pydantic 模式
│ │ └── routers.py # 用户相关的 API 路
由
│ ├── item/ # 商品管理模块
│ │ ├── __init__.py # 标记 item 为
Python 包
│ │ ├── models.py # 商品相关的数据模型
（当前为空）
│ │ ├── schemas.py # 商品相关的
Pydantic 模式（当前为空）
│ │ └── routers.py # 商品相关的 API 路
由
└── middleware/
└── user_middleware.py # 自定义中间
件逻辑几个常用库：
FastAPI 是基于 Python 的高性能异步 Web 接口开发框架，专门用来写后端 HTTP 接口（RESTful API），适配 async/await，是目前 Python 后端主流框架。 底层依赖两大核心：
- Starlette：异步网络底层
- Pydantic：数据校验、类型解析
Pydantic 是 Python 数据校验 + 数据类型转换库，依靠 Python 类型注解做规则约束；FastAPI 的核心底层依赖，Tortoise-ORM 也常搭配它做接口出参序列化。 简单一句话：给入参、JSON、字典做强制格式校验、自动类型转换，不用手写一堆 if 判断。
主流版本：
- Pydantic V1：旧稳定版
- Pydantic V2：重写底层，用 Rust 提速，性能提升数倍，现在新项目首选
Tortoise-ORM 是Python 异步 ORM 数据库框架，对标同步的 SQLAlchemy，适配 asyncio，主打搭配 FastAPI、aiohttp、Sanic 等异步 Web 框架使用，名字 tortoise 本意乌龟，寓意稳定可靠。
*************************************************************************************************************
创建main29.py:
# FastAPI项目结构
from typing import Dict
from fastapi import FastAPI, APIRouter
# 自动类型校验库
from pydantic import BaseModel
# 管理ORM数据库的库
from tortoise import fields, models
from tortoise.contrib.fastapi import register_tortoise
app = FastAPI()
# Tortoise-ORM 数据库配置
TORTOISE_ORM: Dict = {
"connections": {
# 生产环境示例：MySQL
"default": "mysql://root:root@127.0.0.1:3306/fastapi_db1",
},
"apps": {
"models": {
"models": ["main29", "aerich.models"], # 模型模块和 Aerich 迁移模型
"default_connection": "default",
}
},
# 连接池配置（推荐）
"use_tz": False, # 是否使用时区
"timezone": "UTC", # 默认时区
"db_pool": {
"max_size": 10, # 最大连接数
"min_size": 1, # 最小连接数
"idle_timeout": 30 # 空闲连接超时（秒）
}
}
v1_router = APIRouter(prefix='/api/v1')
user_router = APIRouter(tags=['用户应用'], prefix='/user')
item_router = APIRouter(tags=['商品应用'])
@app.middleware("http")
async def middleware(request, call_next):
print("请求前:")
response = await call_next(request)
print("请求后:")
return response
#面向前端接口层，可以做数据校验，且可以限制id和create_at的提交，保护数据库安全
class UserSchema(BaseModel):
name: str
email: str
age: int
#面向数据库数据层
class User(models.Model):
# 主键，数据库会自动管理，不需要管它
id = fields.IntField(pk=True)
name = fields.CharField(max_length=64)
email = fields.CharField(max_length=255)
age = fields.IntField(default=1)
# auto_now_add=True 参数的含义是：在创建记录时自动设置当前时间，且只在创建时设置一次。
# 所以也不用管它
create_at = fields.DatetimeField(auto_now_add=True)
@user_router.post('/create')
async def create_user(user: UserSchema):
# 转换为字典，再创建记录，会自动找相同的字段名对应赋值，前提是字段名一定要相同
# 另外两个属于自动创建字段，不需要赋值
user2 = await User.create(**user.model_dump())
# 等价写法：
# user2 = await User.create(name=user.name, email=user.email, age=user.age)
return user2
@user_router.get('/info')
async def get_info():
# 从数据库中查询第一个用户
user = await User.first()
# 有的时候前端不需要后端的全部字段，比如主键id和创建时间create_at
# 所以在转换为模型时，需要排除这两个字段，使用model_validate方法来过滤出相同字段
user_schema = UserSchema.model_validate(user.__dict__)
return user_schema
@item_router.get('/info')
async def get_info():
return '商品信息'
v1_router.include_router(user_router)
v1_router.include_router(item_router)
app.include_router(v1_router)
# 注册Tortoise-ORM数据库
register_tortoise(app,
config=TORTOISE_ORM,
generate_schemas=True, # 开发环境自动生成表结构
add_exception_handlers=True # 添加默认异常处理
)
'''
# 根据main29.py文件中的User模型去生成迁移文件,因为此文件里的TORTOISE_ORM变量相当于配置文件
# 这两个init初始化命令一般在项目周期里只执行一次
aerich init -t main29.TORTOISE_ORM # 初始化
aerich init-db # 根据写的类来创建表
# 这两个迁移命令一般在项目周期一般执行很多次，执行迁移(migrate)后都需要同步迁移（upgrade）到数据库
aerich migrate --name "注释" # 迁移，即写好了图纸（生成SQL）
aerich upgrade # 同步迁移到数据库，按照图纸施工（执行 SQL）
'''
if __name__ == "__main__":
import uvicorn
uvicorn.run('main29:app', host="127.0.0.1", port=8000, reload=True)
执行如下两个命令初始化数据库
aerich init -t main29.TORTOISE_ORM # 初始化
aerich init-db # 创建迁移脚本，一般是生成一堆类对应的迁移脚本初始化后即可开始测试各功能。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
6
0- 0
已为社区贡献4条内容
所有评论(0)