网站联系方式模板,wordpress页面搜索,网站源码酒类,网站开发项目的需求分析英文版文档#xff1a;https://fastapi.tiangolo.com/ 中文版文档#xff1a;https://fastapi.tiangolo.com/zh/ 1、FastAPI 教程 简 介
FastAPI 和 Sanic 类似#xff0c;都是 Python 中的异步 web 框架。相比 Sanic#xff0c;FastAPI 更加的成熟、社区也更加的活跃。 …
英文版文档https://fastapi.tiangolo.com/ 中文版文档https://fastapi.tiangolo.com/zh/ 1、FastAPI 教程 简 介
FastAPI 和 Sanic 类似都是 Python 中的异步 web 框架。相比 SanicFastAPI 更加的成熟、社区也更加的活跃。
FastAPI 站在巨人的肩膀上 很大程度上来说这个巨人就是指 Flask 框架。 FastAPI 从语法上和 Flask 非常的相似有异曲同工之妙。 其实不仅仅是 FastAPI 就连 Sanic 也是基于 Flask 快速开发的 Web API 框架。
FastAPI 是一个用于构建 API 的现代、快速高性能的 web 框架使用 Python 3.6 并基于标准的 Python 类型提示。
它具有如下这些优点
快速可与 NodeJS 和 Go 比肩的极高性能归功于 Starlette 和 Pydantic。Starlette 用于路由匹配Pydantic 用于数据验证高效编码提高功能开发速度约 200 至 300更少 bug减少约 40 的人为开发者导致错误。智能极佳的编辑器支持。处处皆可自动补全减少调试时间简单设计的易于使用和学习阅读文档的时间更短简短使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少健壮生产可用级别的代码。还有自动生成的交互式文档标准化基于并完全兼容API 的相关开放标准OpenAPI (以前被称为 Swagger) 和 JSON Schema。
FastAPI 最大的特点就是它使用了 Python 的类型注解使用 FastAPI 需要 Python 版本大于等于 3.6。 安装 FastAPI
pip install fastapi会自动安装 Starlette 和 Pydantic然后还要 pip install uvicorn因为 uvicorn 是运行相关应用程序的服务器。或者一步到胃pip install fastapi[all]会将所有依赖全部安装。 教 程
官网教程https://fastapi.tiangolo.com/zh/tutorial/ Python 高性能 web 框架 - FastApi 全面指南https://zhuanlan.zhihu.com/p/397029492 最受欢迎的异步框架https://www.cnblogs.com/traditional/p/14733610.html
高级用户指南https://fastapi.tiangolo.com/zh/advanced/ 第一步路径参数查询参数请求体查询参数和字符串校验路径参数和数值校验请求体 - 多个参数请求体 - 字段请求体 - 嵌套模型模式的额外信息 - 例子额外数据类型Cookie 参数Header 参数响应模型额外的模型响应状态码表单数据请求文件请求表单与文件处理错误路径操作配置JSON 兼容编码器请求体 - 更新数据 依赖项 安全性 中间件CORS跨域资源共享SQL (关系型) 数据库更大的应用 - 多个文件后台任务元数据和文档 URL静态文件测试调试 Python 类型提示简介
https://fastapi.tiangolo.com/zh/python-types/
Python类型注解你需要知道的都在这里https://zhuanlan.zhihu.com/p/419955374 对象注解属性的最佳实践https://docs.python.org/zh-cn/3/howto/annotations.html 2、FastAPI 使用 简单使用示例 示例同步 代码
from typing import Optional
from fastapi import FastAPIapp FastAPI()app.get(/)
def read_root():return {Hello: World}app.get(/items/{item_id})
def read_item(item_id: int, q: Optional[str] None):return {item_id: item_id, q: q}
示例异步 代码
新建一个 main.py 文件
import os
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp FastAPI()app.get(/)
async def root():return {message: Hello World}if __name__ __main__:# os.system(uvicorn main_test:app --reload)uvicorn.run(f{Path(__file__).stem}:app, host0.0.0.0, port5555)pass
FastAPI 推荐使用 uvicorn 来运行服务Uvicorn 是基于uvloop 和 httptools 构建的闪电般快速的 ASGI 服务器。
uvicorn main:app 命令含义如下:
mainmain.py 文件一个 Python「模块」。app在 main.py 文件中通过 app FastAPI() 创建的对象。--reload让服务器在更新代码后重新启动。仅在开发时使用该选项。
访问 http://127.0.0.1:5555 可以看到 JSON 格式的响应{message: Hello World} 请求 查询 参数
「请求路径」也通常被称为「端点」或「路由」。
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp FastAPI() # 创建 api 对象fake_items_db [{item_name: Foo}, {item_name: Bar}, {item_name: Baz}]app.get(/items/)
async def read_item(skip: int 0, limit: int 10):return fake_items_db[skip: skip limit]app.get(/) # 根路由。「请求路径」也通常被称为「端点」或「路由」。
async def root():return {name: king, age: 100}app.get(/say/{data})
async def say(data: str, q: int None):return {data: data, q: q}if __name__ __main__:# os.system(uvicorn main_test:app --reload)uvicorn.run(f{Path(__file__).stem}:app, host127.0.0.1, port5555)pass
浏览器访问 http://127.0.0.1:5555/ http://127.0.0.1:5555/say/test http://127.0.0.1:5555/items/?skip0limit10 该查询是 ? URL中位于关键字之后的一组键值对以字符分隔。 在 url 中进行查询 skip查询的起始参数 limit查询的结束参数 关于 app.get()、router.get() 区别
都用于定义 HTTP GET 请求的处理程序但存在一些区别。
app.get(): 这个方法是在 FastAPI 应用程序实例上直接定义路由的一种简写方式。
from fastapi import FastAPIapp FastAPI()app.get(/)
async def root():return {message: Hello, World!}
router.get(): 这个方法是在 FastAPI 路由器对象Router上定义路由的一种方式。你可以创建一个独立的路由器对象然后使用 router.get() 来定义特定路径的路由。
from fastapi import FastAPI, APIRouterrouter APIRouter()router.get(/items)
async def read_items():return {message: Read all items}
上面示例创建一个名为 router 的路由器对象并使用 router.get() 定义了 /items 路径的 GET 请求处理程序。需要注意的是如果你使用了多个路由器对象最终需要将它们添加到 FastAPI 应用程序实例中才能生效。 # 通过 app.include_router() 方法将路由器对象添加到 FastAPI 应用程序实例中这样它的定义的路由才会被正确处理。 app.include_router(router) 总结app.get() 用于在应用程序实例上定义路由而 router.get() 是在路由器对象上定义路由。两者的区别在于定义位置和应用方式但它们都可以用于处理 HTTP GET 请求。 总 结
导入 FastAPI。创建一个 app 实例。编写一个路径操作装饰器如 app.get(/)。编写一个路径操作函数如上面的 def root(): ...。运行开发服务器如 uvicorn main:app --reload。 请求的 路径 参数
路径参数是必须要体现在参数中。
「请求路径」也通常被称为「端点」或「路由」。
FastAPI 编写一个简单的应用程序
import os
import uvicorn
from pathlib import Path
from fastapi import FastAPI# 类似于 app Flask(__name__)
app FastAPI()# 绑定路由和视图函数
app.get(/)
async def root():return {message: Hello World}# 在 Windows 中必须加上 if __name__ __main__
# 否则会抛出 RuntimeError: This event loop is already running
if __name__ __main__:# 启动服务因为我们这个文件叫做 main_test.py所以需要启动 main_test.py 里面的 app# os.system(uvicorn main_test:app --reload)# uvicorn.run(main_test:app, host0.0.0.0, port8000)uvicorn.run(f{Path(__file__).stem}:app, host0.0.0.0, port5555)pass
在浏览器中输入 localhost:5555 就会显示相应的输出我们看到在视图函数中可以直接返回一个字典。当然除了字典其它的数据类型也是可以的。
# -*- coding:utf-8 -*-
from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/int)
async def index1():return 666app.get(/str)
async def index2():return 佛祖保佑,佛祖保佑app.get(/bytes)
async def index3():return bsatoriapp.get(/tuple)
async def index4():temp_tuple (佛祖保佑, 佛祖保佑)return temp_tupleapp.get(/list)
async def index5():return [{name: 擒贼先擒王, age: 18}, {name: 捉奸先捉双, age: 16}]if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)
直接使用 requests 发送请求 import requests print(requests.get(http://localhost:5555/int).text) print(requests.get(http://localhost:5555/str).text) print(requests.get(http://localhost:5555/bytes).text) print(requests.get(http://localhost:5555/tuple).text) print(requests.get(http://localhost:5555/list).text) 不过元组自动转成列表返回了。这里我们在路由中指定了路径可以看到 FastAPI 中的路径形式和其它框架并无二致只不过目前的路径是写死的如果我们想动态声明路径参数该怎么做呢 路径 参数
可以使用与 Python 格式化字符串相同的语法 来声明路径 参数 或 变量FastAPI 足以辨别 路径参数 和 查询参数。 from fastapi import FastAPI app FastAPI() app.get(/items/{item_id}) async def read_item(item_id): return {item_id: item_id} 路径参数 item_id 的值将作为参数 item_id 传递给你的函数。运行并访问 http://127.0.0.1:5555/items/foo 将会看到响应{item_id:foo} 示例
from fastapi import FastAPIapp FastAPI()app.get(/items/{item_id})
async def read_item(item_id: str, q: str None, short: bool False):item {item_id: item_id}if q:item.update({q: q})if not short:item.update({description: This is an amazing item that has a long description})return item路径参数 item_id 的值将作为参数 item_id 传递给你的函数。声明不属于路径参数的其他函数参数时它们将被自动解释为 查询字符串 参数
看看其访问路径执行以下的任何一种 url 访问方式 http://127.0.0.1:8000/items/老王睡隔壁?short1 http://127.0.0.1:8000/items/老王睡隔壁?shortTrue http://127.0.0.1:8000/items/老王睡隔壁?shorttrue http://127.0.0.1:8000/items/老王睡隔壁?shorton http://127.0.0.1:8000/items/老王睡隔壁?shortyes 可以发现任何大小写的字母等都会被转换成 bool 值的参数 True这就是所谓模糊验证参数对于开发者来说是个好消息。注意如果 short 参数没有默认值则必须传参否则 FastAPI 将报错。 { detail: [ { loc: [ query, needy ], msg: field required, type: value_error.missing } ] } 示例
from fastapi import FastAPIapp FastAPI()fake_items_db [{item_name: Foo}, {item_name: Bar}, {item_name: Baz}]app.get(/items/)
async def read_item(skip: int 0, limit: int 10):return fake_items_db[skip : skip limit]
查询字符串是键值对的集合这些键值对位于 URL 的 ? 之后并以 符号 分隔。
可以使用 Query 对查询进行额外的校验
from typing import Optionalfrom fastapi import FastAPI, Queryapp FastAPI()app.get(/items/)
async def read_items(q: Optional[str] Query(None, max_length50)):results {items: [{item_id: Foo}, {item_id: Bar}]}if q:results.update({q: q})return results
Query 有如下这些字段校验
min_length 最小长度max_length 最大长度regex 正则匹配Query 第一个参数为默认值...表示是必需的
Path 和 Query 用法一样也能对查询字段进行校验。
而且你还可以声明数值校验
from fastapi import FastAPI, Path, Queryapp FastAPI()app.get(/items/{item_id})
async def read_items(*,item_id: int Path(..., titleThe ID of the item to get, ge0, le1000),q: str,size: float Query(..., gt0, lt10.5)
):results {item_id: item_id}if q:results.update({q: q})return resultsgt大于ge大于等于lt小于le小于等于
类似的还有 Cookie
from typing import Optionalfrom fastapi import Cookie, FastAPIapp FastAPI()app.get(/items/)
async def read_items(ads_id: Optional[str] Cookie(None)):return {ads_id: ads_id}以及 Header
from typing import Optionalfrom fastapi import FastAPI, Headerapp FastAPI()app.get(/items/)
async def read_items(user_agent: Optional[str] Header(None)):return {User-Agent: user_agent}请求主体路径查询参数在请求主体的基础上加入 url 动态路径参数 和 查询参数
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str Noneprice: floattax: float Noneapp FastAPI()app.put(/items/{item_id})
async def create_item(item_id: int, item: Item, q: str None):result {item_id: item_id, **item.dict()}if q:result.update({q: q})return result关于模板引擎
FastAPI 不像 Flask 那样自带 模板引擎(Jinja2)也就是说没有默认的模板引擎从另一个角度上说FastAPI 在模板引擎的选择上变得更加灵活极度舒适。 以 Jinja2 模板为例安装依赖 pip install jinja2 pip install aiofiles # 用于 fastapi 的异步静态文件 # -*- coding:utf-8 -*-
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import uvicornapp FastAPI()app.mount(/static, StaticFiles(directorystatic), namestatic) # 挂载静态文件指定目录templates Jinja2Templates(directorytemplates) # 模板目录app.get(/data/{data})
async def read_data(request: Request, data: str):return templates.TemplateResponse(index.html, {request: request, data: data})if __name__ __main__:uvicorn.run(app, host127.0.0.1, port8000)html 文件渲染 html head title士可杀不可辱(you can kill me, but you cant fuck me)/title /head body h1高呼: {{ data }}/h1 /body /html 在浏览器键入 http://127.0.0.1:8000/data/士可杀不可辱
值得注意的是在返回的 TemplateRespone 响应时必须带上 request 的上下文对象传入参数放在同一字典。这样一来又可以像 Flask 一样的使用 Jinja2。 为路径设置 tags 标签进行分组
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp FastAPI()class Item(BaseModel):name: strdescription: Optional[str] Noneprice: floattax: Optional[float] Nonetags: Set[str] []app.post(/items/, response_modelItem, tags[items])
async def create_item(item: Item):return itemapp.get(/items/, tags[items])
async def read_items():return [{name: Foo, price: 42}]app.get(/users/, tags[users])
async def read_users():return [{username: johndoe}]还可以设置 summary 和 description:
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp FastAPI()class Item(BaseModel):name: strdescription: Optional[str] Noneprice: floattax: Optional[float] Nonetags: Set[str] []app.post(/items/,response_modelItem,summaryCreate an item,descriptionCreate an item with all the information, name, description, price, tax and a set of unique tags,
)
async def create_item(item: Item):return item多行注释
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp FastAPI()class Item(BaseModel):name: strdescription: Optional[str] Noneprice: floattax: Optional[float] Nonetags: Set[str] []app.post(/items/, response_modelItem, summaryCreate an item)
async def create_item(item: Item):Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesnt have tax, you can omit this- **tags**: a set of unique tag strings for this itemreturn item 废弃路由
from fastapi import FastAPIapp FastAPI()app.get(/items/, tags[items])
async def read_items():return [{name: Foo, price: 42}]app.get(/users/, tags[users])
async def read_users():return [{username: johndoe}]app.get(/elements/, tags[items], deprecatedTrue)
async def read_elements():return [{item_id: Foo}]有类型的路径参数
可以使用标准的 Python 类型标注为函数中的路径参数声明类型。 from fastapi import FastAPI app FastAPI() app.get(/items/{item_id}) async def read_item(item_id: int): return {item_id: item_id} 在这个例子中item_id 被声明为 int 类型。
运行示例并打开浏览器访问 http://127.0.0.1:5555/items/3 将得到如下响应{item_id:3} 注意函数接收并返回的值为 3是一个 Python int 值而不是字符串 3。 所以FastAPI 通过上面的类型声明提供了对请求的自动解析。
# -*- coding:utf-8 -*-from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/apple/{item_id})
async def get_item(item_id: int):Flask 定义类型是在路由当中也就是在 里面变量和类型通过 : 分隔FastAPI 是使用类型注解的方式此时的 item_id 要求一个整型准确的说是一个能够转成整型的字符串return {item_id: item_id}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)顺序很重要
由于路径操作是按顺序依次运行的所以在定义路由的时候需要注意一下顺序 from fastapi import FastAPI app FastAPI() app.get(/users/me) async def read_user_me(): return {user_id: the current user} app.get(/users/{user_id}) async def read_user(user_id: str): return {user_id: user_id} 因为路径操作是按照顺序进行的所以这里要保证 /users/me 在 /users/{user_id} 的前面否则的话只会匹配到 /users/{user_id}此时如果访问 /users/me那么会返回一个解析错误因为字符串 me 无法解析成整型。 预设值 ( 枚举 )
可以将某个路径参数通过类型注解的方式声明为指定的类型 准确的说是可以转成指定的类型因为默认都是字符串 但如果我们希望它只能是我们规定的几个值之一该怎么做呢
这时可以使用标准的 Python Enum 类型。导入 Enum 并创建一个继承自 str 和 Enum 的子类。通过从 str 继承API 文档将能够知道这些值必须为 string 类型并且能够正确地展示出来。然后创建具有固定值的类属性这些固定值将是可用的有效值
from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet alexnetresnet resnetlenet lenetapp FastAPI()app.get(/models/{model_name})
async def get_model(model_name: ModelName):if model_name is ModelName.alexnet:return {model_name: model_name, message: Deep Learning FTW!}if model_name.value lenet:return {model_name: model_name, message: LeCNN all the images}return {model_name: model_name, message: Have some residuals}路径中包含 /
假设有这样一个路由/files/{file_path}而用户传递的 file_path 中显然是可以带 / 的假设 file_path 是 /root/test.py那么路由就变成了 /files//root/test.py显然这是有问题的。
OpenAPI 不支持任何方式去声明路径参数以在其内部包含路径因为这可能会导致难以测试和定义的情况出现。不过仍然可以通过 Starlette 的一个内部工具在 FastAPI 中实现它。
可以使用直接来自 Starlette 的选项来声明一个包含路径的路径参数/files/{file_path:path}
在这种情况下参数的名称为 file_path结尾部分的 :path 说明该参数应匹配任意的路径。
因此你可以这样使用它 from fastapi import FastAPI app FastAPI() app.get(/files/{file_path:path}) async def read_file(file_path: str): return {file_path: file_path} 你可能会需要参数包含 /home/johndoe/myfile.txt以斜杠/开头。
在这种情况下URL 将会是 /files//home/johndoe/myfile.txt在files 和 home 之间有一个双斜杠//。
from fastapi import FastAPI
import uvicornapp FastAPI()# 声明 file_path 的类型为 path这样它会被当成一个整体
app.get(/files/{file_path:path})
async def get_file(file_path: str):return {file_path: file_path}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)设置 127.0.0.1:8000/docs
至于 127.0.0.1:8000/docs 页面本身也是可以进行设置的
from fastapi import FastAPI
import uvicornapp FastAPI(title测试文档,description这是一个简单的 demo,docs_url/my_docs,openapi_url/my_openapi
)app.get(/apple/{item_id})
async def get_item(item_id: int):return {item_id: item_id}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)路径参数、数据校验
查询参数数据校验使用的是 Query路径参数数据校验使用的是 Path两者的使用方式一模一样没有任何区别
from fastapi import FastAPI, Path
import uvicornapp FastAPI()app.get(/items/{item-id})
async def read_items(item_id: int Path(..., aliasitem-id)):return {item_id: item_id}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)
因为路径参数是必须的它是路径的一部分所以我们应该使用 ... 将其标记为必传参数。当然即使不这么做也无所谓因为指定了默认值也用不上因为路径参数不指定压根就匹配不到相应的路由。至于一些其它的校验和查询参数一模一样所以这里不再赘述了。
不过我们之前说过路径参数应该在查询参数的前面尽管 FastAPI 没有这个要求但是这样写明显更舒服一些。但是问题来了如果路径参数需要指定别名但是某一个查询参数不需要这个时候就会出现问题
app.get(/items/{item-id})
async def read_items(q: str,item_id: int Path(..., aliasitem-id)):return {item_id: item_id, q: q}
显然此时 Python 的语法就决定了 item_id 就必须放在 q 的后面当然这么做是完全没有问题的FastAPI 对参数的先后顺序没有任何要求因为它是通过参数的名称、类型和默认值声明来检测参数而不在乎参数的顺序。但此时我们就要让 item_id 在 q 的前面要怎么做呢
app.get(/items/{item-id})
async def read_items(*, item_id: int Path(..., aliasitem-id),q: str):return {item_id: item_id, q: q}
此时就没有问题了通过将第一个参数设置为 *使得 item_id 和 q 都必须通过关键字传递所以此时默认参数在非默认参数之前也是允许的。当然我们也不需要担心 FastAPI 传参的问题你可以认为它所有的参数都是通过关键字参数的方式传递的。 请求的 查询 参数 查询参数在 FastAPI 中可以通过类型注解的方式进行声明查询参数也可以不写那么请求相关的所有信息都会进入到这个 Request 对象中如果函数中定义了不属于路径参数的参数时那么它们将会被自动解释会查询参数。
示例 from fastapi import FastAPI app FastAPI() fake_items_db [{item_name: Foo}, {item_name: Bar}, {item_name: Baz}] app.get(/items/) async def read_item(skip: int 0, limit: int 10): return fake_items_db[skip : skip limit] 查询字符串是键值对的集合这些键值对位于 URL 的 之后并以 符号分隔。
例如在以下 url 中http://127.0.0.1:5555/items/?skip0limit10
查询参数为
skip对应的值为 0limit对应的值为 10
示例
from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/user/{user_id})
async def get_user(user_id: str, name: str, age: int):函数中参数定义了 user_id、name、age 三个参数显然 user_id 和 路径参数中的 user_id 对应然后 name 和 age 会被解释成查询参数这三个参数的顺序没有要求但是一般都是路径参数在前查询参数在后return {user_id: user_id, name: name, age: age}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)可选参数、必需参数
当 非路径参数 默认值设置为 None则参数是可选的当 非路径参数 没有默认值 时则参数是必需传递的
将它们的默认值设置为 None 来声明可选查询参数 from typing import Union from fastapi import FastAPI app FastAPI() app.get(/items/{item_id}) async def read_item(item_id: str, q: Union[str, None] None): if q: return {item_id: item_id, q: q} return {item_id: item_id} 在这个例子中函数参数 q 默认值为 None所以q可选的。FastAPI 能够分辨出参数 item_id 是路径参数而 q 不是因此 q 是一个查询参数。 多类型 参数 ( Union )
指定多个类型。比如 user_id 按照整型解析、解析不成功退化为字符串。
from typing import Union, Optional
from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/user/{user_id})
async def get_user(user_id: Union[int, str], name: Optional[str] None):通过 Union 来声明一个混合类型int 在前、str 在后。会先按照 int 解析解析失败再变成 str然后是 name它表示字符串类型、但默认值为 None不是字符串那么应该声明为 Optional[str]return {user_id: user_id, name: name}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)所以 FastAPI 的设计还是非常不错的通过 Python 的类型注解来实现参数类型的限定可以说是非常巧妙的因此这也需要我们熟练掌握 Python 的类型注解。 bool 类型自动转换
对于布尔类型FastAPI 支持自动转换
from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/{flag})
async def get_flag(flag: bool):return {flag: flag}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 print(requests.get(http://localhost:5555/1).json()) print(requests.get(http://localhost:5555/True).json()) print(requests.get(http://localhost:5555/true).json()) print(requests.get(http://localhost:5555/False).json()) print(requests.get(http://localhost:5555/false).json()) print(requests.get(http://localhost:5555/on).json()) print(requests.get(http://localhost:5555/yes).json()) print(requests.get(http://localhost:5555/off).json()) print(requests.get(http://localhost:5555/no).json()) 多个路径、查询参数
可以同时声明多个路径参数和查询参数FastAPI 能够识别它们。而且你不需要以任何特定的顺序来声明。它们将通过名称被检测到 from typing import Union from fastapi import FastAPI app FastAPI() app.get(/users/{user_id}/items/{item_id}) async def read_user_item( user_id: int, item_id: str, q: Union[str, None] None, short: bool False ): item {item_id: item_id, owner_id: user_id} if q: item.update({q: q}) if not short: item.update( {description: This is an amazing item that has a long description} ) return item FastAPI 可以定义任意个路径参数只要动态的路径参数在函数的参数中都出现即可。当然查询参数也可以是任意个FastAPI 可以处理的很好。
from typing import Optional
from fastapi import FastAPI
import uvicornapp FastAPI()app.get(/postgres/{schema}/v1/{table})
async def get_data(schema: str,table: str,select: str *,where: Optional[str] None,limit: Optional[int] None,offset: Optional[int] None):标准格式是路径参数按照顺序在前查询参数在后但其实对顺序是没有什么要求的query fselect {select} from {schema}.{table}if where:query f where {where}if limit:query f limit {limit}if offset:query f offset {offset}return {query: query}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555) http://localhost:5555/postgres/ods/v1/staff http://localhost:5555/postgres/ods/v1/staff?selectid, namewhereid 3limit100 Depends ( 依赖注入 )
from typing import Optional
import uvicorn
from fastapi import Depends, FastAPIapp FastAPI()async def common_parameters(q: Optional[str] None, skip: int 0, limit: int 100):return {q: q, skip: skip, limit: limit}app.get(/items/)
async def read_items(commons: dict Depends(common_parameters)):# common_parameters 接收三个参数q、skip、limit# 然后在解析请求的时候会将 q、skip、limit 传递到 common_parameters 中然后将返回值赋值给 commons# 但如果解析不到某个参数时那么会判断函数中参数是否有默认值没有的话就会返回错误而不是传递一个 None 进去return commonsapp.get(/users/)
async def read_users(commons: dict Depends(common_parameters)):return commonsif __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 requests.get(http://localhost:5555/items).json() {q: Noneskip:limit: 100} requests.get(http://localhost:5555/items?gidname).json() {g:id, nameskip: 0limit: 100} 所以 Depends 能够很好的实现依赖注入而且我们特意写了两个路由就是想表示它们是彼此独立的。因此当有共享的逻辑、或者共享的数据库连接、增强安全性、身份验证、角色权限等等会非常的实用。
FastAPI 提供了简单易用但功能强大的依赖注入系统可以让开发人员轻松地把组件集成至FastAPI。
什么是「依赖注入」
依赖注入是一种消除类之间依赖关系的设计模式。把有依赖关系的类放到容器中解析出这些类的实例就是依赖注入。目的是实现类的解耦。
示例
from typing import Optional
from fastapi import Depends, FastAPIapp FastAPI()async def common_parameters(q: Optional[str] None, skip: int 0, limit: int 100):return {q: q, skip: skip, limit: limit}app.get(/items/)
async def read_items(commons: dict Depends(common_parameters)):return commonsapp.get(/users/)
async def read_users(commons: dict Depends(common_parameters)):return commons本例中的依赖项预期接收如下参数
类型为 str 的可选查询参数 q类型为 int 的可选查询参数 skip默认值是 0类型为 int 的可选查询参数 limit默认值是 100
然后依赖项函数返回包含这些值的 dict。
使用Class作为依赖
from typing import Optional
from fastapi import Depends, FastAPIapp FastAPI()fake_items_db [{item_name: Foo}, {item_name: Bar}, {item_name: Baz}]class CommonQueryParams:def __init__(self, q: Optional[str] None, skip: int 0, limit: int 100):self.q qself.skip skipself.limit limitapp.get(/items/)
async def read_items(commons: CommonQueryParams Depends(CommonQueryParams)):response {}if commons.q:response.update({q: commons.q})items fake_items_db[commons.skip: commons.skip commons.limit]response.update({items: items})return response使用嵌套子依赖
from typing import Optional
from fastapi import Cookie, Depends, FastAPIapp FastAPI()def query_extractor(q: Optional[str] None):return qdef query_or_cookie_extractor(q: str Depends(query_extractor), last_query: Optional[str] Cookie(None)
):if not q:return last_queryreturn qapp.get(/items/)
async def read_query(query_or_default: str Depends(query_or_cookie_extractor)):return {q_or_cookie: query_or_default}在路径中使用依赖
from fastapi import Depends, FastAPI, Header, HTTPExceptionapp FastAPI()async def verify_token(x_token: str Header(...)):if x_token ! fake-super-secret-token:raise HTTPException(status_code400, detailX-Token header invalid)async def verify_key(x_key: str Header(...)):if x_key ! fake-super-secret-key:raise HTTPException(status_code400, detailX-Key header invalid)return x_keyapp.get(/items/, dependencies[Depends(verify_token), Depends(verify_key)])
async def read_items():return [{item: Foo}, {item: Bar}]全局依赖项可以为所有路径操作应用该依赖项
from fastapi import Depends, FastAPI, Header, HTTPExceptionasync def verify_token(x_token: str Header(...)):if x_token ! fake-super-secret-token:raise HTTPException(status_code400, detailX-Token header invalid)async def verify_key(x_key: str Header(...)):if x_key ! fake-super-secret-key:raise HTTPException(status_code400, detailX-Key header invalid)return x_keyapp FastAPI(dependencies[Depends(verify_token), Depends(verify_key)])app.get(/items/)
async def read_items():return [{item: Portal Gun}, {item: Plumbus}]app.get(/users/)
async def read_users():return [{username: Rick}, {username: Morty}]查询参数、数据校验
FastAPI 支持我们进行更加智能的数据校验比如一个字符串我们希望用户在传递的时候只能传递长度为 6 到 15 的字符串该怎么做呢
from typing import Optional
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/user)
async def check_length(# 默认值为 None应该声明为 Optional[str]当然声明 str 也是可以的。只不过声明为 str那么默认值应该也是 str# 所以如果一个类型允许为空那么更规范的做法应该是声明为 Optional[类型]。password: Optional[str] Query(None, min_length6, max_length15)
):return {password: password}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)password 是可选的但是一旦传递则必须传递字符串、而且还是长度在 6 到 15 之间的字符串。所以如果传递的是 None那么在声明默认值的时候 None 和 Query(None) 是等价的只不过 Query 还支持其它的参数来对参数进行限制。 requests.get(http://localhost:5555/user?password12345).json() requests.get(http://localhost:5555/user?password123456).json() Query 里面除了限制最小长度和最大长度还有其它的功能
查询参数和字符串校验 from fastapi import FastAPI app FastAPI() app.get(/items/) async def read_items(q: str | None None): results {items: [{item_id: Foo}, {item_id: Bar}]} if q: results.update({q: q}) return results 查询参数 q 的类型为 str默认值为 None因此它是可选的。
额外的校验
添加约束条件即使 q 是可选的但只要提供了该参数则该参数值不能超过50个字符的长度。
从 fastapi 导入 Query from typing import Union from fastapi import FastAPI, Query app FastAPI() app.get(/items/) async def read_items(q: Union[str, None] Query(defaultNone, max_length50)): results {items: [{item_id: Foo}, {item_id: Bar}]} if q: results.update({q: q}) return results 添加更多校验
还可以添加 min_length 参数 from typing import Union from fastapi import FastAPI, Query app FastAPI() app.get(/items/) async def read_items( q: Union[str, None] Query(defaultNone, min_length3, max_length50) ): results {items: [{item_id: Foo}, {item_id: Bar}]} if q: results.update({q: q}) return results 添加正则表达式
可以定义一个参数值必须匹配的正则表达式 from typing import Union from fastapi import FastAPI, Query app FastAPI() app.get(/items/) async def read_items( q: Union[str, None] Query( defaultNone, min_length3, max_length50, pattern^fixedquery$ ) ): results {items: [{item_id: Foo}, {item_id: Bar}]} if q: results.update({q: q}) return results 声明为必须参数
... 是 Python 中的一个特殊的对象通过它可以实现该参数是必传参数。
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/user)
async def check_length(password: str Query(..., min_length6)):将第一个参数换成 ... 即可实现该参数是必传参数return {password: password}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)查询参数变成一个列表
如果我们指定了 a1a2那么我们在获取 a 的时候如何才能得到一个列表呢
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/items)
async def read_items(a1: str Query(...),a2: List[str] Query(...),b: List[str] Query(...)
):return {a1: a1, a2: a2, b: b}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)首先 a2 和 b 都是对应列表然后 a1 只获取了最后一个值。另外可能有人觉得我们这样有点啰嗦在函数声明中可不可以这样写呢
app.get(/items)
async def read_items(a1: str,a2: List[str],b: List[str]
):return {a1: a1, a2: a2, b: b}
对于 a1 是可以的但是 a2 和 b 不行。对于类型为 list 的查询参数无论有没有默认值你都必须要显式的加上 Query 来表示必传参数。如果允许为 None或者有默认值的话那么应该这么写
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/items)
async def read_items(a1: str,a2: Optional[List[str]] Query(None),b: List[str] Query([1, 嘿嘿])
):return {a1: a1, a2: a2, b: b}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555) 给参数起别名
假设我们定义的查询参数名叫 item-query那么由于它要体现在函数参数中、而这显然不符合 Python 变量的命名规范这个时候要怎么做呢答案是起一个别名。
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/items)
async def read_items(# 通过 url 的时候使用别名即可item1: Optional[str] Query(None, aliasitem-query),item2: str Query(哈哈, alias),item3: str Query(..., alias$$$$) # item3 是必传的
):return {item1: item1, item2: item2, item3: item3}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)数值检测
Query 不仅仅支持对字符串的校验还支持对数值的校验里面可以传递 gt、ge、lt、le 这几个参数相信这几个参数不用说你也知道是干什么的我们举例说明
from fastapi import FastAPI, Query
import uvicornapp FastAPI()app.get(/items)
async def read_items(# item1 必须大于 5item1: int Query(..., gt5),# item2 必须小于等于 7item2: int Query(..., le7),# item3 必须必须等于 10item3: int Query(..., ge10, le10)
):return {item1: item1, item2: item2, item3: item3}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 Request 对象
Request 是什么首先我们知道任何一个请求都对应一个 Request 对象请求的所有信息都在这个 Request 对象中FastAPI 也不例外。 路径参数是必须要体现在参数中但是查询参数可以不写了 因为我们定义了 request: Request那么请求相关的所有信息都会进入到这个 Request 对象中 from fastapi import FastAPI, Request
import uvicornapp FastAPI()app.get(/girl/{user_id})
async def read_girl(user_id: str,request: Request):路径参数是必须要体现在参数中但是查询参数可以不写了因为我们定义了 request: Request那么请求相关的所有信息都会进入到这个 Request 对象中header request.headers # 请求头method request.method # 请求方法cookies request.cookies # cookiesquery_params request.query_params # 查询参数return {name: query_params.get(name), age: query_params.get(age), hobby: query_params.getlist(hobby)}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)
通过 Request 对象可以获取所有请求相关的信息我们之前当参数传递不对的时候FastAPI 会自动帮我们返回错误信息但通过 Request 我们就可以自己进行解析、自己指定返回的错误信息了。 Response 对象
既然有 Request那么必然会有 Response尽管我们可以直接返回一个字典但 FastAPI 实际上会帮我们转成一个 Response 对象。
Response 内部接收如下参数
content返回的数据status_code状态码headers返回的请求头media_type响应类型(就是 HTML 中 Content-Type只不过这里换了个名字)background接收一个任务Response 在返回之后会自动异步执行
示例
from fastapi import FastAPI, Request, Response
import uvicorn
import orjsonapp FastAPI()app.get(/girl/{user_id})
async def read_girl(user_id: str, request: Request):query_params request.query_params # 查询参数data {name: query_params.get(name), age: query_params.get(age), hobby: query_params.getlist(hobby)}# 实例化一个 Response 对象response Response(# content我们需要手动转成 json 字符串如果直接返回字典的话那么在包装成 Response 对象的时候会自动帮你转orjson.dumps(data),# status_code状态码201,# headers响应头{Token: xxx},# media_type就是 HTML 中的 Content-Typeapplication/json,)# 如果想设置 cookie 的话那么通过 response.set_cookie 即可# 删除 cookie 则是 response.delete_cookiereturn responseif __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)通过 Response 我们可以实现请求头、状态码、cookie 等自定义。
另外除了 Response 之外还有很多其它类型的响应它们都在 fastapi.responses 中比如FileResponse、HTMLResponse、PlainTextResponse 等等。它们都继承了 Response只不过会自动帮你设置响应类型举个栗子
from fastapi import FastAPI
from fastapi.responses import Response, HTMLResponse
import uvicornapp FastAPI()app.get(/index)
async def index():response1 HTMLResponse(h1你好呀/h1)response2 Response(h1你好呀/h1, media_typetext/html)# 以上两者是等价的在 HTMLResponse 中会自动将 media_type 设置成 text/htmlreturn response1if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)使用 response_model参数来声明用于响应的模型
from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModelapp FastAPI()class Item(BaseModel):name: strdescription: Optional[str] Noneprice: floattax: Optional[float] Nonetags: List[str] []app.post(/items/, response_modelItem)
async def create_item(item: Item):return itemresponse_model_exclude_unsetTrue响应中将不会包含那些默认值而是仅有实际设置的值response_model_include 包含哪些属性response_model_exclude 省略某些属性
status_code参数来声明用于响应的 HTTP 状态码
from fastapi import FastAPIapp FastAPI()app.post(/items/, status_code201)
async def create_item(name: str):return {name: name}表单字段时要使用Form
from fastapi import FastAPI, Formapp FastAPI()app.post(/login/)
async def login(username: str Form(...), password: str Form(...)):return {username: username}File用于定义客户端的上传文件接收上传文件要预先安装python-multipart
from fastapi import FastAPI, File, UploadFileapp FastAPI()app.post(/files/)
async def create_file(file: bytes File(...)):return {file_size: len(file)}app.post(/uploadfile/)
async def create_upload_file(file: UploadFile File(...)):return {filename: file.filename}向客户端返回 HTTP 错误响应可以使用HTTPException。
from fastapi import FastAPI, HTTPExceptionapp FastAPI()items {foo: The Foo Wrestlers}app.get(/items/{item_id})
async def read_item(item_id: str):if item_id not in items:raise HTTPException(status_code404, detailItem not found)return {item: items[item_id]}使用response_description设置响应描述
from typing import Optional, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp FastAPI()class Item(BaseModel):name: strdescription: Optional[str] Noneprice: floattax: Optional[float] Nonetags: Set[str] []app.post(/items/,response_modelItem,summaryCreate an item,response_descriptionThe created item,
)
async def create_item(item: Item):Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesnt have tax, you can omit this- **tags**: a set of unique tag strings for this itemreturn item请求体 当需要将数据从客户端例如浏览器发送给 API 时是将其作为「请求体」发送。 请求体是客户端发送给 API 的数据。 响应体是 API 发送给客户端的数据。 要发送数据你必须使用下列方法之一POST较常见、PUT、DELETE 或 PATCH。 显然对应 POST、PUT 等类型的请求我们必须要能够解析出请求体并且能够构造出响应体。
Model
在 FastAPI 中请求体和响应体都对应一个 Model
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp FastAPI()class Girl(BaseModel):数据验证是通过 pydantic 实现的我们需要从中导入 BaseModel然后继承它name: strage: Optional[str] Nonelength: floathobby: List[str] # 对于 Model 中的 List[str] 我们不需要指定 Query准确的说是 Fieldapp.post(/girl)
async def read_girl(girl: Girl):# girl 就是我们接收的请求体它需要通过 json 来传递并且这个 json 要有上面的四个字段age 可以没有# 通过 girl.xxx 的方式我们可以获取和修改内部的所有属性return dict(girl) # 直接返回 Model 对象也是可以的if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 示例 from fastapi import FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None None price: float tax: float | None None app FastAPI() app.post(/items/) async def create_item(item: Item): item_dict item.dict() if item.tax: price_with_tax item.price item.tax item_dict.update({price_with_tax: price_with_tax}) return item_dict 和声明查询参数时一样当一个模型属性具有默认值时它不是必需的。否则它是一个必需属性。将默认值设为 None 可使其成为可选属性。 使用 Request 对象 传递 body
如果你不想使用 Pydantic 模型你还可以使用 Body 参数。请参阅文档 请求体 - 多个参数请求体中的单一值。
from fastapi import FastAPI, Request
import uvicornapp FastAPI()app.post(/girl)
async def read_girl(request: Request):# 是一个协程所以需要 awaitdata await request.body()print(data)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)使用 requests 模块发送 post 请求的时候可以通过 data 参数传递、也可以通过 json 参数。
当通过 json{name: satori, age: 16, length: 155.5} 传递的时候会将其转成 json 字符串进行传输程序中的 print 打印如下b{name: satori, age: 16, length: 155.5}如果用 data 参数发请求的话值不变那么会将其拼接成 k1v1k2v2 的形式再进行传输相当于表单提交后面说程序中打印如下bnamesatoriage16length155.5
所以我们看到 await request.body() 得到的就是最原始的字节流而除了 await request.body() 之外还有一个 await request.json()只是后者在内部在调用了前者拿到字节流之后、自动帮你 loads 成了字典。因此使用 await request.json() 也侧面要求我们必须在发送请求的时候必须使用 json 参数传递传递的是字典转成的 json、所以也能解析成字典否则使用 await request.json() 是无法正确解析的。 路径参数、查询参数、请求体
如果在 路径 中也声明了该参数它将被用作路径参数。如果参数属于 单一类型比如 int、float、str、bool 等它将被解释为 查询 参数。如果参数的类型被声明为一个 Pydantic 模型它将被解释为 请求体。
# -*- coding:utf-8 -*-
# Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp FastAPI()class Girl(BaseModel):name: strage: Optional[str] Nonelength: floathobby: List[str]app.post(/girl/{user_id})
async def read_girl(user_id, q: str, girl: Girl):return {user_id: user_id, q: q, **dict(girl)}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 指定了路径参数、查询参数和请求体FastAPI 依然是可以正确区分的当然我们也可以使用 Request 对象。
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp FastAPI()app.post(/girl/{user_id})
async def read_girl(user_id, request: Request):q request.query_params.get(q)data: Dict await request.json()data.update({user_id: user_id, q: q})return dataif __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)嵌入 单个请求体 参数
假设你只有一个来自 Pydantic 模型 Item 的请求体参数 item。默认情况下FastAPI 将直接期望这样的请求体。但是如果你希望它期望一个拥有 item 键并在值中包含模型内容的 JSON就像在声明额外的请求体参数时所做的那样则可以使用一个特殊的 Body 参数 embeditem: Item Body(embedTrue) from typing import Annotated from fastapi import Body, FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None price: float tax: float | None None app.put(/items/{item_id}) async def update_item(item_id: int, item: Annotated[Item, Body(embedTrue)]): results {item_id: item_id, item: item} return results 在这种情况下FastAPI 将期望像这样的请求体 { item: { name: Foo, description: The pretender, price: 42.0, tax: 3.2 } } 而不是 { name: Foo, description: The pretender, price: 42.0, tax: 3.2 } 可以添加多个请求体参数到路径操作函数中即使一个请求只能有一个请求体。
还可以声明将作为请求体的一部分所接收的单一值。
还可以指示 FastAPI 在仅声明了一个请求体参数的情况下将原本的请求体嵌入到一个键中。 多个请求体参数
# -*- coding:utf-8 -*-
# Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp FastAPI()class Girl(BaseModel):name: strage: Optional[str] Noneclass Boy(BaseModel):name: strage: intapp.post(/boy_and_girl)
async def read_boy_and_girl(girl: Girl, boy: Boy):return {girl: dict(girl), boy: dict(boy)}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)此时在传递的时候应该按照如下方式传递 应该将两个 json 嵌套在一起组成一个更大的 json至于 key 就是我们的函数参数名。因此这种方式其实就等价于
# -*- coding:utf-8 -*-
# Author: komeiji satori
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp FastAPI()class BoyAndGirl(BaseModel):girl: Dictboy: Dictapp.post(/boy_and_girl)
async def read_boy_and_girl(boy_and_girl: BoyAndGirl):return dict(boy_and_girl)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)这种方式也是可以实现的只不过就字典内部的字典的不可进行限制了。当然啦我们仍然可以使用 Request 对象得到字典之后自己再进行判断因为对于 json 而言内部的字段可能是会变的而且最关键的是字段可能非常多。这个时候我个人更倾向于使用 Request 对象。 混合使用 Path、Query 和请求体参数
可以随意地混合使用 Path、Query 和请求体参数声明FastAPI 会知道该如何处理。 from typing import Annotated from fastapi import FastAPI, Path from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None price: float tax: float | None None app.put(/items/{item_id}) async def update_item( item_id: Annotated[int, Path(titleThe ID of the item to get, ge0, le1000)], q: str | None None, item: Item | None None, ): results {item_id: item_id} if q: results.update({q: q}) if item: results.update({item: item}) return results 请注意在这种情况下将从请求体获取的 item 是可选的。因为它的默认值为 None。 多个 请求体 参数 from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None price: float tax: float | None None class User(BaseModel): username: str full_name: str | None None app.put(/items/{item_id}) async def update_item(item_id: int, item: Item, user: User): results {item_id: item_id, item: item, user: user} return results 请求体 { item: { name: Foo, description: The pretender, price: 42.0, tax: 3.2 }, user: { username: dave, full_name: Dave Grohl } } 请求体中的单一值 from typing import Annotated from fastapi import Body, FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None price: float tax: float | None None class User(BaseModel): username: str full_name: str | None None app.put(/items/{item_id}) async def update_item( item_id: int, item: Item, user: User, importance: Annotated[int, Body()] ): results {item_id: item_id, item: item, user: user, importance: importance} return results 请求体 { item: { name: Foo, description: The pretender, price: 42.0, tax: 3.2 }, user: { username: dave, full_name: Dave Grohl }, importance: 5 } 多个 请求体参数 和 查询参数
默认情况下单一值被解释为查询参数因此你不必显式地添加 Query from typing import Annotated from fastapi import Body, FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None price: float tax: float | None None class User(BaseModel): username: str full_name: str | None None app.put(/items/{item_id}) async def update_item( *, item_id: int, item: Item, user: User, importance: Annotated[int, Body(gt0)], q: str | None None, ): results {item_id: item_id, item: item, user: user, importance: importance} if q: results.update({q: q}) return results Form 表单
调用 requests.post如果参数通过 data 传递的话则相当于提交了一个 form 表单那么在 FastAPI 中可以通过 await request.form() 进行获取注意内部同样是先调用 await request.body()。
from fastapi import FastAPI, Request, Response
import uvicornapp FastAPI()app.post(/girl)
async def girl(request: Request):# 此时 await request.json() 报错因为是通过 data 参数传递的相当于 form 表单提交# 如果是通过 json 参数传递那么 await request.form() 会得到一个空表单form await request.form()return [form.get(name), form.getlist(age)]if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 也可以通过其它方式
# -*- coding:utf-8 -*-
# Author: komeiji satori
from fastapi import FastAPI, Form
import uvicornapp FastAPI()app.post(/user)
async def get_user(username: str Form(...),password: str Form(...)):return {username: username, password: password}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 像 Form 表单查询参数、路径参数等等都可以和 Request 对象一起使用像上面的例子如果我们多定义一个 request: Request那么我们仍然可以通过 await request.form() 拿到相关的表单信息。所以如果你觉得某个参数不适合类型注解那么你可以单独通过 Request 对象进行解析。 文件上传
FastAPI 如何接收用户的文件上传呢首先如果想使用文件上传功能那么你必须要安装一个包 python-multipart直接 pip install python-multipart 即可。
from fastapi import FastAPI, File, UploadFile
import uvicornapp FastAPI()app.post(/file1)
async def file1(file: bytes File(...)):return f文件长度: {len(file)}app.post(/file2)
async def file1(file: UploadFile File(...)):return f文件名: {file.filename}, 文件大小: {len(await file.read())}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 我们看到一个直接获取字节流另一个是获取类似于文件句柄的对象。如果是多个文件上传要怎么做呢
from typing import List
from fastapi import FastAPI, UploadFile, File
import uvicornapp FastAPI()app.post(/file)
async def file(files: List[UploadFile] File(...)):指定类型为列表即可for idx, f in enumerate(files):files[idx] f文件名: {f.filename}, 文件大小: {len(await f.read())}return filesif __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 此时就实现了 FastAPI 文件上传当然文件上传并不影响我们处理表单可以自己试一下同时处理文件和表单。 返回静态资源
需要安装 aiofiles直接 pip 安装即可。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicornapp FastAPI()# name 参数只是起一个名字FastAPI 内部使用
app.mount(/static, StaticFiles(directoryrC:\Users\satori\Desktop\bg), namestatic)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)浏览器输入localhost:5555/static/1.png那么会返回 C:\Users\satori\Desktop\bg 下的 1.png 文件。 子应用
如果你有2个独立的FastAPI的应用你可以设置一个为主应用另外一个为子应用
from fastapi import FastAPIapp FastAPI()app.get(/app)
def read_main():return {message: Hello World from main app}subapi FastAPI()subapi.get(/sub)
def read_sub():return {message: Hello World from sub API}app.mount(/subapi, subapi)代理
可以使用root_path来设置代理。
使用命令行uvicorn main:app --root-path /api/v1
或者在代码中设置
from fastapi import FastAPI, Requestapp FastAPI(root_path/api/v1)app.get(/app)
def read_main(request: Request):return {message: Hello World, root_path: request.scope.get(root_path)}使用模板
你可以在FastAPI中使用任何模板常用的选择是Jinja2。安装pip install jinja2
使用
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFilesfrom fastapi.templating import Jinja2Templatesapp FastAPI()app.mount(/static, StaticFiles(directorystatic), namestatic)templates Jinja2Templates(directorytemplates)app.get(/items/{id}, response_classHTMLResponse)
async def read_item(request: Request, id: str):return templates.TemplateResponse(item.html, {request: request, id: id})模板文件templates/item.html html head titleItem Details/title link href{{ url_for(static, path/styles.css) }} relstylesheet /head body h1Item ID: {{ id }}/h1 /body /html 错误处理
from fastapi import FastAPI, HTTPException
import uvicornapp FastAPI()app.get(/items/{item_id})
async def read_item(item_id: str):if item_id ! foo:# 里面还可以传入 headers 设置响应头raise HTTPException(status_code404, detailitem 没有发现)return {item: bar}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 HTTPException 是一个普通的 Python 异常类继承了 Exception它携带了 API 的相关信息既然是异常那么我们不能 return、而是要 raise。这种方式返回错误因为它能够携带的信息太少了。 自定义异常
FastAPI 内部提供了一个 HTTPException但是我们也可以自定义但是注意我们自定义完异常之后还要定义一个 handler将异常和 handler 绑定在一起然后引发该异常的时候就会触发相应的 handler。
from fastapi import FastAPI, Request
from fastapi.responses import ORJSONResponse
import uvicornapp FastAPI()class ASCIIException(Exception):pass# 通过装饰器的方式将 ASCIIException 和 ascii_exception_handler 绑定在一起
app.exception_handler(ASCIIException)
async def ascii_exception_handler(request: Request, exc: ASCIIException):当引发 ASCIIException 的时候会触发 ascii_exception_handler 的执行同时会将 request 和 exception 传过去return ORJSONResponse(status_code404, content{code: 404, message: 你必须传递 ascii 字符串})app.get(/items/{item_id})
async def read_item(item_id: str):if not item_id.isascii():raise ASCIIExceptionreturn {item: fget {item_id}}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 关于 Request、Response我们除了可以通过 fastapi 进行导入还可以通过 starlette 进行导入因为 fastapi 的路由映射是通过 starlette 来实现的。 自定义 404
当访问一个不存在的 URL我们应该提示用户比如您要找到页面去火星了。
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.exceptions import StarletteHTTPException
import uvicornapp FastAPI()app.exception_handler(StarletteHTTPException)
async def not_found(request, exc):return ORJSONResponse({code: 404, message: 您要找的页面去火星了。。。})if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)此时当我们访问一个不存在的 URL 时就会返回我们自定义的 JSON 字符串。 Background Tasks ( 后台任务 )
background tasks 就是在返回响应之后立即运行的任务。
如果一个请求耗时特别久那么我们可以将其放在后台执行而 FastAPI 已经帮我们做好了这一步。我们来看一下
import time
from fastapi import FastAPI, BackgroundTasks
from starlette.background import BackgroundTask
from fastapi import Response, Request
import uvicorn
import orjsonapp FastAPI()def send_email(email: str, message: str ):发送邮件假设耗时三秒time.sleep(3)print(f三秒之后邮件发送给 {email!r}, 邮件信息: {message!r})app.get(/user/{email})
async def order(email: str, bg_tasks: BackgroundTasks):这里需要多定义一个参数此时任务就被添加到后台当 Response 对象返回之后触发bg_tasks.add_task(send_email, email, message这是一封邮件)# 我们在之前介绍 Response 的时候说过里面有一个参数 background# 所以我们也可以将任务放在那里面# 因此我们还可以# return Response(# orjson.dumps({message: 邮件发送成功}), # backgroundBackgroundTask(send_email, email, message这是一封邮件)# )return {message: 邮件发送成功}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 APIRouter
APIRouter 类似于 Flask 中的蓝图可以更好的组织大型项目举个栗子 在我当前的工程目录中有一个 app 目录和一个 main.py其中 app 目录中有一个 app01.py然后我们看看它们是如何组织的。
app/app01.py
# app/app01.py
from fastapi import APIRouterrouter APIRouter(prefix/router)# 以后访问的时候要通过 /router/v1 来访问
router.get(/v1)
async def v1():return {message: hello world}
main.py
# main.py
from fastapi import FastAPI
from app.app01 import router
import uvicornapp FastAPI()# 将 router 注册到 app 中相当于 Flask 中的 register_blueprint
app.include_router(router)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)
然后可以在外界通过 /router/v1 的方式来访问。
示例
from fastapi import APIRouterrouter APIRouter()router.get(/users/, tags[users])
async def read_users():return [{username: Rick}, {username: Morty}]router.get(/users/me, tags[users])
async def read_user_me():return {username: fakecurrentuser}router.get(/users/{username}, tags[users])
async def read_user(username: str):return {username: username}为所有路径进行同样的操作
from fastapi import APIRouter, Depends, HTTPExceptionfrom ..dependencies import get_token_headerrouter APIRouter(prefix/items,tags[items],dependencies[Depends(get_token_header)],responses{404: {description: Not found}},
)fake_items_db {plumbus: {name: Plumbus}, gun: {name: Portal Gun}}router.get(/)
async def read_items():return fake_items_dbrouter.get(/{item_id})
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code404, detailItem not found)return {name: fake_items_db[item_id][name], item_id: item_id}router.put(/{item_id},tags[custom],responses{403: {description: Operation forbidden}},
)
async def update_item(item_id: str):if item_id ! plumbus:raise HTTPException(status_code403, detailYou can only update the item: plumbus)return {item_id: item_id, name: The great Plumbus}该示例就为所有的路径添加了前缀标签、依赖和返回而不用在每个路径上单独声明简化了代码。 中间件 中间件在 web 开发中可以说是非常常见了说白了中间件就是一个函数或者一个类。在请求进入视图函数之前会先经过中间件被称为请求中间件而在中间件里面我们可以对请求进行一些预处理或者实现一个拦截器等等同理当视图函数返回响应之后也会经过中间件被称为响应中间件在中间件里面我们也可以对响应进行一些润色。 自定义中间件
在 FastAPI 里面也支持像 Flask 一样自定义中间件但是 Flask 里面有请求中间件和响应中间件但是在 FastAPI 里面这两者合二为一了我们看一下用法。
from fastapi import FastAPI, Request, Response
import uvicorn
import orjsonapp FastAPI()app.get(/)
async def view_func(request: Request):return {name: 古明地觉}app.middleware(http)
async def middleware(request: Request, call_next):定义一个协程函数然后使用 app.middleware(http) 装饰即可得到中间件# 请求到来时会先经过这里的中间件if request.headers.get(ping, ) ! pong:response Response(contentorjson.dumps({error: 请求头中缺少指定字段}),media_typeapplication/json,status_code404)# 当请求头中缺少 ping: pong在中间件这一步就直接返回了就不会再往下走了# 所以此时就相当于实现了一个拦截器return response# 然后如果条件满足则执行 await call_next(request)关键是这里的 call_next# 如果该中间件后面还有中间件那么 call_next 就是下一个中间件如果没有那么 call_next 就是对应的视图函数# 这里显然是视图函数因此执行之后会拿到视图函数返回的 Response 对象# 所以我们看到在 FastAPI 中请求中间件和响应中间件合在一起了response: Response await call_next(request)# 这里我们在设置一个响应头response.headers[status] successreturn responseif __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 内置的中间件
通过自定义中间件我们可以在不修改视图函数的情况下实现功能的扩展。但是除了自定义中间件之外FastAPI 还提供了很多内置的中间件。
from fastapi import FastAPIapp FastAPI()# 要求请求协议必须是 https 或者 wss如果不是则自动跳转
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)# 请求中必须包含 Host 字段为防止 HTTP 主机报头攻击并且添加中间件的时候还可以指定一个 allowed_hosts那么它是干什么的呢
# 假设我们有服务 a.example.com, b.example.com, c.example.com
# 但我们不希望用户访问 c.example.com就可以像下面这么设置如果指定为 [*]或者不指定 allow_hosts则表示无限制
from starlette.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(TrustedHostMiddleware, allowed_hosts[a.example.com, b.example.com])# 如果用户的请求头的 Accept-Encoding 字段包含 gzip那么 FastAPI 会使用 GZip 算法压缩
# minimum_size1000 表示当大小不超过 1000 字节的时候就不压缩了
from starlette.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size1000)除了这些还有其它的一些内置的中间件可以自己查看一下不过不是很常用。 CORS ( 跨域设置 )
CORS 过于重要我们需要单独拿出来说。
CORS跨域资源共享是指浏览器中运行的前端里面拥有和后端通信的 JavaScript 代码而前端和后端处于不同源的情况。源协议http、https、域baidu.com、app.com、localhost以及端口80、443、8000只要有一个不同那么就是不同源。比如下面都是不同的源
http://localhosthttps://localhosthttp://localhost:8080
即使它们都是 localhost但是它们使用了不同的协议或端口所以它们是不同的源。假设你的前端运行在 localhost:8080并且尝试与 localhost:5555 进行通信然后浏览器会向后端发送一个 HTTP OPTIONS 请求后端会发送适当的 headers 来对这个源进行授权所以后端必须有一个 允许的源 列表如果前端对应的源是被允许的浏览器才会允许前端向后端发请求否则就会出现跨域失败。
而默认情况下前后端必须是在同一个源如果不同源那么前端就会请求失败。而前后端分离早已成为了主流因此跨域问题是必须要解决的。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicornapp FastAPI()
app.add_middleware(CORSMiddleware,# 允许跨域的源列表例如 [http://www.example.org] 等等[*] 表示允许任何源allow_origins[*],# 跨域请求是否支持 cookie默认是 False如果为 Trueallow_origins 必须为具体的源不可以是 [*]allow_credentialsFalse,# 允许跨域请求的 HTTP 方法列表默认是 [GET]allow_methods[*],# 允许跨域请求的 HTTP 请求头列表默认是 []可以使用 [*] 表示允许所有的请求头# 当然 Accept、Accept-Language、Content-Language 以及 Content-Type 总之被允许的allow_headers[*],# 可以被浏览器访问的响应头, 默认是 []一般很少指定# expose_headers[*]# 设定浏览器缓存 CORS 响应的最长时间单位是秒。默认为 600一般也很少指定# max_age1000
)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)以上即可解决跨域问题。
使用CORSMiddleware来配置跨域
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp FastAPI()origins [http://localhost.tiangolo.com,https://localhost.tiangolo.com,http://localhost,http://localhost:8080,]app.add_middleware(CORSMiddleware,allow_originsorigins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)app.get(/)
async def main():return {message: Hello World}它支持以下参数
allow_origins - 一个允许跨域请求的源列表。例如 [https://example.org, https://www.example.org]。你可以使用 [*] 允许任何源。allow_origin_regex - 一个正则表达式字符串匹配的源允许跨域请求。例如 https://.*\.example\.org。allow_methods - 一个允许跨域请求的 HTTP 方法列表。默认为 [GET]。你可以使用 [*] 来允许所有标准方法。allow_headers - 一个允许跨域请求的 HTTP 请求头列表。默认为 []。你可以使用 [*] 允许所有的请求头。Accept、Accept-Language、Content-Language 以及 Content-Type 请求头总是允许 CORS 请求。allow_credentials - 指示跨域请求支持 cookies。默认是 False。另外允许凭证时 allow_origins 不能设定为 [*]必须指定源。expose_headers - 指示可以被浏览器访问的响应头。默认为 []。max_age - 设定浏览器缓存 CORS 响应的最长时间单位是秒。默认为 600。 高阶操作
看一些 FastAPI 的高阶操作这些操作有的不一定能用上但用上了确实会方便许多。 其它的响应
返回 json 数据可以是JSONResponse、UJSONResponse、ORJSONResponseContent-Type 是 application/json返回 html 是 HTMLResponseContent-Type 是 text/html返回 PlainTextResponseContent-Type 是 text/plain。但是我们还可以有三种响应分别是返回重定向、字节流、文件。 重定向
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import uvicornapp FastAPI()app.get(/index)
async def index():return RedirectResponse(https://www.bilibili.com)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)页面中访问 /index 会跳转到 bilibili。 字节流
返回字节流需要使用异步生成器的方式
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicornapp FastAPI()async def some_video():for i in range(5):yield fvideo {i} bytes .encode(utf-8)app.get(/index)
async def index():return StreamingResponse(some_video())if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 如果有文件对象那么也是可以直接返回的。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicornapp FastAPI()app.get(/index)
async def index():return StreamingResponse(open(main.py, encodingutf-8))if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 文件
返回文件的话还可以通过 FileResponse
from fastapi import FastAPI
from fastapi.responses import FileResponse
import uvicornapp FastAPI()app.get(/index)
async def index():# filename 如果给出它将包含在响应的 Content-Disposition 中。return FileResponse(main.py, filename这不是main.py)if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)HTTP 验证
如果当用户访问某个请求的时候我们希望其输入用户名和密码来确认身份的话该怎么做呢
from fastapi import FastAPI, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import uvicornapp FastAPI()security HTTPBasic()app.get(/index)
async def index(credentials: HTTPBasicCredentials Depends(security)):return {username: credentials.username, password: credentials.password}if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)测试 输入完毕之后信息会保存在 credentials我们可以获取出来进行验证。 websocket
FastAPI 如何实现 websocket
from fastapi import FastAPI
from fastapi.websockets import WebSocket
import uvicornapp FastAPI()app.websocket(/ws)
async def ws(websocket: WebSocket):await websocket.accept()while True:# websocket.receive_bytes()# websocket.receive_json()data await websocket.receive_text()await websocket.send_text(f收到来自客户端的回复: {data})if __name__ __main__:uvicorn.run(main:app, host0.0.0.0, port5555)然后我们通过浏览器进行通信
!DOCTYPE html
html langen
headmeta charsetUTF-8titleTitle/title
/head
bodyscriptws new WebSocket(ws://localhost:5555/ws);//如果连接成功, 会打印下面这句话, 否则不会打印ws.onopen function () {console.log(连接成功)};//接收数据, 服务端有数据过来, 会执行ws.onmessage function (event) {console.log(event)};//服务端主动断开连接, 会执行.//客户端主动断开的话, 不执行ws.onclose function () { }/script
/body
/html测试 示例
from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponseapp FastAPI()html
!DOCTYPE html
htmlheadtitleChat/title/headbodyh1WebSocket Chat/h1form action onsubmitsendMessage(event)input typetext idmessageText autocompleteoff/buttonSend/button/formul idmessages/ulscriptvar ws new WebSocket(ws://localhost:8000/ws);ws.onmessage function(event) {var messages document.getElementById(messages)var message document.createElement(li)var content document.createTextNode(event.data)message.appendChild(content)messages.appendChild(message)};function sendMessage(event) {var input document.getElementById(messageText)ws.send(input.value)input.value event.preventDefault()}/script/body
/html
app.get(/)
async def get():return HTMLResponse(html)app.websocket(/ws)
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data await websocket.receive_text()await websocket.send_text(fMessage text was: {data})FastAPI 服务的部署 使用异步框架最重要的是要搭配一个异步驱动去访问数据库因为 web 服务的瓶颈都是在数据库上面。 上面介绍了 FastAPI 的绝大部分内容然后我们来看看 FastAPI 服务的部署其实部署很简单直接 uvicorn.run 即可。但是这里面有很多的参数我们主要是想要介绍这些参数。
def run(app, **kwargs):config Config(app, **kwargs)server Server(configconfig)......看到 app 和 **kwargs 都传递给了 Config所以我们只需要看 Config 里面都有哪些参数即可。这里选出一部分
app第一个参数不需要解释host监听的ipport监听的端口uds绑定的 unix domain socket一般不用fd从指定的文件描述符中绑定 socketloop事件循环实现可选项为 auto|asyncio|uvloop|iocphttpHTTP 协议实现可选项为 auto|h11|httptoolswswebsocket 协议实现可选项为 auto|none|websockets|wsprotolifespanlifespan 实现可选项为 auto|on|offenv_file环境变量配置文件log_config日志配置文件log_level日志等级access_log是否记录日志use_colors是否带颜色输出日志信息interface应用接口可选 auto|asgi3|asgi2|wsgidebug是否开启 debug 模式reload是否自动重启reload_dirs要自动重启的目录reload_delay多少秒后自动重启workers工作进程数limit_concurrency并发的最大数量limit_max_requests能 hold 住的最大请求数 3、示例fastapi 开发接口 ( 只是api接口不带web渲染 ) 示例豆瓣电影top250
import requests
from scrapy.http import HtmlResponse
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp FastAPI()headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36 Edg/111.0.1661.62,
}def douban_movie_top250(page_num: int):index (page_num - 1) * 25# https://movie.douban.com/top250?start50filterurl fhttps://movie.douban.com/top250?start{index}filter__resp requests.get(url, headersheaders)if 200 __resp.status_code:resp HtmlResponse(url, body__resp.content, encodingutf-8)movie_name_list resp.xpath(//div[classitem]//div[classhd]/a/span[1]/text()).extract()movie_url_list resp.xpath(//div[classitem]//div[classhd]/a/href).extract()movie_info_list list(zip(movie_name_list, movie_url_list))return movie_info_listelse:return {请求失败: f status_code --- {__resp.status_code}}app.get(/douban/movie_top250)
async def get_item(page_num):和 Flask 不同Flask 是使用 而 FastAPI 使用 {}try:page_num_int int(page_num)except BaseException as be:return {错误信息: 页码必须是数字}data douban_movie_top250(page_num_int)return {data: data}if __name__ __main__:http://127.0.0.1:5555/douban/movie_top250?page_num1print(f{Path(__file__).stem}:app)uvicorn.run(f{Path(__file__).stem}:app, host0.0.0.0, port5555)pass访问http://127.0.0.1:5555/douban/movie_top250?page_num5 示例B 站视频 字幕 获取
import uvicorn
from pathlib import Path
from fastapi import FastAPI
import requests
import json
from scrapy.http import HtmlResponseapp FastAPI()headers {accept-language: zh-CN,zh;q0.9,en;q0.8,en-GB;q0.7,en-US;q0.6,cookie: buvid331D606A5-C08F-CF7F-6345-DFE18CDF1FCA25535infoc; b_nut1681180325; CURRENT_FNVAL4048; _uuidFA2D2910A-1F4C-9ED10-24FB-710C228FE726630982infoc; buvid_fpdd9cac90362a92030f254a522e274486; buvid4F13E840C-6245-40B7-9B63-5FDD1992821229542-023041110-0bxmZkxc4Ip6QXGeEfs0Og%3D%3D; CURRENT_PID130742c0-d811-11ed-ac38-37fb01852f74; rpdid|(JYYkY~RRYu0J\uY)uk~lJY|; i-wanna-go-back-1; header_theme_versionCLOSE; home_feed_column5; is-2022-channel1; nostalgia_conf-1; DedeUserID384760568; DedeUserID__ckMd58fd50449771672ee; b_ut5; FEED_LIVE_VERSIONV_NO_BANNER_1; bsourcesearch_bing; browser_resolution1863-969; bp_video_offset_384760568787947892852654100; b_lsidC105138DE_187B25E90A6; SESSDATAd3f7b6a0%2C1697876752%2C413d0%2A42; bili_jcte41d9dfdbd372b0cb95222cfa0d33199; sid59e50ddx; innersign1; PVID1; innersign1
}def get_subtitle(video_idNone):if video_id:url fhttps://www.bilibili.com/video/{video_id}/resp requests.get(url, headersheaders)if 200 resp.status_code:scrapy_resp HtmlResponse(url, bodyresp.content, encodingutf-8)try:temp scrapy_resp.css(html).re(subtitle_url:(.*?))[0]except BaseException as be:return {请求失败: str(be)}subtitle_url temp.replace(r\u002F, /)print(subtitle_url)r requests.get(subtitle_url)if 200 r.status_code:return r.json()else:return {请求失败, 失败状态码: resp.status_code}else:return {请求失败, 失败状态码: resp.status_code}else:return {请求失败: 视频 id 错误}app.get(/bilibili/{video_id})
async def get_item(video_id):和 Flask 不同Flask 是使用 而 FastAPI 使用 {}data get_subtitle(video_id)return {data: data}if __name__ __main__:http://127.0.0.1:5555/bilibili/BV1bW411n7fYprint(f{Path(__file__).stem}:app)uvicorn.run(f{Path(__file__).stem}:app, host0.0.0.0, port5555)pass4、示例fastapi 开发 web 渲染 网站 FastAPI 这个 Python Web 框架并没有带 渲染网页的模板引擎但是也正因为如此它可以使用任何网页模板。官方例子是 jinjia2 。
模板是全栈 Web 开发的重要组成部分。使用 Jinja您可以构建丰富的模板为 Python Web 应用程序的前端提供支持。
Jinja 是一个用 Python 编写的模板引擎旨在帮助 API 响应的渲染过程。在每种模板语言中都有变量被替换为实际传递给它们的值当模板被渲染时有控制模板逻辑的标签。 安装 jinja2
安装pip install jinja2 aiofiles
Jinja 模板只是一个文本文件。 Jinja 可以生成任何基于文本的格式HTML、XML、CSV、LaTeX 等。 Jinja 模板不需要有特定的扩展名.html、.xml 或任何其他扩展名都可以。
关于模版的扩展名任何文件都可以作为模板加载无论文件扩展名如何。添加 .jinja 扩展名如 user.html.jinja 可能会使某些 IDE 或编辑器插件更容易但这不是必需的。自动转义可以基于文件扩展名应用因此在这种情况下您需要考虑额外的后缀。 识别模板的另一个很好的启发式方法是它们位于模板 templates 文件夹中而不管扩展名是什么。这是项目的常见布局。 Jinja 模板引擎使用花括号 {} 来区分其表达式和语法以及与常规 HTML、文本和模板文件中的任何其他变量。{{}} 语法称为变量块。{% %} 语法包含控制结构如 if/else 、循环和宏。Jinja 模板语言中使用的三种常见语法块包括以下内容
{% ... %}这种语法用于控制结构等语句。{{ todo.item }}这个语法用于打印出传递给它的表达式的值。{# Test #}: 这种语法在写评论时使用不在网页上显示。
Jinja2 是一种流行的模板语言被 Flask、Bottle、Pelican 使用也可被 Django 使用。 渲染第一个 Jinja 模板
代码
import jinja2
environment jinja2.Environment()
template environment.from_string(Hello, {{ name }}!)
result template.render(name渲染第一个jinja2模板)
print(result)
Jinja 的核心组件是 Environment() 类。在此示例中创建了一个不带任何参数的 Jinja 环境。然后通过environment.from_string 来自定义环境。这里是创建一个普通环境并在其中加载字符串 Hello, {{ name }}! 作为模板。
这个例子显示了在使用 Jinja 时通常会执行的两个重要步骤
加载模板加载包含占位符变量的源。默认情况下它们包含在一对大括号 {{ }} 中。渲染模板用内容填充占位符。您可以提供字典或关键字参数作为上下文。
执行结果如下 使用外部文件作为模板
与上述方式同理我们可以使用外部文件作为我们的模版来源在我们的项目中创建一个新文件夹。在工作目录中创建一个名为 templates/ 的文件夹。
然后您可以在 template 目录中创建 index.html 模板文件并使用 Jinja2 语法来呈现它们。例如在template/index.html 中写入如下内容:
!DOCTYPE html html head titleWelcome/title link href{{ url_for(static, path/styles.css) }} relstylesheet /head body h1Hello, {{ name }}/h1 /body /html
回到 main.py 中
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templatesapp FastAPI()
app.mount(/static, StaticFiles(directorystatic), namestatic)
templates Jinja2Templates(directorytemplates)app.get(/{name})
async def home(request: Request, name: str):return templates.TemplateResponse(index.html, {request: request,name: name})if __name__ __main__:pass
整个文件的目录结构如下 启动 FastAPI 服务uvicorn main:app --reload --port 8888
然后另外打开一个终端执行 curl 127.0.0.1:8888/Yuzhou1su 命令可以看到如下 name 被渲染出来的结果 通过浏览器访问这个 http://127.0.0.1:8888/Yuzhou1su 就能看到 css 渲染的颜色 Jinja 模板变量可以是任何 Python 类型或对象只要它们可以转换为字符串。可以将模型、列表或字典类型传递到模板中并通过将这些属性放置在先前列出的第二个块中来显示其属性。在下一节中我们将看一下过滤器。过滤器是每个模板引擎的重要组成部分在 Jinja 中过滤器使我们能够执行某些函数例如从列表中连接值和检索对象的长度等等。Jinja 中常用的功能变量、过滤器、if 语句、循环、宏和模板继承。 变量
模板变量由传递给模板的上下文字典定义。
在模板中只要应用程序传递了变量您就可以随意操作这些变量。变量可能还具有您可以访问的属性或元素。变量具有哪些属性取决于提供该变量的应用程序。
除了标准的 Python __getitem__ “下标”语法 [] 之外您还可以使用点. 来访问变量的属性。
以下行执行相同的操作
{{ foo.bar }}
{{ foo[bar] }} Filters
尽管 Python 和 Jinja 的语法非常相似但是像连接字符串、将字符串的第一个字符设置为大写等修改操作不能使用Python 的语法在 Jinja 中完成。因此为了执行这样的修改操作我们在 Jinja 中使用过滤器。
变量可以被过滤器修改。过滤器与变量用管道符号|分隔并且可以在括号中包含可选参数。可以链接多个过滤器。一个过滤器的输出应用于下一个。过滤器的定义格式如下
{{ variable | filter_name(*args) }}
不加参数的过滤器
{{ variable | filter_name }}
{{ name|striptags|title }}
default 过滤器: 如果该值未定义它将返回传递的默认值否则返回变量的值
{{ my_variable | default(my_variable is not defined) }}
escape 过滤器: 这个过滤器用于渲染原始 HTML 输出将字符串 s 中的字符 ” 转换为 HTML 安全序列。如果您需要在 HTML 中显示可能包含此类字符的文本请使用此选项。将返回值标记为标记字符串。
{{ titleTodo Application/title | escape }}
titleTodo Application/title
类型转换过滤器: 这些过滤器包括 int 和 float 过滤器用于从一种数据类型转换到另一种数据类型
{{ 3.142 | int }}
3
{{ 20 | float }}
20.0
join 过滤器join(*value*, *du* , *attributeNone*)返回一个字符串它是序列中字符串的串联。元素之间的分隔符默认为空字符串您可以使用可选参数定义它
{{ [1, 2, 3] | join(|) }}- 1|2|3
{{ [1, 2, 3] | join }}- 123
也可以连接对象的某些属性
{{ users|join(, , attributeusername) }}
长度 filter 这个过滤器返回一个序列或集合的长度它的作用与 Python 中 len() 函数的作用相同
Todo count: {{ todos | length }}
Todo count: 4 if 条件
Jinja 中 if 语句的用法与 Python 中的用法类似。在 {% %} 控制块中使用。让我们看一个例子
{% if todos %}
ul
{% for todo in todos %}li{{ todo.name|e }}/li
{% endfor %}
/ul
{% endif %} Loop 条件
我们也可以在Jinja中对变量进行迭代。这可以是一个列表或一个一般的函数、 比如说下面这个例如
{% for todo in todos %}li{{ todo.name|e }}/li
{% endfor %}
你可以在 for 循环中访问特殊的变量比如 loop.index 它给出了当前迭代的索引。 宏
宏可与常规编程语言中的函数相媲美。它们有助于将常用的习语放入可重用的函数中以免重复自己“DRY” 原则。
{% macro input(name, value, typetext, size20) %}div classforminput type{{ type }} name{{ name }}value{{ value|escape }} size{{ size }}/div
{% endmacro %}
现在为了在你的表单中快速创建一个输入调用了这个宏
{{ input(item) }}
渲染完成后将会返回 div classforminput typetext nameitem value size20 //div FastAPI 中的 Jinja
FastAPI 实际上是为构建 API 和微服务而设计的。它可用于构建使用 Jinja 提供 HTML 服务的 Web 应用程序但这并不是它真正优化的目的。
如果您想构建一个在服务器上呈现大量 HTML 的大型网站Django 可能是更好的选择。
但是如果您正在使用 React、Angular 或 Vue 等前端框架构建现代网站那么从 FastAPI 获取数据是一个不错的选择。 5、FastAPI、vue 开发 web 网站 5.1 关于 FastAPI 与 Vue3 的通信
https://zhuanlan.zhihu.com/p/632387477
基于Vue3和FastAPI对数据库进行操作https://zhuanlan.zhihu.com/p/632393099 5.2 连接 Vue.js 作为前端Fastapi 作为后端
https://www.cnblogs.com/hahaha111122222/p/15904405.html 目录结构 ├── main.py └── templates └── home.html 后端 fastapi pip install fastapi[all] pip install jinja2 main.py
我们在 / 中服务于我们的前端并在该路径中呈现我们的home.html。我们使用templates文件夹保存我们的HTML并将其传递给Jinja。另外我们将从我们的front-end向/add发送一个请求。
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from pydantic import BaseModeltemplates Jinja2Templates(directorytemplates) app FastAPI()class TextArea(BaseModel):content: strapp.post(/add)
async def post_textarea(data: TextArea):print(data.dict())return {**data.dict()}app.get(/)
async def serve_home(request: Request):return templates.TemplateResponse(home.html, {request: request})前端 - home.html
让我们创建一个有文本区域和按钮的虚拟应用程序。我们正在使用Axios将请求发送到后端。因为它们在同一个端口上运行所以我们可以直接将/add传递给Axios。
html
title/title
script srchttps://cdn.jsdelivr.net/npm/vue/dist/vue.js/script
script srchttps://unpkg.com/axios/dist/axios.min.js/scriptbodydiv idapptextarea name idcontent cols30 rows10 v-modelcontent/textareabutton clickaddText idadd-textareaclick me/button/divscriptnew Vue({el: #app,data: {title: ,content: },methods: {addText() {return axios.post(/add, {content: this.content}, {headers: {Content-type: application/json,}}).then((response) {console.log(content: this.content);});}}});/script
/body/html运行访问测试 命令uvicorn main:app --reload 最后你会有一个可怕的文本区和一个按钮。但它会帮助你更好地理解事情。 5.3 FastApiVueLayUI实现简单的前后端分离 demo
实际使用中通常建议前后端项目分离。下面使用FastApiVueLayUI做一个前后端分离的Demo。 后端
后端采用 FastApi代码 test.py
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import JSONResponse
from pathlib import Path
import uvicorn
import subprocessapp FastAPI()
templates Jinja2Templates(directorytemplates)app.get(/info)
async def user_list():# vue的响应数据ret_list [{id: 1, value: one},{id: 2, value: two},{id: 3, value: three},]return JSONResponse(contentret_list)app.get(/check)
async def home(request: Request):return templates.TemplateResponse(index.html, {request: request})if __name__ __main__:http://127.0.0.1:5555/check# print(f{Path(__file__).stem}:app)uvicorn.run(f{Path(__file__).stem}:app, host0.0.0.0, port5555)pass前端
前端直接导入Vue、LayUI、Axios 的 JS 和 CSS 的 CDN 资源在 Vue 实例的 mount 阶段使用axios 调用后端接口拿到数据使用 LayUI 的样式对 table 元素进行美化。
代码
!DOCTYPE html
html langen
headmeta charsetUTF-8meta http-equivX-UA-Compatible contentIEedgemeta nameviewport contentwidthdevice-width, initial-scale1.0script srchttps://unpkg.com/vue3/dist/vue.global.js/scriptscript srchttps://unpkg.com/axios/dist/axios.min.js/script!-- 引入 layui.css --link relstylesheet hrefhttps://www.layuicdn.com/layui/css/layui.css/!-- 引入 layui.js --script srchttps://www.layuicdn.com/layui/layui.js typetext/javascript charsetutf-8/scripttitleHome/title
/head
body
div idapptable classlayui-tabletr v-forfo in info_listtd [[ fo.id ]] /tdtd [[ fo.value ]] /td/tr/table
/div
table idtest classlayui-table/tablescriptconst {createApp, ref} Vueconst vue_app createApp({data() {return {info_list: [{id: 1, name: 默认值}],// info: hello vue...}},mounted() {this.showinfo();},methods: {showinfo() {axios.get(/info).then(response {this.info_list response.data;// console.log(response);console.log(this.info_list --- ${this.info_list.toString()});}, err {console.log(err);})}}});// vue_app.config.delimiters [[[, ]]];vue_app.config.compilerOptions.delimiters [[[, ]]]vue_app.mount(#app);
/script
/body
/html
vue 和 jinja2 默认都使用 {{内容}} 在前端进行显示变量的值所以会造成冲突。
可以修改 vue 显示值得方式即修改 插值符
vue 2 方式 script const vue new Vue({ el:#app, delimiters: [[[, ]]], data:{ selects:[enu,cha], userData:[] } /script vue 3 方式 在 Vue 3 中默认的差值符号是双大括号{{ }}用于渲染动态数据到模板中。然而如果你希望修改默认的差值符号Vue 3 提供了一种简单的方式来实现。 可以在创建 Vue 应用程序实例之前使用 createApp 函数的 config 方法来配置全局的差值符号 import { createApp } from vue; const app createApp({}); //app.config.delimiters [${, }]; app.config.compilerOptions.delimiters [${, }]; app.mount(#app); 上述代码中我们通过 app.config.delimiters 来修改差值符号为 ${ }。 修改之后你可以在模板中使用新的差值符号来显示动态数据 template div p${ message }/p /div /template script export default { data() { return { message: Hello, world! }; } }; /script 在上述示例中我们使用 ${ } 差值符号来显示 message 数据。 需要注意的是修改差值符号后你需要确保新的差值符号与模板中的变量名不会发生冲突。同时修改差值符号只在当前应用程序实例范围内有效不会影响其他应用程序实例。 运行项目
启动 FastApi 后端服务器访问 /test/check 接口。 使用 axios 发送 get 和 post 请求详解
https://blog.csdn.net/grand_brol/article/details/108167088 QA
Q为什么在请求/info 接口总会出现一个Temporary Redirect 重定向呢 A原因是因为我们在 FastApi 接口定义的时候uri 的格式不规范导致uri 的结尾不需要/如果你接口增加了/我们使用浏览器访问 uri浏览器会忽略结尾的/FastApi 会在内部进行查重定向将浏览器不带/的请求重定向到我们定义的带/的视图函数上。 5.4 使用 python fastapivue 快速搭建网站
https://www.elprup.com/2020/09/19/fastapi_vue/
传统网站由一个 web 框架完全承担例如基于 nodejs 的 expresskoa基于 python 的 djangotornado。新型网站演变为用 vue 开发前端系统使用 api 框架开发后端 api 请求接口的模式。 5.5 用 FastAPI 和 Vue.js 开发一个单页应用程序 原文地址https://testdriven.io/blog/developing-a-single-page-app-with-fastapi-and-vuejs/ 源码地址https://github.com/testdrivenio/fastapi-vue 翻译1https://juejin.cn/post/7113790977848360967 翻译2https://www.cnblogs.com/leimu/p/16992966.html 文章转载自： http://www.morning.cybch.cn.gov.cn.cybch.cn http://www.morning.mxhys.cn.gov.cn.mxhys.cn http://www.morning.hcgbm.cn.gov.cn.hcgbm.cn http://www.morning.lmpfk.cn.gov.cn.lmpfk.cn http://www.morning.drspc.cn.gov.cn.drspc.cn http://www.morning.zlkps.cn.gov.cn.zlkps.cn http://www.morning.fgxnb.cn.gov.cn.fgxnb.cn http://www.morning.yjprj.cn.gov.cn.yjprj.cn http://www.morning.kbgzj.cn.gov.cn.kbgzj.cn http://www.morning.iterlog.com.gov.cn.iterlog.com http://www.morning.pxwzk.cn.gov.cn.pxwzk.cn http://www.morning.fjntg.cn.gov.cn.fjntg.cn http://www.morning.spfq.cn.gov.cn.spfq.cn http://www.morning.tscsd.cn.gov.cn.tscsd.cn http://www.morning.njntp.cn.gov.cn.njntp.cn http://www.morning.mcndn.cn.gov.cn.mcndn.cn http://www.morning.zpzys.cn.gov.cn.zpzys.cn http://www.morning.wkxsy.cn.gov.cn.wkxsy.cn http://www.morning.tbstj.cn.gov.cn.tbstj.cn http://www.morning.clccg.cn.gov.cn.clccg.cn http://www.morning.wtrjq.cn.gov.cn.wtrjq.cn http://www.morning.lqlhw.cn.gov.cn.lqlhw.cn http://www.morning.wlbwp.cn.gov.cn.wlbwp.cn http://www.morning.bctr.cn.gov.cn.bctr.cn http://www.morning.cgmzt.cn.gov.cn.cgmzt.cn http://www.morning.rzmzm.cn.gov.cn.rzmzm.cn http://www.morning.blqmn.cn.gov.cn.blqmn.cn http://www.morning.kyhnl.cn.gov.cn.kyhnl.cn http://www.morning.lyzwdt.com.gov.cn.lyzwdt.com http://www.morning.qkcyk.cn.gov.cn.qkcyk.cn http://www.morning.btlmb.cn.gov.cn.btlmb.cn http://www.morning.nkmw.cn.gov.cn.nkmw.cn http://www.morning.wfzdh.cn.gov.cn.wfzdh.cn http://www.morning.tyklz.cn.gov.cn.tyklz.cn http://www.morning.rywr.cn.gov.cn.rywr.cn http://www.morning.tpnxj.cn.gov.cn.tpnxj.cn http://www.morning.rsjng.cn.gov.cn.rsjng.cn http://www.morning.tnzwm.cn.gov.cn.tnzwm.cn http://www.morning.lmrcq.cn.gov.cn.lmrcq.cn http://www.morning.prddj.cn.gov.cn.prddj.cn http://www.morning.bccls.cn.gov.cn.bccls.cn http://www.morning.jqmmf.cn.gov.cn.jqmmf.cn http://www.morning.gydth.cn.gov.cn.gydth.cn http://www.morning.tkcz.cn.gov.cn.tkcz.cn http://www.morning.xykst.cn.gov.cn.xykst.cn http://www.morning.ykwqz.cn.gov.cn.ykwqz.cn http://www.morning.jwsrp.cn.gov.cn.jwsrp.cn http://www.morning.jlxld.cn.gov.cn.jlxld.cn http://www.morning.leboju.com.gov.cn.leboju.com http://www.morning.mxgpp.cn.gov.cn.mxgpp.cn http://www.morning.zwzwn.cn.gov.cn.zwzwn.cn http://www.morning.txfxy.cn.gov.cn.txfxy.cn http://www.morning.sqmlw.cn.gov.cn.sqmlw.cn http://www.morning.rsqpc.cn.gov.cn.rsqpc.cn http://www.morning.drcnf.cn.gov.cn.drcnf.cn http://www.morning.madamli.com.gov.cn.madamli.com http://www.morning.wslr.cn.gov.cn.wslr.cn http://www.morning.wmgjq.cn.gov.cn.wmgjq.cn http://www.morning.fksdd.cn.gov.cn.fksdd.cn http://www.morning.wqrk.cn.gov.cn.wqrk.cn http://www.morning.lbjdx.cn.gov.cn.lbjdx.cn http://www.morning.zrlms.cn.gov.cn.zrlms.cn http://www.morning.gghhmi.cn.gov.cn.gghhmi.cn http://www.morning.fyskq.cn.gov.cn.fyskq.cn http://www.morning.ldcrh.cn.gov.cn.ldcrh.cn http://www.morning.jpbky.cn.gov.cn.jpbky.cn http://www.morning.rkwlg.cn.gov.cn.rkwlg.cn http://www.morning.brwei.com.gov.cn.brwei.com http://www.morning.thzwj.cn.gov.cn.thzwj.cn http://www.morning.lcbt.cn.gov.cn.lcbt.cn http://www.morning.nzmqn.cn.gov.cn.nzmqn.cn http://www.morning.zwppm.cn.gov.cn.zwppm.cn http://www.morning.qkbwd.cn.gov.cn.qkbwd.cn http://www.morning.tgfsr.cn.gov.cn.tgfsr.cn http://www.morning.prhqn.cn.gov.cn.prhqn.cn http://www.morning.jgzmr.cn.gov.cn.jgzmr.cn http://www.morning.thmlt.cn.gov.cn.thmlt.cn http://www.morning.ymjrg.cn.gov.cn.ymjrg.cn http://www.morning.njftk.cn.gov.cn.njftk.cn http://www.morning.wqbbc.cn.gov.cn.wqbbc.cn
