基于Python3.6+版本的、用于构建API现代的、高性能的web框架。FastAPI是建立在Pydantic和Starlette基础上的，Pydantic是一个基于Python类型提示来定义数据验证、序列化和文档的库。Starlette是一种轻量级的ASGI框架/工具包，是构建高性能Asyncio服务的理性选择。

Uvicorn 是一个闪电般快速的 ASGI 服务器，基于 uvloop 和 httptools 构建。
实现一个基于ASGI（异步服务器网关接口）的最小应用程序接口。
Starlette 负责 web 部分。
Pydantic 负责数据部分。
https://fastapi.tiangolo.com/zh/tutorial/first-steps/

2、运行方式

运行命令 uvicorn main:app --reload
pycharm运行 (通过 uvicorn 命令行 uvicorn 脚本名:app对象--reload 参数启动服务)


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

3、创建接口步骤

导入 FastAPI（from fastapi import FastAPI）；
创建一个 app 实例（app = FastAPI()）；
编写一个路径操作装饰器（如 @app.get("/")）；
编写一个路径操作函数（如上面的 def root(): ...）；
定义返回值运行开发服务器（如 uvicorn main:app --reload）；

4、自动生成API文档

4.1 交互式API文档

在浏览器中请求 http://127.0.0.1:8000/docs ，显示交互式API文档，自动交互式 API 文档（由 Swagger UI 提供），如下图：

4.2 备用API文档

在浏览器中请求 http://127.0.0.1:8000/redoc ，显示备用API文档，备用的自动交互式文档（由 ReDoc 提供），如下图：

5、FastApi 执行顺序

按照路径顺序匹配，默认匹配的是第一个带参数的路径

6、Python-Pydantic库

pydantic库是一种常用的用于数据接口schema定义与检查的库。
通过pydantic库，我们可以更为规范地定义和使用数据接口。
pydantic库的数据定义方式是通过BaseModel类来进行定义的，所有基于pydantic的数据类型本质上都是一个BaseModel类。调用时，我们只需要对其进行实例化即可。
是一个用来执行数据校验的python库。


from pydantic import BaseModel
 
# schema基本定义方法
class Person(BaseModel):
    name: str
 
# 基本的schema实例化方法-直接传入
p = Person(name="ABC123")
print(p.json())
 
>>> {"name": "ABC123"}

6.1 BaseModel模型

可以在代码运行时强制执行类型提示，并在数据校验无效时提供友好的错误提示。
是一个解析库，而不是验证库（简单来说：pydantic保证输出模型的类型和约束）
所有基于 pydantic 的数据类型本质上都是一个 BaseModel 类
使用pydantic模型作请求体
需要安装 pip install pydantic
使用时需要导入 from pydantic import BaseMode


from pydantic import BaseModel
 
class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    status: Optional[bool] = None

6.2 请求体 + 路径参数 + 查询参数


from typing import Union
 
from fastapi import FastAPI
from pydantic import BaseModel
 
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
 
app = FastAPI()
 
"""
函数参数将依次按如下规则进行识别：
● 如果在路径中也声明了该参数，它将被用作路径参数（例如：item_id）。
● 如果参数属于单一类型（比如 int、float、str、bool 等）它将被解释为查询参数（例如：q）。
● 如果参数的类型被声明为一个 Pydantic 模型，它将被解释为请求体（例如：item）。
"""
 
@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

7、Query/Path/Body/Field 参数(额外的校验) 与字符串验证

body，Query，Path方法的父类都是直接或者间接的继承FieldInfo，而Field就FieldInfo的实例化，FieldInfo继承了Representation，它们本质上就是Representation类的子类。
从 fastapi 导入 Query、Path 等对象时，他们实际上是返回特殊类的函数。
使用Path提取和验证路径参数；使用Query提取和验证请求中的参数；使用Body将参数让客户端由body(默认application/json方式)传入。
Field 可用于提供有关字段和验证的额外信息，如设置必填项和可选，设置最大值和最小值，字符串长度等限制
参数说明：
1. Field(None) 是可选字段，不传的时候值默认为None
2. Field(…) 是设置必填项字段
3. title 自定义标题，如果没有默认就是字段属性的值
4. description 定义字段描述内容
5. gt：大于（greater than）
6. ge：大于等于（greater than or equal）
7. lt：小于（less than）
8. le：小于等于（less than or equal）


from fastapi import FastAPI, Query
 
# m是可选参数，参数长度2-10，以name开头
@app.get("/update_items/")
# 当使用 Query 且需要声明一个值是必需的时，可以将 ... 用作第一个参数值
# def update_items(m: Optional[str] = Query(..., max_length=10,min_length=2,regex="^name")):
def update_items(m: Optional[str] = Query(None, max_length=10,min_length=2,regex="^name")):
    results = {"items": [{"oneid": "北京"}, {"two": "上海"}]}
    if m:
        results.update({"上海": m})
    return results
 
 
"""
声明数值校验：
gt：大于（greater than）
ge：大于等于（greater than or equal）
lt：小于（less than）
le：小于等于（less than or equal）
regex正侧参数的写法：
以 ^ 符号开头
以 $ 符号结尾
使用省略号(...)声明必需参数
"""
 
# 需求：id大于5才能返回
@app.get("/id")
async def read_id(*, id: int = Query(..., ge=5, ), q_value: str):
    results = {"message": f"{q_value}"}
    if id:
        results.update({"new_value": q_value})
    return results
 
"""
● 传递 * 作为函数的第一个参数，
● 如果单独出现星号 *，
● 则星号 * 后的参数必须用关键字传入
"""
 
@app.get("/items/{item_id}")
# 使用 Path 为路径参数声明相同类型的校验和元数据
async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results
 
 
"""
●嵌入单个请求体参数,期望一个拥有 item 键并在值中包含模型内容的 JSON,
●返回如下：
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}
●可以使用一个特殊的 Body 参数 embed : item: Item = Body(embed=True)
"""
 
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
 
 
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(embed=True)):
    results = {"item_id": item_id, "item": item}
    return results
 
 
# 通过get方式在URL路径中接收请求参数
@app.get("/items/{item_id}")
async def read_root1(item_id: int = Path(..., gt=0)):
    return {"item_id": item_id}
 
# 虽然是通过post方式提交请求，但item_id仍然是在URL中通过?item_id进行请求
@app.post("/items")
async def read_root2(item_id: int = Query(..., ge=1)):
    return {"item_id": item_id}
 
# post方式提交请求，但item_id仍然是在URL中通过?item_id进行请求
@app.post("/items")
async def read_root2(
	item_id: int = Body(..., ge=1, embed=True),
	item_name: str = Body(None, max_length=20)):
    return {"item_id": item_id, "item_name": item_name}

8、typing类型注解

typing —— 类型注解支持 — Python 3.11.0 文档

8.1 常用类型提示

前两行小写的不需要 import，后面三行都需要通过 typing 模块 import

int,long,float: 整型,长整形,浮点型;
bool,str: 布尔型，字符串类型；
List, Tuple, Dict, Set:列表，元组，字典, 集合;
Iterable,Iterator:可迭代类型，迭代器类型；
Generator：生成器类型；


# 下面的函数接收与返回的都是字符串，注解方式如下：
def items(name: str) -> str:
    return "hello" + name
 
print(items("123"))
 
>>>hello123

8.2 Optional 可选类型

(声明a :Optional[int] = None) ，参数除了给定的默认值外还可以是None（作用是让编译器识别到该参数有一个类型提示，可以使指定类型，也可以是None，且参数是可选非必传的）。注意 Optional[] 里面只能写一个数据类型。即 Optional[X] 等价于 Union[X, None]

typing —— 类型注解支持 — Python 3.11.0 文档


# 参数end是可选择的参数，有end就返回需求的，没有end返回所有的
@app.get("/params")
async def read_param(start: int = 0, end: Optional[int] = None):
    if end:
        return data[start:end]
    return data[start::]

8.3 Union 联合类型


# Union[int, None] 表示既可以是 int，也可以是 None。没有顺序的说法
 
python 3.6 及以上版本，需要导入typing中的Union
from typing import Union
 
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[int, None] = None
 
 
python 3.9 及以上版本 ，不需要导入typing中的Union
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: int | None = None

8.4 typing List

typing 模块中导入 List，可以定义类型sellarea : List[str] = []；
typing 的 List、Set、Tuple 都会指定里面参数的数据类型；
FastAPI 会对声明了数据类型的数据进行数据校验，所以会针对序列里面的参数进行数据校验；
如果校验失败，会报一个友好的错误提示；
使用时需要导入 from typing import List, Tuple；


from typing import Optional
 
import uvicorn
from fastapi import FastAPI, Body
from typing import List, Tuple
 
app = FastAPI()
 
 
@app.put("/items/{item_id}")
async def update_item(
        list_: List[int] = Body(...),
        tuple_: Tuple[int] = Body(...),
):
    results = {"list_": list_, "tuple_": tuple_}
    return results
 
 
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8080)

9、请求示例展示在接口文档中

使用Config 和 schema_extra 为Pydantic模型声明一个简单的示例


import uvicorn
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel, Field
from typing import Union
 
class Items(BaseModel):
    name: str
    desc: Optional[str] = None
    price: float
    tax: Optional[float] = None
    
# 接口请求示例展现在接口文档中
"""使用Config 和 schema_extra 为Pydantic模型声明一个简单的示例"""
    class Config:
        schema_extra = {
            "example": {
                "name": "书名",
                "price": 20,
                "decs": "描述信息",
                "tax": 0.5
            }
        }
 
 
@app.post("/items1")
async def retrun_item(item: Items):
    results = {"item": item}
    return results
 
 
"""
通过工厂函数，增加example参数
注意：传递的额外参数不会添加任何验证，只会添加注释，用于文档的目的
"""
 
class Item(BaseModel):
    name: str = Field(example="Foo")
    description: Union[str, None] = Field(default=None, example="A very nice Item")
    price: float = Field(example=35.4)
    tax: Union[float, None] = Field(default=None, example=3.2)
 
 
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results
 
if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

10 、Cookie，Header参数

Header 是 Path, Query 和 Cookie 的兄弟类型。它也继承自通用的 Param 类；
从fastapi导入 Query, Path, Header, 或其他时，实际上导入的是返回特定类型的函数；


"""
header的必须有token且token必须是456，没有返回无权限，
cookie必须有一个name，且等于123，否则返回认证失败
"""
from typing import Optional
from fastapi import Cookie, FastAPI,Header
app = FastAPI()
@app.get("/items/")
def read_items(name:  Optional[str] = Cookie(None),
               token: Optional[str] =  Header(None)):
    if token is None or token!='456':
        return '无权限'
    if name is None or name !="123":
        return  "认证失败"

11 、响应模型 response_model


class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: Optional[str] = None
 
class Userout(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
 
@app.post("/user/", response_model=Userout)
def create_user(user: UserIn):
    return user

12 、响应状态码-使用 status_code 参数来声明

https://fastapi.tiangolo.com/zh/tutorial/response-status-code/

在任意的路径操作中使用 status_code 参数来声明用于响应的 HTTP 状态码
status_code 参数接收一个表示 HTTP 状态码的数字
@app.post("/items/", status_code=201)

13 、Form表单数据

需预先安装 pip install python-multipart；
使用时需要导入 from fastapi import FastAPI, Form；
Form 是直接继承自 Body 的类；
表单数据的「媒体类型」编码一般为 application/x-www-form-urlencoded；


import uvicorn
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/",status_code=200)
def login(username: str = Form(...), password: str = Form(...)):
    if password == "123456":
        return {"username": username}
    return "密码错误"
    
# 注册用户，username长度8-16位，password长度6-16位，符合需求返回对应username
@app.post("/register", status_code=200)
async def register(username: str = Form(..., min_length=8, max_length=16, regex='^[a-zA-Z]'),
                   password: str = Form(..., min_length=8, max_length=16, regex='^[0-9]')):
    return {"username": username}
 
 
if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

14 、上传文件-File, UploadFile

需要预先安装 pip install python-multipart；
使用时需要导入 fromfastapiimportFastAPI,File,UploadFile；
声明文件体必须使用 File，否则，FastAPI 会把该参数当作查询参数或请求体（JSON）参数；
UploadFile 的属性如下：
1. filename：上传文件名字符串（str）；
2. UploadFile 支持async 方法；
3. file： SpooledTemporaryFile（ file-like 对象）。其实就是 Python文件，可直接传递给其他预期 file-like 对象的函数或支持库；
4. content_type：内容类型（MIME 类型 / 媒体类型）字符串（str）；
UploadFile 与 bytes 相比有更多优势：
1. 使用 spooled 文件：存储在内存的文件超出最大上限时，FastAPI 会把文件存入磁盘；
2. 暴露的 Python SpooledTemporaryFile 对象，可直接传递给其他预期「file-like」对象的库；
3. 自带 file-like async 接口；
4. 可获取上传文件的元数据；
5. 这种方式更适于处理图像、视频、二进制文件等大型文件，好处是不会占用所有内存；
UploadFile 直接继承自 Starlette 的 UploadFile；


from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/files/")
def create(file: bytes = File(...)):
    return {"file_size": len(file)}
 
@app.post("/uploadfile/")
def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

15 、处理错误-HTTPException

使用时需要导入 from fastapi import FastAPI, HTTPException；
触发 HTTPException 时，可以用参数 detail 传递任何能转换为 JSON 的值，不仅限于str；


from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"test": "浅说测试开发"}
@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

16、jsonable_encoder() 函数

https://fastapi.tiangolo.com/zh/tutorial/encoder/#__tabbed_1_2

FastAPI 内部用来转换数据的；
它接收一个对象，比如Pydantic模型，并会返回一个JSON兼容的版本；
使用时需要导入 from fastapi.encoders import jsonable_encoder；

17 、依赖注入-Depends

from fastapi import Depends, FastAPI；
不带括号时，调用的是这个函数本身，是整个函数体，是一个函数对象，不须等该函数执行完成；
带括号（参数或者无参），调用的是函数的执行结果，须等该函数执行完成的结果；
多次使用同一个依赖项：
1. 如果在同一个路径操作多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项；
2. FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」；


from fastapi import Depends
 
# 定义依赖项函数
def com_methods(q: Optional[str] = None, a: int = 0, b: int = 10):
    return {"q": q, "a": a, "b": b}
 
@app.get("/com_item1")
def com_item1(item1: dict = Depends(com_methods)):
    return item1
 
@app.post("/com_item2")
def com_item2(item2: dict = Depends(com_methods)):
    return item2
 
 
"""
class CommonQueryParams:
类实现依赖注入2中写法：
1.commons: CommonQueryParams = Depends(CommonQueryParams)
2.commons: CommonQueryParams = Depends()
"""
 
# 全局都需要校验token
 
from fastapi import  FastAPI,Header, HTTPException,Depends
 
def verify_token(token: str = Header(...)):
    if token!="asdfghjkl":
        raise HTTPException(status_code=400, detail="Token header invalid")
app = FastAPI(dependencies=[Depends(verify_token)])

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/344788

FastAPI 基本使用(一)

1、基本介绍