首页/资讯中心/详情

基于FastAPI的AI智能体Web系统构建（四）

基于FastAPI的AI智能体Web系统构建（四）
📅 发布时间：2026/7/27 7:17:29

目录

一.FastAPI中间件

二.FastAPI跨域资源共享

三.FastAPI业务划分

四.FastAPI项目结构和搭建

一.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 # 创建迁移脚本，一般是生成一堆类对应的迁移脚本

初始化后即可开始测试各功能。