FastAPI 学习笔记
1.pycharm新建FastAPI项目2.右上角运行项目3.浏览器访问访问 http://127.0.0.1:8000和 http://127.0.0.1:8000/hello/fastapi能看到以下 JSON 即成功{message:Hello fastapi}访问High Performance Web Crawler API - Swagger UI能看到交互式文档即完全成功。4.安装 FastAPI 和 Uvicorn 依赖这里提示可以更新版本就更新了一下pip install fastapi uvicorn5.添加测试代码app.get(/hello/{user_id}) def get_user(user_id: int): return {user_id: user_id,msg:查询用户成功}6.重新启动项目7.浏览器访问成功请求8.多功能测试练习不同类型的请求、响应和中间件from fastapi import FastAPI app FastAPI() app.get(/) async def root(): return {message: Hello World} app.get(/hello/{name}) async def say_hello(name: str): return {message: fHello {name}} # ------------------------------- # 测试 app.get(/hello/{user_id}) def get_user(user_id: int): return {user_id: user_id,msg:查询用户成功} # --------------------------------------------------- # 不同请求方法的路由 # --------------------------------------------------- # GET请求查询所有用户 app.get(/users) def get_users(): return {msg: 查询所有用户成功} # POST请求新增用户 app.post(/users) def add_user(): return {msg: 新增用户成功} # PUT请求修改指定用户 app.put(/users/1) def update_user(): return {msg: 修改ID为1的用户成功} # DELETE请求删除指定用户 app.delete(/users/1) def delete_user(): return {msg: 删除ID为1的用户成功} # --------------------------------------------------- # 路径参数的使用 # --------------------------------------------------- # 基础路径参数提取user_id注解为int类型 app.get(/users/{user_id}) def get_user(user_id: int): return {user_id: user_id, msg: 查询用户成功} # 多路径参数同时提取order_id和item_id app.get(/orders/{order_id}/items/{item_id}) def get_order_item(order_id: int, item_id: str): return {order_id: order_id, item_id: item_id} # 请求参数接收前端传递的数据 # 示例 1查询参数: # 带默认值的查询参数可选 app.get(/items) def get_items(skip: int 0, limit: int 10): # 访问/items → skip0, limit10 # 访问/items?skip5limit20 → skip5, limit20 return {skip: skip, limit: limit, items: [商品1, 商品2]} # 必选查询参数不加默认值 app.get(/search) def search(q: str): # 必须传q参数如/search?q手机 return {q: q, msg: f搜索关键词{q}} # ----------------------------------------------- # 示例 2请求体需导入 BaseModel from pydantic import BaseModel # 1. 定义请求体模型数据模板 class User(BaseModel): username: str # 必选字段 password: str # 必选字段 age: int 18 # 可选字段默认值18 # 2. POST请求接收请求体 app.post(/login) def login(user: User): # FastAPI自动把前端传的JSON转成User对象 if user.username admin and user.password 123456: return {code: 200, msg: 登录成功, age: user.age} else: return {code: 400, msg: 用户名或密码错误} # --------------------------------------------------- # 示例3混合参数路径 查询 请求体 # 定义请求体模型 class ItemUpdate(BaseModel): name: str price: float # 路径参数item_id查询参数is_verify请求体ItemUpdate app.put(/items/{item_id}) def update_item(item_id: int,item: ItemUpdate, is_verify: bool True): return { item_id: item_id, is_verify: is_verify, item_name: item.name, item_price: item.price } # 响应处理返回前端的内容 # 示例 1基础响应 # 返回字典推荐自动转JSON app.get(/dict) def resp_dict(): return {code: 200, msg: 成功, data: [1,2,3]} # 返回字符串 app.get(/str) def resp_str(): return Hello FastAPI # 返回数字 app.get(/num) def resp_num(): return 200 # --------------------------------------------------- # 示例 2自定义响应状态码 from fastapi.responses import JSONResponse # 方式1装饰器指定状态码推荐 app.post(/users, status_code201) # 201表示创建成功 def add_user(): return {msg: 用户创建成功} # 方式2返回JSONResponse自定义响应头 app.get(/custom) def custom_resp(): return JSONResponse( content{code: 200, msg: 自定义响应}, status_code200, headers{X-Token: 123456} # 自定义响应头 ) # 中间件 from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import time app FastAPI() # 1. 基础中间件统计请求耗时 app.middleware(http) async def add_process_time_header(request: Request, call_next): print(Middleware 1 start...) # 请求到达时执行前置逻辑 start_time time.time() # 调用下一个中间件/路由处理函数 response await call_next(request) # 响应返回时执行后置逻辑 process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) # 给响应加自定义头 print(Middleware 1 end...) return response # 2. 中间件全局异常捕获 app.middleware(http) async def global_exception_handler(request: Request, call_next): print(Middleware 2 start...) try: response await call_next(request) except Exception as e: return JSONResponse( status_code500, content{detail: f服务器内部错误: {str(e)}} ) print(Middleware 2 end...) return response # 测试路由 app.get(/test) async def test(): return {message: Hello Middleware} # ----------------------------------------------------------- # 常用内置中间件 类似Java中的servlet Filter from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware # 跨域中间件最常用 app.add_middleware( CORSMiddleware, allow_origins[*], # 允许的源生产环境需指定具体域名 allow_credentialsTrue, allow_methods[*], # 允许的请求方法 allow_headers[*], # 允许的请求头 ) # Gzip压缩中间件优化响应体积 app.add_middleware(GZipMiddleware, minimum_size1000) # 仅压缩大于1000字节的响应