FastAPI
本文最后更新于132 天前,其中的信息可能已经过时,如有错误请发送邮件到zhizihua030611@163.com

声明类型

在创建函数以及所输入的变量后,需要声明变量类型

简单类型

def get_name_with_age(name: str, age: int):
   name_with_age = name + " is this old: " + age
   return name_with_age

嵌套类型

有些容器数据结构可以包含其他的值,比如 dictlistsettuple。它们内部的值也会拥有自己的类型。

你可以使用 Python 的 typing 标准库来声明这些类型以及子类型。

它专门用来支持这些类型提示。

列表

from typing import List


def process_items(items: List[str]):
   for item in items:
       print(item)

元组和集合

from typing import Set, Tuple


def process_items(items_t: Tuple[int, int, str], items_s: Set[bytes]):
   return items_t, items_s

字典

from typing import Dict


def process_items(prices: Dict[str, float]):
   for item_name, item_price in prices.items():
       print(item_name)
       print(item_price)

类作为类型

可以将类生命为对象的类型

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


def get_person_name(one_person: Person):
   return one_person.name

Pydantic 模型

Pydantic是一个用来执行数据校验的库。

你可以将数据的“结构”声明为具有属性的类。

每个属性都有类型。

接着你用一些值来创建这个类的实例,这些值会被校验,并被转换成适当的类型(在需要的情况下),返回一个包含所有数据的对象。

from datetime import datetime

from pydantic import BaseModel


class User(BaseModel):
   id: int
   name: str = "John Doe"
   signup_ts: datetime | None = None
   friends: list[int] = []


external_data = {
   "id": "123",
   "signup_ts": "2017-06-01 12:22",
   "friends": [1, "2", b"3"],
}
user = User(**external_data)
print(user)
# > User id=123 name='John Doe' signup_ts=datetime.datetime(2017, 6, 1, 12, 22) friends=[1, 2, 3]
print(user.id)
# > 123

可以这么说,整个Fastapi建立在Pydantic的基础上

FastAPI 利用这些类型提示来做下面几件事。

使用 FastAPI 时用类型提示声明参数可以获得:

  • 编辑器支持
  • 类型检查

.并且 FastAPI 还会用这些类型声明来:

  • 定义参数要求:声明对请求路径参数、查询参数、请求头、请求体、依赖等的要求。
  • 转换数据:将来自请求的数据转换为需要的类型。
  • 校验数据 : 对于每一个请求:当数据校验失败时自动生成错误信息返回给客户端。
  • 使用 OpenAPI记录API:然后用于自动生成交互式文档的用户界面。

并发 async/await

使用

如果你正在使用第三方库,它们会告诉你使用 await 关键字来调用它们,就像这样:

results = await some_library()

然后,通过 async def 声明你的 路径操作函数

@app.get('/')
async def read_results():
   results = await some_library()
   return results

你只能在被 async def 创建的函数内使用 await

不使用

如果你正在使用一个第三方库和某些组件(比如:数据库、API、文件系统…)进行通信,第三方库又不支持使用 await (目前大多数数据库三方库都是这样),这种情况你可以像平常那样使用 def 声明一个路径操作函数,就像这样:

@app.get('/')
def results():
results = some_library()
return results

如果你的应用程序不需要与其他任何东西通信而等待其响应,请使用 async def,如果不清楚,用 def 就好

异步代码

和node中的异步机制一样。

使用 FastAPI,你可以利用 Web 开发中常见的并发机制的优势(NodeJS 的主要吸引力)。

并且,你也可以利用并行和多进程(让多个进程并行运行)的优点来处理与机器学习系统中类似的 CPU 密集型 工作。

CPU 密集型操作的常见示例是需要复杂的数学处理。

例如:

  • 音频图像处理;
  • 计算机视觉: 一幅图像由数百万像素组成,每个像素有3种颜色值,处理通常需要同时对这些像素进行计算;
  • 机器学习: 它通常需要大量的”矩阵”和”向量”乘法。想象一个包含数字的巨大电子表格,并同时将所有数字相乘;
  • 深度学习: 这是机器学习的一个子领域,同样适用。只是没有一个数字的电子表格可以相乘,而是一个庞大的数字集合,在很多情况下,你需要使用一个特殊的处理器来构建和使用这些模型。

如果你的函数是通过 def 声明的,它将被直接调用(在代码中编写的地方),而不会在线程池中,如果这个函数通过 async def 声明,当在代码中调用时,你就应该使用 await 等待函数的结果。

教程

第一步

最简单的FastAPI文件

from fastapi import FastAPI

app = FastAPI()


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

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

交互式 API 文档

跳转到 http://127.0.0.1:8000/docs你将会看到自动生成的交互式 API 文档(由Swagger UI提供)

前往 http://127.0.0.1:8000/redoc

可选的 API 文档

你将会看到可选的自动生成文档 (由 ReDoc 提供)

分步概括

导入FastAPI
创建一个FastAPI实例
创建一个路径操作

路径:指的是url中第一个/起的后半部分

操作:(get,set,post等之类的通信方法)

定义一个路径操作装饰器

定义路径操作函数

这是我们的「路径操作函数」:

  • 路径:是 /
  • 操作:是 get
  • 函数:是位于「装饰器」下方的函数(位于 @app.get("/") 下方)。

路径参数

声明路径参数的类型

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

数据转换

运行上面的示例代码,访问 http://127.0.0.1:8000/items/3,返回结果如下、

{"item_id":3}

数据校验

如果访问http://127.0.0.1:8000/items/foo,返回结果如下

{
   "detail": [
      {
           "loc": [
               "path",
               "item_id"
          ],
           "msg": "value is not a valid integer",
           "type": "type_error.integer"
      }
  ]
}

这是因为路径参数item_id 的值是str类型,不是所规定的int类型

规定路由的顺序也很重要

比如要使用 /users/me 获取当前用户的数据。

然后还要使用 /users/{user_id},通过用户 ID 获取指定用户的数据。

由于路径操作是按顺序依次运行的,因此,一定要在 /users/{user_id} 之前声明 /users/me

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


@app.get("/users/{user_id}")
async def read_user(user_id: str):
   return {"user_id": user_id}

否则,/users/{user_id} 将匹配 /users/me,FastAPI 会认为正在接收值为 "me"user_id 参数

预设值

路径操作使用 Python 的 Enum 类型接收预设的路径参数

创建 Enum 类(枚举类)

导入 Enum 并创建继承自 strEnum 的子类。

通过从 str 继承,API 文档就能把值的类型定义为字符串,并且能正确渲染。

然后,创建包含固定值的类属性,这些固定值是可用的有效值:

from enum import Enum

from fastapi import FastAPI


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


app = FastAPI()


@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
   if model_name is ModelName.alexnet:
       return {"model_name": model_name, "message": "Deep Learning FTW!"}

   if model_name.value == "lenet":
       return {"model_name": model_name, "message": "LeCNN all the images"}

   return {"model_name": model_name, "message": "Have some residuals"}

上面的这个案例,同时还使用Enum类创建使用类型注解的路径参数

查询参数

声明的参数不是路径参数时,路径操作函数会把该参数自动解释为查询参数

from fastapi import FastAPI

app = FastAPI()

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


@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
   return fake_items_db[skip : skip + limit]

查询字符串是键值对的集合,这些键值对位于URL的?之后,以&分隔

上述例子,打开默认页面是http://127.0.0.1:8000/items/?skip=0&limit=10

查询参数类型转换

参数可以声明bool类型,FastAPI会自动转换参数类型

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None, short: bool = False):
   item = {"item_id": item_id}
   if q:
       item.update({"q": q})
   if not short:
       item.update(
          {"description": "This is an amazing item that has a long description"}
      )
   return item

访问:

http://127.0.0.1:8000/items/foo?short=1
http://127.0.0.1:8000/items/foo?short=True
http://127.0.0.1:8000/items/foo?short=true
http://127.0.0.1:8000/items/foo?short=on
http://127.0.0.1:8000/items/foo?short=yes

或其它任意大小写形式(大写、首字母大写等),函数接收的 short 参数都是布尔值 True。值为 False 时也一样。

多个路径和查询参数

FastAPI 可以识别同时声明的多个路径参数和查询参数。

而且声明查询参数的顺序并不重要。

FastAPI 通过参数名进行检测

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 = None, short: bool = False
):
   item = {"item_id": item_id, "owner_id": user_id}
   if q:
       item.update({"q": q})
   if not short:
       item.update(
          {"description": "This is an amazing item that has a long description"}
      )
   return item

请求体

FastAPI 使用请求体从客户端(例如浏览器)向 API 发送数据。

请求体是客户端发送给 API 的数据。响应体是 API 发送给客户端的数据。

API 基本上肯定要发送响应体,但是客户端不一定发送请求体

使用Pydantic模型声明请求体。

创建数据模型

把数据模型声明为继承BaseModel的类

再去声明所有属性

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

下面就是上述声明的例子

{
   "name": "Foo",
   "description": "An optional description",
   "price": 45.2,
   "tax": 3.5
}

使用模型

在路径操作函数内部直接访问模型对象的属性

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item):
   item_dict = item.dict()
   if item.tax is not None:
       price_with_tax = item.price + item.tax
       item_dict.update({"price_with_tax": price_with_tax})
   return item_dict

请求体+路径参数

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

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

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
   result = {"item_id": item_id, **item.dict()}
   if q:
       result.update({"q": q})
   return result

三种参数的分别

函数参数按如下规则进行识别:

  • 路径中声明了相同参数的参数,是路径参数
  • 类型是(intfloatstrbool 等)单类型的参数,是查询参数
  • 类型是 Pydantic 模型的参数,是请求体

查询参数和字符串校验

可选参数

Query库

要求:查询参数q是可选的,且只要提供了整个参数,则改参数值不能超过50个字符长度

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):
   results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
   if q:
       results.update({"q": q})
   return results

可以添加更多的校验规则

q: Union[str, None] = Query(
       default=None, min_length=3, max_length=50, pattern="^fixedquery$"
  )

pattern是表示可以添加正则表达式

默认值

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: str = Query(default="fixedquery", min_length=3)):
   results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
   if q:
       results.update({"q": q})
   return results

必须参数

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: str = Query(min_length=3)):
   results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
   if q:
       results.update({"q": q})
   return results

查询参数列表/多个值

当你使用 Query 显式地定义查询参数时,你还可以声明它去接收一组值,或换句话来说,接收多个值。

例如,要声明一个可在 URL 中出现多次的查询参数 q,你可以这样写:

from typing import List, Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):
   query_items = {"q": q}
   return query_items

然后,输入如下网址:

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

你会在路径操作函数函数参数 q 中以一个 Python list 的形式接收到查询参数 q 的多个值(foobar)。

因此,该 URL 的响应将会是:

{
 "q": [
   "foo",
   "bar"
]
}

tips:要声明类型为 list 的查询参数,如上例所示,你需要显式地使用 Query,否则该参数将被解释为请求体

路径参数和数值校验

使用Path

声明元数据

你可以声明与Query相同的所有参数

例如,要声明路径参数item_id 的title元数据值,

from typing import Annotated

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
   item_id: Annotated[int, Path(title="The ID of the item to get")],
   q: Annotated[str | None, Query(alias="item-query")] = None,
):
   results = {"item_id": item_id}
   if q:
       results.update({"q": q})
   return results

tip:路径参数总是必需的,因为它必须是路径的一部分。

所以,你应该在声明时使用 ... 将其标记为必需参数。

然而,即使你使用 None 声明路径参数或设置一个其他默认值也不会有任何影响,它依然会是必需参数。

按需对参数排序

文末附加内容
暂无评论

发送评论 编辑评论


				
|´・ω・)ノ
ヾ(≧∇≦*)ゝ
(☆ω☆)
(╯‵□′)╯︵┴─┴
 ̄﹃ ̄
(/ω\)
∠( ᐛ 」∠)_
(๑•̀ㅁ•́ฅ)
→_→
୧(๑•̀⌄•́๑)૭
٩(ˊᗜˋ*)و
(ノ°ο°)ノ
(´இ皿இ`)
⌇●﹏●⌇
(ฅ´ω`ฅ)
(╯°A°)╯︵○○○
φ( ̄∇ ̄o)
ヾ(´・ ・`。)ノ"
( ง ᵒ̌皿ᵒ̌)ง⁼³₌₃
(ó﹏ò。)
Σ(っ °Д °;)っ
( ,,´・ω・)ノ"(´っω・`。)
╮(╯▽╰)╭
o(*////▽////*)q
>﹏<
( ๑´•ω•) "(ㆆᴗㆆ)
😂
😀
😅
😊
🙂
🙃
😌
😍
😘
😜
😝
😏
😒
🙄
😳
😡
😔
😫
😱
😭
💩
👻
🙌
🖕
👍
👫
👬
👭
🌚
🌝
🙈
💊
😶
🙏
🍦
🍉
😣
Source: github.com/k4yt3x/flowerhd
颜文字
Emoji
小恐龙
花!
上一篇