FastAPI：Python 世界里最受欢迎的异步框架

2021-10-23 03:28:48

楔子

这次我们来聊一聊 FastAPI，它和我们之前介绍的 Sanic 是类似的，都是 Python 中的异步 web 框架。相比 Sanic，FastAPI 更加的成熟、社区也更加的活跃，那么 FastAPI 都有哪些特点呢？

  • 快速：拥有非常高的性能，归功于 Starlette 和 Pydantic；Starlette 用于路由匹配，Pydantic 用于数据验证
  • 开发效率：功能开发效率提升 200% 到 300%
  • 减少 bug：减少 40% 的因为开发者粗心导致的错误
  • 智能：内部的类型注解非常完善，编辑器可处处自动补全
  • 简单：框架易于使用，文档易于阅读
  • 简短：使代码重复最小化，通过不同的参数声明实现丰富的功能
  • 健壮：可以编写出线上使用的代码，并且会自动生成交互式文档
  • 标准化：兼容 API 相关开放标准

FastAPI 最大的特点就是它使用了 Python 的类型注解，我们后面会详细说，下面来安装一下 FastAPI。

使用 FastAPI 需要 Python 版本大于等于 3.6。

首先是 pip install fastapi，会自动安装 Starlette 和 Pydantic；然后还要 pip install uvicorn，因为 uvicorn 是运行相关应用程序的服务器。或者一步到胃：pip install fastapi[all]，会将所有依赖全部安装。

请求与响应

我们来使用 FastAPI 编写一个简单的应用程序：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

# 类似于 app = Flask(__name__)
app = FastAPI()

# 绑定路由和视图函数
@app.get("/")
async def index():
    return {"name": "古明地觉"}


# 在 Windows 中必须加上 if __name__ == "__main__"，否则会抛出 RuntimeError: This event loop is already running
if __name__ == "__main__":
    # 启动服务，因为我们这个文件叫做 main.py，所以需要启动 main.py 里面的 app
    # 第一个参数 "main:app" 就表示这个含义，然后是 host 和 port 表示监听的 ip 和端口
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

整个过程显然很简单，然后我们我们在浏览器中输入 "localhost:5555" 就会显示相应的输出，我们看到在视图函数中可以直接返回一个字典。当然除了字典，其它的数据类型也是可以的，举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/int")
async def index1():
    return 666

@app.get("/str")
async def index2():
    return "古明地觉"

@app.get("/bytes")
async def index3():
    return b"satori"

@app.get("/tuple")
async def index4():
    return ("古明地觉", "古明地恋", "雾雨魔理沙")

@app.get("/list")
async def index5():
    return [{"name": "古明地觉", "age": 17}, {"name": "古明地恋", "age": 16}]


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

这里我们直接使用 requests 发请求：

FastAPI：Python 世界里最受欢迎的异步框架

我们看到基本上都是支持的，只不过元组自动转成列表返回了。这里我们在路由中指定了路径，可以看到 FastAPI 中的路径形式和其它框架并无二致，只不过目前的路径是写死的，如果我们想动态声明路径参数该怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id):
    """和 Flask 不同，Flask 是使用 <>，而 FastAPI 使用 {}"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

整体非常简单，路由里面的路径参数我们可以放任意个，只是 {} 里面的参数必须要在定义的视图函数的参数中出现。但是问题来了，我们好像没有规定类型啊，如果我们希望某个路径参数只能接收指定的类型要怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/apple/{item_id}")
async def get_item(item_id: int):
    """和 Flask 不同，Flask 定义类型是在路由当中，也就是在 <> 里面，变量和类型通过 : 分隔
       而 FastAPI 是使用类型注解的方式，此时的 item_id 要求一个整型（准确的说是一个能够转成整型的字符串）"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

如果我们传递的值无法转成整型的话，那么会进行提示：告诉我们 value 不是一个有效的整型，可以看到给的提示信息还是非常清晰的。

所以通过 Python 的类型声明，FastAPI提供了数据校验的功能，当校验不通过的时候会清楚地指出没有通过的原因。在我们开发和调试的时候，这个功能非常有用。

交互式文档

FastAPI 会自动提供一个类似于 Swagger 的交互式文档，我们输入 "localhost:5555/docs" 即可进入。

FastAPI：Python 世界里最受欢迎的异步框架

有兴趣可以自己尝试测试一下，然后我们注意一下里面的 /openapi.json，我们可以点击进去，会发现里面包含了我们定义的路由信息。

FastAPI：Python 世界里最受欢迎的异步框架

至于 "localhost:5555/docs" 页面本身，我们也是可以进行设置的：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI(title="测试文档",
              description="这是一个简单的 demo",
              docs_url="/my_docs",
              openapi_url="/my_openapi")

@app.get("/apple/{item_id}")
async def get_item(item_id: int):
    """和 Flask 不同，Flask 定义类型是在路由当中，也就是在 <> 里面，变量和类型通过 : 分隔
       而 FastAPI 是使用类型注解的方式，此时的 item_id 要求一个整型（准确的说是一个能够转成整型的字符串）"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我们再重新进入，此时就需要通过 "localhost:5555/my_docs"：

FastAPI：Python 世界里最受欢迎的异步框架

整体没什么难度，我们还可以指定其它参数，比如 version 表示版本，可以自己试试。该页面主要用来测试自己编写的 API 服务，不过个人更喜欢使用 requests 发请求。

路由顺序

然后我们在定义路由的时候需要注意一下顺序，举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

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: int):
    return {"user_id": user_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

因为路径操作是按照顺序进行的，所以这里要保证 /users/me 在 /users/{user_id} 的前面，否则的话只会匹配到 /users/{user_id}，此时如果访问 /users/me，那么会返回一个解析错误，因为字符串 "me" 无法解析成整型。

使用枚举

我们可以将某个路径参数通过类型注解的方式声明为指定的类型（准确的说是可以转成指定的类型，因为默认都是字符串），但如果我们希望它只能是我们规定的几个值之一该怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from enum import Enum
from fastapi import FastAPI
import uvicorn

app = FastAPI()

class Name(str, Enum):
    satori = "古明地觉"
    koishi = "古明地恋"
    marisa = "雾雨魔理沙"

@app.get("/users/{user_name}")
async def get_user(user_name: Name):
    return {"user_id": user_name}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

通过枚举的方式可以实现这一点，我们来测试一下：

FastAPI：Python 世界里最受欢迎的异步框架

结果和我们期望的是一样的，我们可以再来看看 docs 生成的文档：

FastAPI：Python 世界里最受欢迎的异步框架

提示我们，可以用的值：古明地觉、古明地恋、雾雨魔理沙。

路径中包含 /

假设我们有这样一个路由：/files/{file_path}，而用户传递的 file_path 中显然是可以带 / 的，假设 file_path 是 /root/test.py，那么路由就变成了 /files//root/test.py，显然这是有问题的。

那么为了防止解析出错，我们需要做一个类似于 Flask 中的操作：

from fastapi import FastAPI
import uvicorn

app = 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", host="0.0.0.0", port=5555)

然后我们来访问一下：

FastAPI：Python 世界里最受欢迎的异步框架

查询参数

查询参数在 FastAPI 中依旧可以通过类型注解的方式进行声明，如果函数中定义了不属于路径参数的参数时，那么它们将会被自动解释会查询参数。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = 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", host="0.0.0.0", port=5555)

注意：name 和 age 是没有默认值的，这意味着它们是必须要传递的，否则报错。

FastAPI：Python 世界里最受欢迎的异步框架

我们看到当不传递 name 和 age 的时候，会直接提示你相关的错误信息。如果我们希望用户可以不传递的话，那么必须要指定一个默认值。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str = "UNKNOWN", age: int = 0):
    return {"user_id": user_id, "name": name, "age": age}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

对于查询参数，由于它们指定了类型，所以我们也要传递正确类型的数据。假设这里的 age 传递了一个 "abc"，那么也是通不过的。

如果默认值和类型不相同怎么办？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str = "UNKNOWN", age: int = "哈哈哈"):
    return {"user_id": user_id, "name": name, "age": age}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

这里的 age 需要接收一个整型，但是默认值却是一个字符串，那么此时会有什么情况发生呢？我们来试一下：

FastAPI：Python 世界里最受欢迎的异步框架

我们看到，传递的 age 依旧需要整型，只不过在不传的时候会使用字符串类型的默认值，所以类型和默认值类型不同的时候也是可以的，只不过这么做显然是不合理的。但是问题来了，我们可不可以指定多个类型呢？比如 user_id 按照整型解析、解析不成功退化为字符串。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Union, Optional
from fastapi import FastAPI
import uvicorn

app = 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

所以 FastAPI 的设计还是非常不错的，通过 Python 的类型注解来实现参数类型的限定可以说是非常巧妙的，因此这也需要我们熟练掌握 Python 的类型注解。

bool 类型自动转换

对于布尔类型，FastAPI 支持自动转换，举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/{flag}")
async def get_flag(flag: bool):
    return {"flag": flag}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

多个路径和查询参数

我们之前说过，可以定义任意个路径参数，只要动态的路径参数{} 里面的 在函数的参数中都出现即可。当然查询参数也可以是任意个，FastAPI 可以处理的很好。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
from fastapi import FastAPI
import uvicorn

app = 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 = f"select {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", host="0.0.0.0", port=5555)

然后我们使用 requests 来测试一下：

print(
    requests.get("http://localhost:5555/postgres/ods/v1/staff").json()
)  # {'query': 'select * from ods.staff'}
print(
    requests.get("http://localhost:5555/postgres/ods/v1/staff?select=id, name&where=id > 3&limit=100").json()
)  # {'query': 'select id, name from ods.staff where id > 3 limit 100'}

Depends

这个老铁比较特殊，它是用来做什么的呢？我们来看一下：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
import uvicorn
from fastapi import Depends, FastAPI

app = 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 commons


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


if __name__ == "__main__":
    uvicorn.run("main1:app", host="0.0.0.0", port=5555)

我们来测试一下：

FastAPI：Python 世界里最受欢迎的异步框架

所以 Depends 能够很好的实现依赖注入，而且我们特意写了两个路由，就是想表示它们是彼此独立的。因此当有共享的逻辑、或者共享的数据库连接、增强安全性、身份验证、角色权限等等，会非常的使用。

查询参数和数据校验

FastAPI 支持我们进行更加智能的数据校验，比如一个字符串，我们希望用户在传递的时候只能传递长度为 6 到 15 的字符串该怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/user")
async def check_length(
        # 默认值为 None，应该声明为 Optional[str]，当然声明 str 也是可以的。只不过声明为 str，那么默认值应该也是 str
        # 所以如果一个类型允许为空，那么更规范的做法应该是声明为 Optional[类型]。
        password: Optional[str] = Query(None, min_length=6, max_length=15)
):
    return {"password": password}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

password 是可选的，但是一旦传递则必须传递字符串、而且还是长度在 6 到 15 之间的字符串。所以如果传递的是 None，那么在声明默认值的时候 None 和 Query(None) 是等价的，只不过 Query 还支持其它的参数来对参数进行限制。

FastAPI：Python 世界里最受欢迎的异步框架

Query 里面除了限制最小长度和最大长度，还有其它的功能：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/user")
async def check_length(
        password: str = Query("satori", min_length=6, max_length=15, regex=r"^satori")
):
    """此时的 password 默认值为 'satori'，并且传递的时候必须要以 'satori' 开头
       但是值得注意的是 password 后面的是 str，不再是 Optional[str]，因为默认值不是 None 了
       当然这里即使写成 Optional[str] 也是没有什么影响的
    """
    return {"password": password}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

声明为必须参数

我们通过 Query 可以限制参数的长度，但是问题来了，这个时候我还希望这个参数是必传的该怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/user")
async def check_length(
        password: str = Query(..., min_length=6)
):
    """将第一个参数换成 ... 即可实现该参数是必传参数
    """
    return {"password": password}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

... 是 Python 中的一个特殊的对象，可以了解一下，通过它可以实现该参数是必传参数。

查询参数变成一个列表

如果我们指定了 a=1&a=2，那么我们在获取 a 的时候如何才能得到一个列表呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = 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", host="0.0.0.0", port=5555)

我们访问一下，看看结果：

FastAPI：Python 世界里最受欢迎的异步框架

首先 "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（或者有默认值）的话，那么应该这么写：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

给参数起别名

问题来了，假设我们定义的查询参数名叫 item-query，那么由于它要体现在函数参数中、而这显然不符合 Python 变量的命名规范，这个时候要怎么做呢？答案是起一个别名。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/items")
async def read_items(
        # 通过 url 的时候使用别名即可
        item1: Optional[str] = Query(None, alias="item-query"),
        item2: str = Query("哈哈", alias="@@@@"),
        item3: str = Query(..., alias="$$$$")  # item3 是必传的
):
    return {"item1": item1, "item2": item2, "item3": item3}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

数值检测

Query 不仅仅支持对字符串的校验，还支持对数值的校验，里面可以传递 gt、ge、lt、le 这几个参数，相信这几个参数不用说你也知道是干什么的，我们举例说明：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/items")
async def read_items(
        # item1 必须大于 5
        item1: int = Query(..., gt=5),
        # item2 必须小于等于 7
        item2: int = Query(..., le=7),
        # item3 必须必须等于 10
        item3: int = Query(..., ge=10, le=10)
):
    return {"item1": item1, "item2": item2, "item3": item3}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

Query 还是比较强大了，当然内部还有一些其它的参数是针对 docs 交互文档的，有兴趣可以自己了解一下。

路径参数和数据校验

查询参数数据校验使用的是 Query，路径参数数据校验使用的是 Path，两者的使用方式一模一样，没有任何区别。

from fastapi import FastAPI, Path
import uvicorn

app = FastAPI()

@app.get("/items/{item-id}")
async def read_items(item_id: int = Path(..., alias="item-id")):
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

因为路径参数是必须的，它是路径的一部分，所以我们应该使用 ... 将其标记为必传参数。当然即使不这么做也无所谓，因为指定了默认值也用不上，因为路径参数不指定压根就匹配不到相应的路由。至于一些其它的校验，和查询参数一模一样，所以这里不再赘述了。

不过我们之前说过，路径参数应该在查询参数的前面，尽管 FastAPI 没有这个要求，但是这样写明显更舒服一些。但是问题来了，如果路径参数需要指定别名，但是某一个查询参数不需要，这个时候就会出现问题：

@app.get("/items/{item-id}")
async def read_items(q: str,
                     item_id: int = Path(..., alias="item-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(..., alias="item-id"),
                     q: str):
    
    return {"item_id": item_id, "q": q}

此时就没有问题了，通过将第一个参数设置为 *，使得 item_id 和 q 都必须通过关键字传递，所以此时默认参数在非默认参数之前也是允许的。当然我们也不需要担心 FastAPI 传参的问题，你可以认为它所有的参数都是通过关键字参数的方式传递的。

Request

Request 是什么？首先我们知道任何一个请求都对应一个 Request 对象，请求的所有信息都在这个 Request 对象中，FastAPI 也不例外。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request
import uvicorn

app = 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  # cookies
    query_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", host="0.0.0.0", port=5555)

我们通过 Request 对象可以获取所有请求相关的信息，我们之前当参数传递不对的时候，FastAPI 会自动帮我们返回错误信息，但通过 Request 我们就可以自己进行解析、自己指定返回的错误信息了。

FastAPI：Python 世界里最受欢迎的异步框架

Response

既然有 Request，那么必然会有 Response，尽管我们可以直接返回一个字典，但 FastAPI 实际上会帮我们转成一个 Response 对象。

Response 内部接收如下参数：

  • content：返回的数据
  • status_code：状态码
  • headers：返回的请求头
  • media_type：响应类型(就是 HTML 中 Content-Type，只不过这里换了个名字)
  • background：接收一个任务，Response 在返回之后会自动异步执行（这里不做介绍，后面会说）

举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request, Response
import uvicorn
import orjson

app = 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-Type
        "application/json",  
    )
    # 如果想设置 cookie 的话，那么通过 response.set_cookie 即可
    # 删除 cookie 则是 response.delete_cookie
    return response


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

通过 Response 我们可以实现请求头、状态码、cookie 等自定义。

另外除了 Response 之外还有很多其它类型的响应，它们都在 fastapi.responses 中，比如：FileResponse、HTMLResponse、PlainTextResponse 等等。它们都继承了 Response，只不过会自动帮你设置响应类型，举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import Response, HTMLResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():

    response1 = HTMLResponse("<h1>你好呀</h1>")
    response2 = Response("<h1>你好呀</h1>", media_type="text/html")
    # 以上两者是等价的，在 HTMLResponse 中会自动将 media_type 设置成 text/html
    return response1


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

其它类型的请求与响应

FastAPI 除了 GET 请求之外，还支持其它类型，比如：POST、PUT、DELETE、OPTIONS、HEAD、PATCH、TRACE 等等。而常见的也就 GET、POST、PUT、DELETE，介绍完了 GET，我们来说一说其它类型的请求。

显然对应 POST、PUT 等类型的请求，我们必须要能够解析出请求体，并且能够构造出响应体。

Model

在 FastAPI 中，请求体和响应体都对应一个 Model，举个栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    """数据验证是通过 pydantic 实现的，我们需要从中导入 BaseModel，然后继承它"""
    name: str
    age: Optional[str] = None
    length: float
    hobby: List[str]  # 对于 Model 中的 List[str] 我们不需要指定 Query（准确的说是 Field）


@app.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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

除了使用这种方式之外，我们还可以使用之前说的 Request 对象：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

@app.post("/girl")
async def read_girl(request: Request):
    # 是一个协程，所以需要 await
    data = await request.body()
    print(data)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先我们在使用 requests 模块发送 post 请求的时候可以通过 data 参数传递、也可以通过 json 参数。

当通过 json={"name": "satori", "age": 16, "length": 155.5} 传递的时候，会将其转成 json 字符串进行传输，程序中的 print 打印如下：

b'{"name": "satori", "age": 16, "length": 155.5}'

如果我们是用过 data 参数发请求的话（值不变），那么会将其拼接成 k1=v1&k2=v2 的形式再进行传输（相当于表单提交，后面说），程序中打印如下：

b'name=satori&age=16&length=155.5'

所以我们看到 await request.body() 得到的就是最原始的字节流，而除了 await request.body() 之外还有一个 await request.json()，只是后者在内部在调用了前者拿到字节流之后、自动帮你 loads 成了字典。因此使用 await request.json() 也侧面要求，我们必须在发送请求的时候必须使用 json 参数传递（传递的是字典转成的 json、所以也能解析成字典），否则使用 await request.json() 是无法正确解析的。

路径参数、查询参数、请求体

我们可以将三者混合在一起：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    name: str
    age: Optional[str] = None
    length: float
    hobby: 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

里面我们同时指定了路径参数、查询参数和请求体，FastAPI 依然是可以正确区分的，当然我们也可以使用 Request 对象。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = 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 data


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己测试一下，仍然是可以正确返回的。

多个请求体参数

我们上面的只接收一个 json 请求体，如果是接收两个呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    name: str
    age: Optional[str] = None


class Boy(BaseModel):
    name: str
    age: int


@app.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", host="0.0.0.0", port=5555)

此时在传递的时候，应该按照如下方式传递：

FastAPI：Python 世界里最受欢迎的异步框架

应该将两个 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 uvicorn

app = FastAPI()

class BoyAndGirl(BaseModel):
    girl: Dict
    boy: Dict

@app.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", host="0.0.0.0", port=5555)

这种方式也是可以实现的，只不过就字典内部的字典的不可进行限制了。当然啦，我们仍然可以使用 Request 对象，得到字典之后自己再进行判断，因为对于 json 而言，内部的字段可能是会变的，而且最关键的是字段可能非常多。这个时候，我个人更倾向于使用 Request 对象。

Form 表单

我们调用 requests.post，如果参数通过 data 传递的话，则相当于提交了一个 form 表单，那么在 FastAPI 中可以通过 await request.form() 进行获取，注意：内部同样是先调用 await request.body()。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request, Response
import uvicorn

app = 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

当然我们也可以通过其它方式：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Form
import uvicorn

app = 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

像 Form 表单，查询参数、路径参数等等，都可以和 Request 对象一起使用，像上面的例子，如果我们多定义一个 request: Request，那么我们仍然可以通过 await request.form() 拿到相关的表单信息。所以如果你觉得某个参数不适合类型注解，那么你可以单独通过 Request 对象进行解析。

文件上传

那么问题来了，FastAPI 如何接收用户的文件上传呢？首先如果想使用文件上传功能，那么你必须要安装一个包 python-multipart，直接 pip install python-multipart 即可。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, File, UploadFile
import uvicorn

app = 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", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

我们看到一个直接获取字节流，另一个是获取类似于文件句柄的对象。如果是多个文件上传要怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import List
from fastapi import FastAPI, UploadFile, File
import uvicorn

app = 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 files

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

此时我们就实现了 FastAPI 文件上传，当然文件上传并不影响我们处理表单，可以自己试一下同时处理文件和表单。

返回静态资源

下面来看看 FastAPI 如何返回静态资源，首先我们需要安装 aiofiles，直接 pip 安装即可。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicorn

app = FastAPI()

# name 参数只是起一个名字，FastAPI 内部使用
app.mount("/static", StaticFiles(directory=r"C:\Users\satori\Desktop\bg"), name="static")

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

浏览器输入：localhost:5555/static/1.png，那么会返回 C:\Users\satori\Desktop\bg 下的 1.png 文件。

错误处理

错误处理也是一个不可忽视的点，错误有很多种，比如：

  • 客户端没有足够的权限执行此操作
  • 客户端没有访问某个资源的权限
  • 客户端尝试访问一个不存在的资源
  • ...

这个时候我们应该将错误通知相应的客户端，这个客户端可以浏览器、代码程序、IoT 设备等等。

但是就我个人而言，更倾向于使用 Response 对象，将里面的  status_code 设置为 404，然后在返回的 json 中指定错误信息。不过 FastAPI 内部也提供了一些异常类：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id != "foo":
        # 里面还可以传入 headers 设置响应头
        raise HTTPException(status_code=404, detail="item 没有发现")
    return {"item": "bar"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

HTTPException 是一个普通的 Python 异常类（继承了 Exception），它携带了 API 的相关信息，既然是异常，那么我们不能 return、而是要 raise。

个人觉得这个不是很常用，至少我本人很少用这种方式返回错误，因为它能够携带的信息太少了。

自定义异常

FastAPI 内部提供了一个 HTTPException，但是我们也可以自定义，但是注意：我们自定义完异常之后，还要定义一个 handler，将异常和 handler 绑定在一起，然后引发该异常的时候就会触发相应的 handler。

from fastapi import FastAPI, Request
from fastapi.responses import ORJSONResponse
import uvicorn

app = FastAPI()


class ASCIIException(Exception):
    """何もしません"""

# 通过装饰器的方式，将 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_code=404, content={"code": 404, "message": "你必须传递 ascii 字符串"})

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if not item_id.isascii():
        raise ASCIIException
    return {"item": f"get {item_id}"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

关于 Request、Response，我们除了可以通过 fastapi 进行导入，还可以通过 starlette 进行导入，因为 fastapi 的路由映射是通过 starlette 来实现的。当然我们直接从 fastapi 中进行导入即可。

自定义 404

当访问一个不存在的 URL，我们应该提示用户，比如：您要找到页面去火星了。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.exceptions import StarletteHTTPException
import uvicorn

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def not_found(request, exc):
    return ORJSONResponse({"code": 404, "message": "您要找的页面去火星了。。。"})

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

此时当我们访问一个不存在的 URL 时，就会返回我们自定义的 JSON 字符串。

后台任务

如果一个请求耗时特别久，那么我们可以将其放在后台执行，而 FastAPI 已经帮我们做好了这一步。我们来看一下：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
import time
from fastapi import FastAPI, BackgroundTasks
import uvicorn

app = 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": "邮件发送成功"}), background=BackgroundTask(send_email, email, message="这是一封邮件"))
    return {"message": "邮件发送成功"}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先请求肯定是成功的：

FastAPI：Python 世界里最受欢迎的异步框架

然后响应的 3 秒，终端会出现如下打印：

FastAPI：Python 世界里最受欢迎的异步框架

所以此时任务是被后台执行了的，注意：任务是在响应返回之后才后台执行。

APIRouter

APIRouter 类似于 Flask 中的蓝图，可以更好的组织大型项目，举个栗子：

FastAPI：Python 世界里最受欢迎的异步框架

在我当前的工程目录中有一个 app 目录和一个 main.py，其中 app 目录中有一个 app01.py，然后我们看看它们是如何组织的。

# app/app01.py
from fastapi import APIRouter

router = APIRouter(prefix="/router")

# 以后访问的时候要通过 /router/v1 来访问
@router.get("/v1")
async def v1():
    return {"message": "hello world"}


# main.py
from fastapi import FastAPI
from app.app01 import router
import uvicorn

app = FastAPI()

# 将 router 注册到 app 中，相当于 Flask 中的 register_blueprint
app.include_router(router)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后可以在外界通过 /router/v1 的方式来访问。

CORS

CORS（跨域资源共享）是指浏览器中运行的前端拥有和后端通信的 JavaScript 代码，而前端和后端处于不同源的情况。源：协议（http、https）、域（baidu.com、app.com、localhost）以及端口（80、443、8000），只要有一个不同，那么就是不同源。比如下面都是不同的源：

  • http://localhost
  • https://localhost
  • http://localhost:8080

即使它们都是 localhost，但是它们使用了不同的协议或端口，所以它们是不同的源。假设你的前端运行在 localhost:8080，并且尝试与 localhost:5555 进行通信；然后浏览器会向后端发送一个 HTTP OPTIONS 请求，后端会发送适当的 headers 来对这个源进行授权；所以后端必须有一个 "允许的源" 列表，如果前端对应的源是被允许的，浏览器才会允许前端向后端发请求，否则就会出现跨域失败。

而默认情况下，前后端必须是在同一个源，如果不同源那么前端就会请求失败。而前后端分离早已成为了主流，因此跨域问题是必须要解决的。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    # 允许跨域的源列表，例如 ["http://www.example.org"] 等等，["*"] 表示允许任何源
    allow_origins=["*"],
    # 跨域请求是否支持 cookie，默认是 False，如果为 True，allow_origins 必须为具体的源，不可以是 ["*"]
    allow_credentials=False,
    # 允许跨域请求的 HTTP 方法列表，默认是 ["GET"]
    allow_methods=["*"],
    # 允许跨域请求的 HTTP 请求头列表，默认是 []，可以使用 ["*"] 表示允许所有的请求头
    # 当然 Accept、Accept-Language、Content-Language 以及 Content-Type 总之被允许的
    allow_headers=["*"],
    # 可以被浏览器访问的响应头, 默认是 []，一般很少指定
    # expose_headers=["*"]
    # 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600，一般也很少指定
    # max_age=1000
)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

CORS 算是中间件的一种，我们后面也会详细介绍中间件，只不过 CORS 太重要了，所以需要单独拿出来说一下。

高阶操作

下面我们看一些 FastAPI 的高阶操作，这些操作有的不一定能用上，但用上了确实会方便许多。

其它的响应

返回 json 数据可以是：JSONResponse、UJSONResponse、ORJSONResponse，Content-Type 是 application/json；返回 html 是 HTMLResponse，Content-Type 是 text/html；返回 PlainTextResponse，Content-Type 是 text/plain。但是我们还可以有三种响应，分别是返回重定向、字节流、文件。

重定向

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():
    return RedirectResponse("https://www.bilibili.com")


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

页面中访问 /index 会跳转到 bilibili。

字节流

返回字节流需要使用异步生成器的方式：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicorn

app = FastAPI()

async def some_video():
    for i in range(5):
        yield f"video {i} bytes ".encode("utf-8")

@app.get("/index")
async def index():
    return StreamingResponse(some_video())


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

如果有文件对象，那么也是可以直接返回的。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():
    return StreamingResponse(open("main.py", encoding="utf-8"))


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

FastAPI：Python 世界里最受欢迎的异步框架

文件

返回文件的话，还可以通过 FileResponse：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import FileResponse
import uvicorn

app = FastAPI()

@app.get("/index")
async def index():
    # filename 如果给出，它将包含在响应的 Content-Disposition 中。
    return FileResponse("main.py", filename="这不是main.py")

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己发请求测试一下。

HTTP 验证

如果当用户访问某个请求的时候，我们希望其输入用户名和密码来确认身份的话该怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import uvicorn

app = 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", host="0.0.0.0", port=5555)

当用户访问 /index 的时候，会提示输入用户名和密码：

FastAPI：Python 世界里最受欢迎的异步框架

输入完毕之后，信息会保存在 credentials，我们可以获取出来进行验证。

websocket

然后我们来看看 FastAPI 如何实现 websocket：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.websockets import WebSocket
import uvicorn

app = 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", host="0.0.0.0", port=5555)

然后我们通过浏览器进行通信：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <script>
        ws = new WebSocket("ws://localhost:5555/ws");
 
        //如果连接成功, 会打印下面这句话, 否则不会打印
        ws.onopen = function () {
            console.log('连接成功')
        };
 
        //接收数据, 服务端有数据过来, 会执行
        ws.onmessage = function (event) {
            console.log(event)
        };
 
        //服务端主动断开连接, 会执行.
        //客户端主动断开的话, 不执行
        ws.onclose = function () {  }
 
    </script>
</body>
</html>

FastAPI：Python 世界里最受欢迎的异步框架

FastAPI 服务的部署

目前的话，算是介绍了 FastAPI 的绝大部分内容，然后我们来看看 FastAPI 服务的部署，其实部署很简单，直接 uvicorn.run 即可。但是这里面有很多的参数，我们主要是想要介绍这些参数。

def run(app, **kwargs):
    config = Config(app, **kwargs)
    server = Server(config=config)
    ...
    ...

我们看到 app 和 **kwargs 都传递给了 Config，所以我们只需要看 Config 里面都有哪些参数即可。这里选出一部分：

  • app：第一个参数，不需要解释
  • host：监听的ip
  • port：监听的端口
  • uds：绑定的 unix domain socket，一般不用
  • fd：从指定的文件描述符中绑定 socket
  • loop：事件循环实现，可选项为 auto|asyncio|uvloop|iocp
  • http：HTTP 协议实现，可选项为 auto|h11|httptools
  • ws：websocket 协议实现，可选项为 auto|none|websockets|wsproto
  • lifespan：lifespan 实现，可选项为 auto|on|off
  • env_file：环境变量配置文件
  • log_config：日志配置文件
  • log_level：日志等级
  • access_log：是否记录日志
  • use_colors：是否带颜色输出日志信息
  • interface：应用接口，可选 auto|asgi3|asgi2|wsgi
  • debug：是否开启 debug 模式
  • reload：是否自动重启
  • reload_dirs：要自动重启的目录
  • reload_delay：多少秒后自动重启
  • workers：工作进程数
  • limit_concurrency：并发的最大数量
  • limit_max_requests：能 hold 住的最大请求数

小结

总的来说，FastAPI 算是当前最流行的异步框架了，并且它完全可以在生产中使用，是值得信赖的。当然使用异步框架，最重要的是要搭配一个异步驱动去访问数据库，因为 web 服务的瓶颈都是在数据库上面。

上一篇：uvicorn 使用方法


下一篇：ubuntu 20.10下uvicorn+nginx部署django