PS：本来想等自己的公众号数量完所有的并发相关的知识点，再来梳理这个的，鉴于后续公众号还需要些时间整理，所以暂时提前再续前缘讲一下这个框架。

原版的地址： juejin.cn/post/684490…

前置基础

1 Fastapi 安装

pip安装：

pip install fastapi[all] 

pip install uvicorn

或

pip install fastapi

pip install uvicorn

或直接使用IDE安装

2 Fastapi最贱示例

1）编写frun.py模块的代码:

from fastapi import FastAPI
import uvicorn


app = FastAPI()


@app.get('/')
async def login():
    return '大爷'


@app.get('/s')
async def index():
    return '你好'


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
    #  uvicorn xxx:app --reload
    uvicorn.run('frun:app',host="127.0.0.1", port=8000, debug=True, reload=True)

启动的参数的解释：

• 启动的app对象：需要模块名称+app对象名
• host 启动的地址
• post 端口号
• debug 是否开启调试
• reload 是否启动热更新

2）直接的右键启动服务:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [11708] using statreload
INFO:     Started server process [9140]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

3）然后访问： http://127.0.0.1:8000接口

4）打开接口文档： 访问默认的地址:http://127.0.0.1:8000/docs#/

框架基础使用详细开展

1 多路由方法方式

多装饰器一起装饰一个函数

示例：

@app.post("/")
@app.put("/")
@app.delete("/")
@app.get("/")
@app.options("/")
@app.head("/")
@app.patch("/")
@app.trace("/")
async def login():
    return '大爷'

>PS: 它竟然不支持和Flask一样使用methods定义修正

示例：

@app.route('/', methods=['GET','POST'], name='asdasd', include_in_schema=True)
async def loginjig():
    return '大爷'

上面这种方式会引发参数异常，说我缺少参数！但是我已经按规定的指定了还是不支持。

TypeError: loginjig() takes 0 positional arguments but 1 was given

修正一下上面说不支持的@app.route('/a', methods=['GET','POST'], name='a')说法，其实他是支持，只是我们的需要显示的传入必要的参数req:Request在定义的函数上，且我们的函数返回的值必须是一个可调用的对象。 如下正确示例：

@app.route('/a', methods=['GET','POST'], name='a')
def loginjig(req:Request):
    print(req)
    return PlainTextResponse('大爷')

PS:不过上面定义路由的方式的缺点是：在文档里无法显示使用上面形式定义的接口。

2 路由分类

• 静态路由：固定好的请求URL
• 动态路由:可自定可变的请求URL

如静态路由地址,请求的地址为:

• http://127.0.0.1:8000/user/login

@app.get('/user/login')
async def login():
    return '大爷'

如动态路由地址,请求的地址为:

• http://127.0.0.1:8000/user/loginbyuserid/6

@app.get('/user/loginbyuserid/{userid}')
async def loginbyuserid(userid:int):
    return '大爷'

pS：动态路由上对这个传入的参数值是进行严格的校验规定，他必须的是一个int类型的，否则是无法传入的，如操作文档提示：

• 错误的情况
• 正常的情况

关于参数校验后续涉及相关的校验的再叙说。

3 路由URL同名的匹配优先级

在请求的路由中，如果不巧存在同一个路由地址，而且一个静态的，一个动态的话，那么它优先级是啥呢？

如下的示例：

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}

文档接口

请求地址为：http://127.0.0.1:8000/users/me

请求地址为：http://127.0.0.1:8000/users/tytr

更换路由顺序的话: 如下的示例：

from fastapi import FastAPI
app = FastAPI()


@app.get("/users/{user_id}")
async def read_user(user_id: str):
   return {"被优先匹配到：": user_id}

@app.get("/users/me")
async def read_user_me():
   return {"user_id": "the current user"}

请求地址为：http://127.0.0.1:8000/users/me

请求地址为：http://127.0.0.1:8000/users/tytr

说明同URL的情况，路由注册的优先级很关键！

4 路由参数提交分类

4.1路由URL上的参数的类型

• Query查询参数类型
• Path路径参数类型（动态路径的参数）
• 复合型查询参数（Path路径参数+Query查询参数结合）

4.2查询参数类型示例

所谓的查询的参数就是砸url后面自动的添加上的&xxx=xxx&ksakka=yyyy的这样形式的参数。

from fastapi import FastAPI
import uvicorn

app = FastAPI()

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]


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
    #  uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

请求地址

http://127.0.0.1:8000/items/?skip=0&limit=10

文档接口

说明：

• skip: int = 0, limit: int = 10 就是我们的所谓的路基参数，它会?skip=0&limit=10的方式附加到我们的请求的URL上

• skip: int = 0, limit: int = 10 严格的要求我们的参数传入必须的int类型的，否则会报错

传入非int类型参数

4.2复合型查询参数（Path路径参数+Query查询参数结合）

所谓的复合型查询参数就是多路径和查询参数，也就是URL上包含了有动态的参数，还有需要通过&分隔符提交的参数，这情况，通常再GET提交中也很常见，那么如何处理呐？

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: str = 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


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
    #  uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

请求地址

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=True

请求地址

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=False

4.3 参数的必选和可选定义方式

• 参数的可选和必选主要是通过是否给默认值来决定的
• 无默认值则为必填的参数，会对参数进行必填的校验
• 有默认的值的则为选填的参数，不会进行校验
• 可以通过使用Optional来觉得觉得参数的类型，且设置默认值

4.3.1必填的参数的校验示例

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    item = {"item_id": item_id, "needy": needy}
    return item

上述的代码中 needy 没有给与默认的值，当个没提交这个值的时候，会提示错误：

4.3.2使用Optional来定义需要提交的数据类型

from typing import Optional

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, limit: Optional[int] = None):
    item = {"item_id": item_id, "limit": limit}
    return item

参数定义地方：

• item_id: str 没有默认的值，则是必填的
• needy: str 没有默认值的则是必填的

这种没有默认值就是必填的参数其实和我们的自己的定义的函数一样，一个函数的的传参如果没有的默认值的话，那么调用这个函数的时候，就必须是传入的一个参数！

我们把查询参数limit规定为了int类型，但是它是可选的的参数，设置为了None:

4.3.2.1 Optional+Query 且是非必传参数

import uvicorn

from fastapi import FastAPI, Query,Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()


@app.get("/items/{item_id}")
async def read_user_item(item_id: str, limit: Optional[str] = Query(None, min_length=3, max_length=50)):
    item = {"item_id": item_id, "limit": limit}
    return item

上面的示例可以扩展为对我们的limit进一步的确定类型+限制

4.3.2.2 Optional+Query 且是必传参数

：关于Optional和Query在一起的且我们的参数是必填的时候：

PS:当使用Query声明改查询参数必须的时候 可以使用... 作为第一个参

import uvicorn

from fastapi import FastAPI, Query,Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()


@app.get("/items/{item_id}")
async def read_user_item(item_id: str, limit: Optional[str] = Query(..., min_length=3, max_length=50)):
    # 当使用Query声明改查询参数必须的时候 可以使用... 作为第一个参
    item = {"item_id": item_id, "limit": limit}
    return item

上面的示例通常其实我们的没必要使用下面的

limit: Optional[str] = Query(..., min_length=3, max_length=50)

方法来定义必传参数，且需要更多的校验，其实可以简化为:

limit: str = Query(..., min_length=3, max_length=50)

4.4 带‘/’等关键子的路径参数问题

如果存在我们的URL地址上需要传入需要携带的路径的参数值的，会和我们的URL中的'/'进行冲突，如何避免这种冲突呢？

如：

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get('/files/path/{filepath}')
def getfiles(filepath):

    return filepath

请求地址为: http://127.0.0.1:8000/files/path/%2Fyyytyu%2Fretr

修改代码，给参加加上path的修饰：

@app.get('/files/path/{filepath:path}')
def getfiles(filepath):

    return filepath

请求地址为: http://127.0.0.1:8000/files/path/%2Fyyytyu%2Fretr

4.5 路径参数的枚举

路径参数的枚举就是定义一个枚举类，提供默认的选择的输入：

如示例:

import uvicorn
from fastapi import FastAPI
from enum import Enum


class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"


app = FastAPI()


@app.get("/model/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == 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"}

文档示例

4.6 引入Fastapi中封装好的Query来校验查询参数

4.6.1 单个同名的参数值情况

Fastapi中为我们的路基参数定义好了一个Query类来规范参数的校验的明细。 如下示例:

import uvicorn

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results
 
@app.get("/items2/")
async def read_items2(xiaozhong:str='你是大爷',,q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

@app.get("/items3/")
async def read_items3(xiaozhong:bool=False,,q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
    #  uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

上面的参数：

q:str = Query(None, min_length=3,max_length=50),regex="^fixedquery$")

意思是：

• q参数是可选的参数，但是如果填写的话，最大长度必须是小于50内,且最小的长度必须大于3： 且需要符合regex的匹配

当然None可以修改为其他默认值，可以写如：

q: q: str = Query('xiaozhong', min_length=3,max_length=50),regex="^fixedquery$")

对于参数的几种校验:

不传q的情况下:

http://127.0.0.1:8000/items/

传q的情况下且长度大于50:

http://127.0.0.1:8000/items/

传q的情况下且长度小于3:

http://127.0.0.1:8000/items/?q=4

查询参数Query的参数正则校验

4.6.2 多同名的查询参数-Query参数多值列表

一般在我们的接口中很少说同一个参数提交多个值如：

http://localhost:8000/items/?q=foo&q=bar

但也不排查这种情况的存在，所以也可以定义我们的参数类似必须是列表的形式：

from typing import List

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: List[str] = Query(["foo", "bar"])):
    # <!--也可以使用list直接代替List[str]:-->
    query_items = {"q": q}
    return query_items

默认值： 非默认值：

4.7 引入Fastapi中封装好Patht来校验路径参数

对于查询参数可以通过Query，同样对于路径参数也可以使用Fastapi自带的Path来进行校验。

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    q: str, item_id: int = Path(..., title="The ID of the item to get")
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

对于路径参数校验中，还可以对item_id进行大于或等于的校验如：

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
 *, item_id: int = Path(..., title="The ID of the item to get", ge=1), q: str ):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

在上面代码意思是，当ge = 1时，item_id必须是整数“ g大于或等于e等于1”。

4.7 POST或PUT下的结合模型验参

请求体主要涉及的问题，是如何对于的Basemodel进行对应，方便模型进行解析，进行相关的序列化和反序列等操作。

PS 通常非表单的提交的时候我们的boby提交都是使用的请求头是:

application/json的方式

4.7.1 静态路由+简单请求体的校验

对于使用POST的和PUT提交上来的参数，通常会通常会请求体，在postman上体现在：

一般对于Request Body不会通过get提交，对于get提交的参数一般称为是查询参数。所以，如果是通过POTS,PUT等方式提交的参数信息，一般是放到Request Body来提交到我们的后端。

对于如何接收和校验请求体，FastApi提供的形式是使用：

from pydantic import BaseModel

示例如下：

from fastapi import FastAPI
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item):
    return item

在上面的模型中我，定义如果提交的Item它必须是怎么样的一个格式，比如

• name是必选字段
• description是可选且默认为None
• price是必选，且需要是float类型的
• tax是可须且默认为None

那客户端如何提交上面那些参数呐？尝试提交参数什么都不写的情况下：

使用JSON格式提交参数的情况下：

故意提交错误参数格式请求：

4.7.2 URL混合型+简单请求体的校验

Request Body 和 Query 和 Path的混合 在设计一些API过程中难免的可能也会需要综合遇到上述的一些混搭的组合，需要同时多个参数的提交和获取

那么我们通常接收这次参数的话一般怎么接收呐？ 示例代码如：

from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    q: str = None,
    item: Item = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results

通过之前的学习，其实也很简单道理也还是一样，如上的示例请求的话：

其中上面的参数：

• q 是查询类型参数
• item_id 是路径类型参数
• item 是请求体参数

4.7.3复杂组合类型请求体的校验

4.7.3.1 单一结构体

更复杂的业务其实会存在多体的Boay的提交，之前做的商城下单里面，客户端有可能就会同时提交多个实体的对象信息到后端，如订单实体，地址实体，商品信息实体等。

那么在Fastapi如何接受多个Body实体呐？通常以前的话，在bottle,通常直接的request.body 或 request.json就可以获取客户端部提交的信息了。

在Fastapi假设客户端提交的参数是这样的形式：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

那如何的接收处理呐？

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel):
    username: str
    full_name: str = 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

这种情况，其实就是客户端提交多个实体对象。那可以定义多个模型对象即可。fastapi它会自动帮你处理提取信息。

4.7.3.2 单一结构体+单一类型Body参数

如果另外再假设：

在Fastapi假设客户端提交的参数是这样的形式：单一类型Body参数

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

其实这种可能也不是不存在滴，那如何的读取解析importance参数呐？既然参数有Query 和 Path，当然也会有 Body 。

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel):
    username: str
    full_name: str = None


@app.put("/items/{item_id}")
async def update_item(
    *, item_id: int, item: Item, user: User, importance: int = Body(...,gt=0)
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

上面的代码中我们引入了Body 并且在importance: int = Body(...)进行处理和提取：

4.7.3.3 嵌套模型结构体

如果另外再假设,客户端提交一个更复杂的嵌套模型的话，怎么办？麻蛋的 肯定也是会有这样的情况滴！ 嵌套里面有列表有实体。

如：

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": ["rock", "metal", "bar"],
    "image": {
        "url": "http://example.com/baz.jpg",
        "name": "The Foo live"
    }
}

这时候，我们就需要所谓的子内嵌啦：

from typing import Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None
    tags: Set[str] = []
    image: Image = None


@app.put("/items/{item_id}")
async def update_item(*, item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

如上代码，Item里面包含了Image，也包含了，tags类型的列表定义。

4.7.3.4 更复杂嵌套模型结构体

MMP更深层的嵌套也是可以定义的如：

{
    "name":"Foo",
    "description":"The pretender",
    "price":42,
    "items":[
        {
            "name":"Foo",
            "description":"The pretender",
            "price":42,
            "tax":3.2,
            "tags":[
                "rock",
                "metal",
                "bar"
            ],
            "image":{
                "url":"http://example.com/baz.jpg",
                "name":"The Foo live"
            }
        },
        {
            "name":"Foo2",
            "description":"The 2",
            "price":422,
            "tax":3.2,
            "tags":[
                "rock",
                "metal",
                "bar"
            ],
            "image":{
                "url":"http://example.com/baz.jpg",
                "name":"The Foo live"
            }
        }
    ]
}

对应的解析为：

from typing import Set
from typing import List, Set

class Image(BaseModel):
    url: str
    name: str
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None
    tags: Set[str] = []
    # images: List[Image] = None
    image: Image = None


class Offer(BaseModel):
    name: str
    description: str = None
    price: float
    items: List[Item]


@app.post("/offers/")
async def create_offer(*, offer: Offer):
    return offer

4.7.3.5 Request Body的Field类

Field字段的意思其实就是类似上面Query, Path，也同样给Body内的字段的信息添加相关的校验。

也就是说。通过Field来规范提交的Body参数信息。

如：

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = Field(None, title="标题啊",description="错误提示文字啊", max_length=300)
    price: float = Field(..., gt=0, description="错误提示文字啊")
    tax: float = 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

上面的意思就是和之前定义参数校验其实一样

正常情况： 异常情况：

上面的对于的文档可以了解到具体的定义和相关的参数描述：

对于Field定义的标题和描述，在线的Schemas中:

注意：price: float = Field(..., gt=0, description="错误提示文字啊") 中的...是标记为当前的这个字段是必传的字段

如下：注意带*的校验提示

4.8 Form表单提交校验

如果你的设计前端提交表单的的话，那就需要使用表单的方式来获取参数。 fastapi对表单的处理，如果你不是全家桶的安装的话，通常还需要额外的安装先的这个处理，才能处理表单的提交接收。

pip install python-multipart

相关的示例如下:

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

PS 关于表单的提交需要注意的的请求头的提交需使用：

application/x-www-form-urlencoded的方式

4.9 关于输入的模型的校验后转换成字典

如果有必要需要把我们的在输出模型的参数就那些转换的，Fastapi自带了一个转换的处理：

示例如：

import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel
from fastapi.encoders import jsonable_encoder
app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel):
    username: str
    full_name: str = None


@app.put("/items")
async def update_item(item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    print(json_compatible_item_data)
    return json_compatible_item_data

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

打印输出L

{'name': 'string', 'description': 'string', 'price': 0.0, 'tax': 0.0}

4.10 其他类型参数的请求体的校验

其他数据类型¶ 以下是您可以使用的一些其他数据类型（来自官方文档）：

• UUID:
  • 一个标准的“通用唯一标识符”，在许多数据库和系统中常见于ID。
  • 在请求和答复中，将表示为str.
• datetime.datetime:
  • 一只Pythondatetime.datetime.
  • 在请求和答复中，将表示为str采用ISO 8601格式，如：2008-09-15T15:53:00+05:00.
• datetime.date:
  • Pythondatetime.date.
  • 在请求和答复中，将表示为str采用ISO 8601格式，如：2008-09-15.
• datetime.time:
  • 一只Pythondatetime.time.
  • 在请求和答复中，将表示为str采用ISO 8601格式，如：14:23:55.003.
• datetime.timedelta:
  • 一只Pythondatetime.timedelta.
  • 在请求和答复中，将表示为float总秒数。
  • Pydantic还允许将其表示为“ISO 8601时间差异编码”，有关更多信息，请参阅文档。.
• frozenset:
  • 在请求和答复中，将其视为set:
  • 在请求中，将读取列表，消除重复，并将其转换为set.
  • 在答复中，set将转换为list.
  • 生成的架构将指定set值是唯一的(使用JSONSchema的uniqueItems).
• bytes:
  • 标准Pythonbytes.
  • 在请求和答复中将被视为str.
  • 生成的架构将指定它是str带着binary“格式”。
• Decimal:
  • 标准PythonDecimal.
  • 在请求和响应中，处理方式与float.

所以我还可以使用其他类型来校验：

from datetime import datetime, time, timedelta
from uuid import UUID

from fastapi import Body, FastAPI

app = FastAPI()


@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: datetime = Body(None),
    end_datetime: datetime = Body(None),
    repeat_at: time = Body(None),
    process_after: timedelta = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }

5 Headr的操作

5.1 请求头的获取

之前在我们的发生想的表单的时候提到，需要请求提交特殊的请求类型，但是如果有些特殊的需求，我们的项目需要从请求头里获取相关的参数的提交的话，比如有些自定义的请求头，如toekn等，那该如何处理呢？

Fastapi也直接提供好了给我们的封装，类似的Query，Path等一样的类，导入Header即可使用。

示例如：

import uvicorn

from fastapi import FastAPI,Header
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()


@app.get("/Header")
async def Header_handel(ua: Optional[str] = Header(None, convert_underscores=True)):
   # convert_underscores 参数的意思是: 如果请求头的里的参数是带有下划线的话，是否进行转换
   #如 auth_token ===会转换成 auth-toekn
    return ua

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
    #  uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

运行示例:

PS：在获取请求头的时候，我们需要注意一点就是关于写请求头的带斜杠的问题。

之前我Flask框架的时候，遇到过自定义的请求体在其他浏览器是正常的获取到的，但是在UC的浏览器，就扑街了！死活拿不到，因为在Python中，如User-Agent这种方式的参数，它还认为不是一个合法的变量名称。，所以为了能正常获取到请求头的值，Header模块会把参数名称中的"_"转换成"-"。所以涉及到时一个斜杠转换的问题。上面的示例中：convert_underscores 参数的意思是: 如果请求头的里的参数是带有下划线的话，是否进行转换

#如 auth_token ===会转换成 auth-toekn

convert_underscores =True 表示转换 =False 表示转换

补充说明：Http Headers大小写不敏感的滴，通常会默认转换

同名多指的请求头的处理，主要是这种情况和查询参数多值提交的一样的，这种情况，我们也可以按照之前的类似查询参数的多值处理的方式，定义为列表，来获取：

如下示例：

import uvicorn

from fastapi import FastAPI,Header,Cookie,Response
from pydantic import BaseModel, Field
from typing import List

app = FastAPI()


@app.get("/headerlist/")
async def read_headerlist(x_token: List[str] = Header(None)):
    return {"X-Token values": x_token}

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

5.2 响应头的设置

有请求头也，肯定有响应头:

这里就涉及都我们的如何设置我们的自定义的响应头了，以前做的Flask的跨域的处理的时候，其实就是想办法在响应体里添加入支持跨域的请求的响应头。所以我们有必要要学会如何写入自定义响应头。

对于如何写入呢？按官方文档，给我们展示两个方案. 方案1：使用Response

import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/set_response_headers")
def set_response_headers(response: Response):
    response.headers["X-Token"] = "888888888888888888888888888"
    response.headers["Access-Control-Allow-Methods"] = " GET, OPTIONS, HEAD, PUT, POST"
    return {"message": "OK"}

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

方案2：使用JSONResponse

import uvicorn

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse

app = FastAPI()


@app.get("/set_response_headers")
def set_response_headers():
    content = {"message": "OK"}

    headers = {
        "X-Token": "888888888888888888888888888",
        "Access-Control-Allow-Methods": " GET, OPTIONS, HEAD, PUT, POST"
    }
  
    return JSONResponse(content=content, headers=headers)


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

6 Cookie的操作

6.1 Cookie的操作-写入

对于依赖前端提交的Cookie来说，写入我们的对于的Cookie值，就需要在响应报文里处理。 因为这里涉及响应报文的处理，如果我们的在使用模型直接输出的话，通常需要封装处理下。

对于如何写入呢？按官方文档，给我们展示两个方案.

方案1：使用Response

import uvicorn

from fastapi import FastAPI,Header,Cookie,Response
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()


@app.get("/Cookie")
async def Cookier_handel(cookie: Optional[str] = Cookie(None)):

    return cookie


@app.get("/set_cookie/")
def setcookie(response: Response):
    response.set_cookie(key="xiaozhong", value="sdasdasdaxxxxxxxxxxxxxxxxxxxda")
    return '设置Cookie成功'

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

方案2：使用JSONResponse

import uvicorn

from fastapi import FastAPI,Header,Cookie,Response
from pydantic import BaseModel, Field
from typing import Optional

from fastapi.responses import JSONResponse
app = FastAPI()


@app.get("/Cookie")
async def Cookier_handel(cookie: Optional[str] = Cookie(None)):

    return cookie


@app.get("/set_cookie/")
def setcookie(response: Response):
    response.set_cookie(key="xiaozhong", value="sdasdasdaxxxxxxxxxxxxxxxxxxxda")
    return '设置Cookie成功'

@app.getcookiebyjson("/getcookiebyjson/")
def setcookieyjson():
    content = {"msg": "大爷设置Cookie"}
    response = JSONResponse(content=content)
    response.set_cookie(key="xiaozhong", value="sdasdasdaxxxxxxxxxxxxxxxxxxxda")
    return response

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

6.2 Cookie的操作-获取

通常我们的前端请求发起的时候通常会提交对于的前端的Cookie值上来做相关的验证。 其实Fastapi也直接提供好了给我们的封装，类似的Query，Path等一样的类，导入Cookie即可使用，和获取请求头的方式一样。

我们的cookie一般其实也是在请求头里提交:

示例如：

import uvicorn

from fastapi import FastAPI,Header,Cookie
from typing import Optional

app = FastAPI()


@app.get("/Cookie")
async def Cookier_handel(cookie: Optional[str] = Cookie(None)):

    return cookie

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

7 使用模型来做响应报文体

7.1 使用response_model定义

请求一个接口返回来我们客户端可见的东西都是所谓的响应报文，如响应头，响应码，响应内容等。

通常不会那么傻的用户输入什么就返回什么。以下的官网示例纯粹的演示看：

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()



class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: str = None


class UserOut(BaseModel):
    username: str
    email: str
    full_name: str = None


@app.post("/user/", response_model=UserOut)
async def create_user(*, user: UserIn):
    return user

请求之后，获取到是UserOut的内容信息：

response_model就是我们的响应的报文体，但是一般如果是做API的开发的，除非的你的特殊接口，通常我们的会按API的固定的格式返回我们的响应报文，而不是直接的输出一个具体的模型，当然如果 你的你的这个输出的模型就是你的JSON格式的也是可以的如

class UserOut(BaseModel):
    code: str
    message: str
    data: list = None

或者是下面的这种形式,把response_model为一个字典返回:

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.get("/keyword-weights/", response_model=Dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

7.2 response_model中的包含(排除剩下的)和排除

• response_model_exclude：如果我们的输出的response_model有部分的字段需要过滤
• response_model_include：如果我们的输出的response_model必须纯在某个字段的，则使用它来包含。
• response_model_exclude_unset：不返回没有设置的参数（包括如果是0的参数）
• response_model_exclude_defaults：不返回是默认值的字段
• response_model_exclude_none：不返回是None的字段（response_model_exclude_none=True）

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
import uvicorn
app = FastAPI()


class UserIn(BaseModel):
   username: str
   password: str
   email: str
   full_name: str = None


class UserOut(BaseModel):
   username: str
   email: str
   full_name: str = None

# 必须包含name
@app.post("/user1/", response_model=UserOut, response_model_include={"username", "full_name"})
async def create_user1(*, user: UserIn):
   return user


# 过滤email
@app.post("/user2/", response_model=UserOut, response_model_exclude={"email", })
async def create_user2(*, user: UserIn):
   return user


if __name__ == '__main__':
   # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
   uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

7.3 关于响应状态码status_code

通常的一个接口请求完成，如果没有什么异常通常会返回200： 如日志打印出来一样：

INFO:     127.0.0.1:58141 - "POST /user/ HTTP/1.1" 400
INFO:     127.0.0.1:58315 - "POST /user/ HTTP/1.1" 200

FastAPI运行我们的指定返回的status_code

7.3.1-方案1:

如下示例：

@app.post("/user/", response_model=UserOut,status_code=500)
async def create_user(*, user: UserIn):
    return user

导致请求的接口返回：

6.3.2-方案2:

甚至还可以通过导入status来指定：

from fastapi import FastAPI, status

app = FastAPI()


@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}
    

7.3.3-方案3:

还可以在我们的

@app.get("/set_response_headers")
def set_response_headers(response: Response):
    response.headers["X-Token"] = "888888888888888888888888888"
    response.headers["Access-Control-Allow-Methods"] = " GET, OPTIONS, HEAD, 
    PUT, POST"
    response.status_code = status.HTTP_201_CREATED
    return {"message": "OK"}

7.3.4-方案4:

还可以在我们的

@app.get("/set_response_headers")
def set_response_headers():
    content = {"message": "OK"}

    headers = {
        "X-Token": "888888888888888888888888888",
        "Access-Control-Allow-Methods": " GET, OPTIONS, HEAD, PUT, POST"
    }

    return JSONResponse(status_code=status.HTTP_201_CREATED, content=content, headers=headers)

8 异常错误的处理

8.1 HTTPException异常抛出

再之前Bottle 中其实有一个就是HttpError异常类，在FastAPI也存在这么一个HTTPException。

如示例：

from fastapi import FastAPI, HTTPException

app = 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_code=404, detail="Item not found")
    return {"item": items[item_id]}

在上面的代码中，通过判断item_id是不是存在于items来主动的抛出了一个404的错误

我们查看HTTPException和StarletteHTTPException的源码发现他们也是继承与Exception：

class HTTPException(StarletteHTTPException):
    def __init__(
        self, status_code: int, detail: Any = None, headers: dict = None
    ) -> None:
        super().__init__(status_code=status_code, detail=detail)
        self.headers = headers

所以我们对于异常通常可以直接的使用 raise来抛出异常。

8.2 HTTPException且返回新增自定义请求头

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

8.3 自定义返回HTTPException

类似之前Bottle我们通过添加一个自定义的全局的错误，来统一的处理返回。FastAPI其实也提供一个自定义错误的机制：

官方示例如下：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}
    

观察请求结果：

当请求name == yolo的时候，我们主动抛出了UnicornException，而且我们，@app.exception_handler(UnicornException)也捕获到相关的异常信息，且返回了相关的信息。

8.4 FastAPI HTTPException 和 Starlette HTTPException

• FastAPI HTTPException 继承自 Starlette's HTTPException。
• HTTPException和StarletteHTTPException继承与Exception：

FastAPI HTTPException 和 Starlette HTTPException的区别：

• FastAPI HTTPException允许你在response添加头信息。主要在内部用于OAuth 2.0以及一些安全相关的功能。
• Starlette HTTPException 无法添加头信息

所以，通常我们在代码中抛出FastAPI HTTPException异常。

但是，当我们注册异常处理器的时候，我们应该注册为Starlette HTTPException。

这样，当Starlette的内部代码或者Starlette扩展插件抛出Starlette HTTPException时，我们的处理器才能正常捕获和处理这个异常。

8.5 覆盖FastAPI默认的异常处理

按官方文档说明就是，当请求包含无效的数据的时候，或参数提交异常错误的时候，会抛出RequestValidationError，

那其实我也可以通过上面的自定义异常的方式来覆盖重写我们的RequestValidationError所返回信息：

如： 默认代码没有添加覆盖处理的话： 发生异常的时候是提示是：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     return JSONResponse({'mes':'触发了RequestValidationError错误，，错误信息:%s 你妹的错了！'%(str(exc))})



@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}



if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

发生异常的请求下返回：

恢复覆盖的时候：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse({'mes':'触发了RequestValidationError错误，，错误信息:%s 你妹的错了！'%(str(exc))})


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}



if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

请求结果：

上面的返回其实我们还可以修改一下返回如下，指定响应码：

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

说明：

# 注意fastapi包中的HTTPException才可以定义请求头
from fastapi import Depends, status, HTTPException
# from starlette.exceptions import HTTPException

8.6 RequestValidationError类型的判断

后续如果我们的有必要的进行使用反函数类型进行匹配相关的错误类型的话，定制自己的中文错误的返回的话，就需要判断RequestValidationError类型，比较这错误类型比较多。

如：

from fastapi import FastAPI, HTTPException,status,Response
from fastapi.exceptions import RequestValidationError
from pydantic.errors import *
from pydantic import ValidationError
class RequestValidationErrorException():

    @staticmethod
    def excaction(exc=RequestValidationError) ->JSONResponse:
        print("参数提交异常错误selfself", exc.errors())
        print("参数提交异常错误selfself", exc.errors()[0].get('loc'))
        # 路径参数错误
        if 'path' in exc.errors()[0].get('loc'):
            # 判断错误类型
            if isinstance(exc.raw_errors[0].exc,IntegerError):
                return CustomizeParameterException(msg='%s 参数 %s 类型错误,必须是Integer类型'% (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1]))
            elif isinstance(exc.raw_errors[0].exc,MissingError):
                return CustomizeParameterException(msg='%s 参数 %s 缺失,参数是比传参数' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1]))
            elif isinstance(exc.raw_errors[0].exc,NumberNotLeError):
                return CustomizeParameterException(msg='%s 参数 %s 有限制,参数必须小于等于 %s ' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1],exc.errors()[0].get('ctx').get('limit_value')))
            elif isinstance(exc.raw_errors[0].exc, NumberNotLtError):
                return CustomizeParameterException(msg='%s 参数 %s 有限制,参数必须小于 %s ' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1], exc.errors()[0].get('ctx').get('limit_value')))
            else:
                CustomizeParameterException(msg='路径参数错误,请核查参数提交格式和要求')
        # body参数模型校验类型的错误
        elif 'body'in exc.errors()[0].get('loc') :
           pass
           if isinstance(exc.raw_errors[0].exc, ValidationError):

             

            # return CustomizeParameterException(msg='路径参数 %s 类型错误,必须是Integer类型' % (exc.errors()[0].get('loc')[1]))

        return CustomizeParameterException()

8.7 继承JSONResponse，进行各自响应体的返回

使用flask的中我习惯的方式来返回我们的响应体报文。

from typing import Any, Dict


# 自定义返回的错误的响应体信息
from fastapi.responses import PlainTextResponse,JSONResponse

class ApiResponse(JSONResponse):
    # 定义返回响应码--如果不指定的话则默认都是返回200
    http_status_code = 200
    # 默认成功
    code = 0
    data = None  # 结果可以是{} 或 []
    msg = '成功'

    def __init__(self,http_status_code=None,  data=None,msg=None, **options):

        if data:
            self.data = data
        if msg:
            self.msg = msg

        if http_status_code:
            self.http_status_code = http_status_code

        # 返回内容体
        body = dict(
            msg=self.msg,
            code=self.code,
            data=self.data,
        )
        super(ApiResponse, self).__init__(status_code=self.http_status_code,content=body, **options)



class BadrequestException(ApiResponse):
    http_status_code = 400
    #  error_code = 10032
    code = 10032
    msg = '错误的请求'


class ParameterException(ApiResponse):
    http_status_code = 400
    code = 400
    msg = '参数校验错误'

class UnauthorizedException(ApiResponse):
    http_status_code = 401
    code = 401
    msg = '未经许可授权'


class ForbiddenException(ApiResponse):
    http_status_code = 403
    code = 403
    msg = '当前访问没有权限'


class NotfoundException(ApiResponse):
    http_status_code = 404
    code = 404
    msg = '访问地址不存在'


class MethodnotallowedException(ApiResponse):
    http_status_code = 405
    code = 405
    msg = '不支持使用此方法提交访问'


class OtherException(ApiResponse):
    http_status_code = 800
    code = 800
    msg = '未知的其他HTTPEOOER异常'
    error_code = 10034


class InternalErrorException(ApiResponse):
    http_status_code = 500
    code = 500
    data = None  # 结果可以是{} 或 []
    # msg = 'Internal Server Error'
    msg = ' 服务崩溃异常'

class RateLimitApiException(ApiResponse):
    http_status_code = 429
    code = 429
    data = None  # 结果可以是{} 或 []
    # msg = 'Internal Server Error'
    msg = '请求次数受限'


class CustomizeApiResponse(ApiResponse):
    http_status_code = 200
    code = 200
    data = None  # 结果可以是{} 或 []
    # msg = 'Internal Server Error'
    msg = '成功'


class CustomizeParameterException(ApiResponse):
    http_status_code = 200
    code = 200
    msg = '参数校验错误'
    error_code = 10031

9：FastAPI 中间件

9.1 中间件简要说明

所谓的中间件，其实和我们bottle中的中间件作用是一致。有些方法或操作需要在所有路由之前执行，比如要加一个http访问的拦截器，可以对部分接口API需要授权才能访问的接口进行验证之类的。

FastAPI提供了一个@app.middleware("http")可以做到类似上面的拦截功能。其实和bottle或flask 钩子函数很相似

示例如下：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

import time
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse({'mes':'触发了RequestValidationError错误，，错误信息:%s 你妹的错了！'%(str(exc))})


@app.get("/items/{item_id}")
async def read_item(item_id: int):

    return {"item_id": item_id}


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

然后我们请求完成后发现，我们的响应头里多了一个新增的请求头：

9.2 HTTPSRedirectMiddleware中间件

官网的示例中讲述，这个中间件HTTPSRedirectMiddleware主要是强制我们的请求协议必须是https或者wss。意思就是需要安全传输。

示例如：

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

app.add_middleware(HTTPSRedirectMiddleware)


@app.get("/")
async def index():
    return {"message": "OK"}

9.3 TrustedHostMiddleware中间件

rustedHostMiddleware它强制我们发来的请求必须在请求头Header信息中设置了Host选项，主要是为了避免HTTP Host Header攻击。

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"]
)


@app.get("/")
async def main():
    return {"message": "Hello World"}

9.3 GZipMiddleware中间件

主要是起到资源压缩的作用。

它的机制是:当请求的头信息 Accept-Encoding 字段带有"gzip"时，GZipMiddleware负责完成相应的返回结果处理。

GZipMiddleware 支持 minimum_size 参数：当返回结果大小小于指定值时不启用压缩。(单位为字节，默认值为500)

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

app.add_middleware(GZipMiddleware, minimum_size=1000)

10：FastAPI 中间件方式的跨域处理

为啥需要跨域处理，通常我们的API一般是给到前端去调用，但是前端可能使用域名和没提供的API域名是不一样，这就引发了浏览器同源策略问题，所以我们需要做跨域请求支持。

更专业的说法：来自developer.mozilla.org/en-US/docs/…

跨源资源共享(CORS) 跨源资源共享 (CORS)是Http-基于标头的机制，允许服务器指示任何其他起源S(域、方案或端口)，而不是浏览器应该允许加载资源的自身。CORS还依赖于浏览器向承载跨源资源的服务器发出“预飞”请求的机制，以检查服务器是否允许实际请求。在该预运行中，浏览器发送指示将在实际请求中使用的HTTP方法和标头的标头。

跨源请求的示例：domain-a.com使用XMLHttpRequest请求https://domain-b.c….

出于安全考虑，浏览器限制从脚本启动的跨源HTTP请求.例如,XMLHttpRequest而取API跟随同源政策。这意味着使用这些API的Web应用程序只能从应用程序加载的相同来源请求资源，除非其他来源的响应包含正确的CORS头。

CORS机制支持浏览器和服务器之间的安全跨源请求和数据传输。现代浏览器在api中使用cors，例如XMLHttpRequest或去取以减少跨源HTTP请求的风险。

FastAPI支持跨域的话，可以通过添加中间的形式，和bottle也有相似之处。不仅如此他还支持仅限于支持哪些域名进行跨域请求：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def main():
    return {"message": "Hello World"}

懒得起一个js访问了，所以这个暂时不测试了，后期有机会再测试验证一下，感觉应该就是这样的。

11：FastAPI 依赖注入之Depends（补充）

看官网的描述Depends的使用，似乎比较懵懵懂懂的，于是乎还是需要花些时间再次学习一下关于依赖注入。

首先依赖注入它可以是函数也可以是类，如下面的函数形式的依赖注入：

11.1 简单的依赖说明

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: 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)):
    commons.update({'小钟':'同学'})
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)

梳理一下接口请求的流程：

• 1： 上面的commons: dict = Depends(common_parameters)它声明了一个依赖关系： Depends(common_parameters)： 这对接口的依赖进行了一个声明，表示的是接口参数请求依赖于common_parameters的函数。

  当接口被调用的时候，回调用common_parameters函数进行请求处理。

• 2： common_parameters函数主要是负责接收函数，处理后返回一个字典，

• 3：然后把Depends(common_parameters)返回的结果 传递给commons: dict，这个就是一个依赖注入的过程。

所以在上面的示例中common_parameters是我们被依赖对象

这个被依赖的对象，对接口请求的要求就是:

• 可选查询参数q那是一个str.
• 可选查询参数skip那是int，默认情况下是0.
• 可选查询参数limit那是int，默认情况下是100.
• 返回一个字典

请求示例：

这依赖注入的方式其实也挺方便，类似于接口装饰器的方式，比如common_parameters中我们可以先对相关的参数进行校验拦截，还可以再传递。 场景可以和我们之前的bottle的装饰器差不多类似：

• 相同的逻辑判断处理
• 用户身份鉴权

PS：依赖注入的意思就是：被注入的函数依赖于我的这个函数，没有我的这个依赖就无法运行下去。

11.2 把类当作被依赖对象

上面我们的被依赖的对象是以函数的形式出现，那FastAPI它也支持以类的形式来表达。按官网说法被依赖对象它应该是必须一个可以调用的对象比如：类，函数之类的···

这里看一下以类的形式：

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    # 如果q存在
    if commons.q:
        # 我们就把q加到一个新字典
        response.update({"q": commons.q})
        response.update({"小钟": '同学'})
    #然后在我们的fake_items_db进行截取
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

上面我们的CommonQueryParams是一个类，它和我的函数其实差不多，当我们的接口被调用的时候，类对象就回被初始化， 按官网的说法： commons: CommonQueryParams = Depends(CommonQueryParams) 和 commons = Depends(CommonQueryParams) 是等价的。 还有一种是： commons: CommonQueryParams = Depends()

PS：既然的初始化的操作，那我们的可以在我们的依赖注入里面就可以定义自己的一些校验方法来进行相关的校验。 所有我们的依赖注入其实也可以当做某种类似的权限的校验前提。

示例运行演示

有Q参数：

没有Q参数：

11.3 多层嵌套依赖

多层嵌套的意思就是可以类可以类的意思。函数可以依赖函数。其实和我们的之前的参数校验一样。

比如下面的代码：

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str = None):
   return q


def query_or_cookie_extractor(
   q: str = Depends(query_extractor), last_query: str = Cookie(None)
):
   if not q:
       return last_query
   return q


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
   return {"q_or_cookie": query_or_default}

其实意思就是query_or_cookie_extractor 依赖于query_extractor，然后query_or_cookie_extractor被注入到接口上也被依赖的对象。

官网的截图上面接口执行流程如下：

对于同一个依赖，如果处理的结果是一样的，就是返回值是一样的话，我们可以进行多次调用依赖，这时候可以对被依赖的对象设置是否使用缓存机制：

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
  return {"fresh_value": fresh_value}

11.4 在Header请求头里的list列表依赖，

我们先看官方提供的示例代码：

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header(...)):
 if x_token != "fake-super-secret-token":
     raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
 if x_key != "fake-super-secret-key":
     raise HTTPException(status_code=400, detail="X-Key header invalid")
 return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
 return [{"item": "Foo"}, {"item": "Bar"}]

上述的代码意思是对我们的请求头部信息Header进行验证，因为示例是...三个点，说明是必选的字段：

分析上述的代码之后，运行一下试一试看看结果：

1：什么头部参数都不传递的情况提示，我们的头部参数异常

2：头部参数填写的情况：

注意点：参数提交的格式，因为是头部的参数，所以我们的代码上的x_token 会应该要写：x-token才对

错误的示例：

所以上面列表的依赖的意思就是必须两天条件都成立才通过。这个感觉后期还是用到的比较多的哟！

11.5 多依赖对象注入和列表其实是一样的：

from fastapi import Depends, FastAPI


from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi import Depends, FastAPI

app = FastAPI()




async def verify_token(x_token: str = Header(...)):
  if x_token != "fake-super-secret-token":
      raise HTTPException(status_code=400, detail="X-Token header invalid")
  return x_token

async def verify_key(x_key: str = Header(...)):
  if x_key != "fake-super-secret-key":
      raise HTTPException(status_code=400, detail="X-Key header invalid")
  return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
  return [{"item": "Foo"}, {"item": "Bar"}]


@app.get("/items2/")
async def items2(xt: str = Depends(verify_token),xk: str = Depends(verify_key)):
 return {"xt": xt,'xk':xk}



if __name__ == '__main__':
  import uvicorn
  uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)

如上面的xt: str = Depends(verify_token),xk: str = Depends(verify_key)，也是需要必须两个条件成立才可以。

正常情况： 非正常情况：

12：FastAPI 文件上传接收处理（补充）

对于文件的上传的接收处理，也是常规的操作，Fastapi其实对于这些方面的考虑其实已经很周到了，也给我们封装了对于的处理机制。

注意文件上传也依赖于:

pip install python-multipart

• File 接收文件，File对象接收的是字节类型的bytes,它会使用bytes的方式读入内存，仅仅合适于小的文件的处理。

• UploadFile适合上传图片其他大文件，他内部的机制是也写入到内存中，但是当读取文件到一定的阈值后，会自动的保存到磁盘，不会出现内存爆满的情况。另外他还可以获取上传文件的很多元数据信息，且有文件对象的异步操作处理，可以异步的读取文件和写入。当然如果可以的话，对于的使用aiofile来处理也可以！

import uvicorn

from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
from fastapi.encoders import jsonable_encoder
from typing import List
app = FastAPI()

# 单文件的上传
@app.post("/files")
async def files(file_obj: bytes = File(...)):
    return {'文件大小': len(file_obj)}


# 多文件的上传
@app.post("/files2")
async def files2(file_objs:List[bytes] = File(...)):
    return {'文件大小': len(file_objs)}


@app.post("/uploadfiles1")
async def files3(file_obj: UploadFile = File(...)):
    return {'文件大小': len(file_obj)}

@app.post("/uploadfiles2")
async def files4(file_objs: List[UploadFile] = File(...)):
    return {'文件大小': len(file_objs)}


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

查看文档：

13：FastAPI 后台任务-BackgroundTasks（补充）

后台异步的任务顾名思义就是，我一个查询不需要等到厚爱任务的执行完成，我就直接的返回，比如我一些后台同步的任务，只需要发起同步的请求，就返回告诉我已经提交执行任务就可以了！

13.1 后台任务简单示例

示例如:

import uvicorn

from fastapi import FastAPI, File, UploadFile


app = FastAPI()

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def tasks(msg):
    print('耗时任务开始执行',msg)
    import time
    time.sleep(5)
    print('耗时任务开始结束', msg)


@app.post("/backgroundTasks")
async def runBackgroundTasks(msg: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(tasks,msg)
    return "任务已经再处理中！"

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

输出的结果：

耗时任务开始执行 爷爷
间隔了5秒
耗时任务开始结束 爷爷

13.2 使用依赖注入方式实现后台任务

def tasks(msg):
    print('耗时任务开始执行', msg)
    import time
    time.sleep(5)
    print('耗时任务开始结束', msg)


def tasksaction(background_tasks: BackgroundTasks, msg: str):
    message = f"你叶大爷: {msg}\n"
    background_tasks.add_task(tasks, message)


@app.post("/backgroundTasks")
async def runBackgroundTasks(msg: str = Depends(tasksaction)):
    print(msg)
    return "任务已经再处理中！"


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

PS:这种后台任务，他的底层是开启线程的方式处理的！

14：FastAPI 普通函数和异步函数同时存在运行分析（补充）

示例分析：

import uvicorn

from fastapi import FastAPI, File, UploadFile
from fastapi import BackgroundTasks, FastAPI, Depends

app = FastAPI()

import threading
@app.get("/sync")
def sync():
    print('sync-当前线程的名称:',threading.current_thread().getName())

    return "sync任务已经处理完成！"


@app.get("/async")
async def asyncaction():
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成！"


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

输出测试结果:

sync-当前线程的名称: ThreadPoolExecutor-0_0
INFO:     127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK
sync-当前线程的名称: ThreadPoolExecutor-0_1
INFO:     127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK
sync-当前线程的名称: ThreadPoolExecutor-0_2
INFO:     127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK
asyncaction-当前线程的名称: MainThread
INFO:     127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OK
asyncaction-当前线程的名称: MainThread
INFO:     127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OK
asyncaction-当前线程的名称: MainThread
INFO:     127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OK

从输出的结果可以看，异步函数的使用的协程的方式，同步到额方式是实现是使用线程池的方式。 官网的建议就是：如果你不知道该使用异步还是同步，最好的就是全部使用同步，避免异步里面使用了同步，这样的协程的下的单线程的方式就卡死！因为协程是单线程的运行滴！

总结

希望冷饭再炒，还是别有一点点风味！以上纯属个人的学习笔记，如有纰漏还希望多多指点。

这一次的真的希望会坚持继续出下一集，主要是整理自己的使用过程的中的提炼的一些使用小经验。

END

小钟同学 | 文 【原创】【转载请联系本人】| QQ：308711822