本文主要针对使用FastAPI构建App可能用到的知识点做一个归纳总结,让我们从一个简单的Hello world开始吧。
Hello world
# 步骤1:导入FastAPI
from fastapi import FastAPI
#步骤2:创建一个FastAPI“实例”
app = FastAPI()
#步骤3:创建路径操作
@app.get("/")
async def root():
#步骤4:定义路径操作功能
return {"message": "Hello World"}
if __name__ == '__main__':
#步骤5:运行开发服务器
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
直接运行上面的代码或者命令行窗口运行uvicorn main:app --reload
:
main
:文件main.py(Python“模块”)。app
:main.py在线内创建的对象app = FastAPI()。--reload
:更改代码后使服务器重新启动。仅用于开发。
这样在浏览器输入http://127.0.0.1:8000/可以看到返回{"message":"Hello World"}
下面让我们一点一点来扩展
您可以在FastAPI应用程序app中配置几件事:title、description、version
from fastapi import FastAPI
app = FastAPI(
title="My First Project",
description="Let's have fun with it.",
version="0.0.1",)
@app.get("/")
async def root():
return {"message": "Hello World"}
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)!
FastAPI提供的自动API文档(http://127.0.0.1:8000/docs)会根据你的配置进行变更:
[一]路径参数
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: int):
return {"user_id": user_id}
注意路径操作是按顺序评估,因此如果你输入http://127.0.0.1:8000/users/me
虽然以上两个路径参数都符合,但会按排在前面的进行执行。
我们看看async def read_user(user_id: int)
这里的int
是用来声明的,它有两个作用,数据转换和数据验证。
如果你输入http://127.0.0.1:8000/users/3
请注意,函数返回的值不是string "3", 而是int 3
如果你输入http://127.0.0.1:8000/users/3.3,则会收到一个HTTP错误:
{
"detail": [
{
"loc": [
"path",
"user_id"
],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
如果你要对一个路径进行操作, 可以这样:
@app.get("/files/{file_path:path}")
async def read_user_me(file_path: str):
return {"file_path": file_path}
:path告诉参数应与任何路径匹配,否则路径一般带有一个斜杠(/),可能不会被识别。例如/home/johndoe/myfile.txt, 注意在files和你的路径参数之间应该使用双斜杠(//),URL应该:http://127.0.0.1:8000/files//home/johndoe/myfile.txt
路径“操作”其实是指HTTP“方法”之一,它是一个装饰器。
常用的有
- POST: 创建数据
- GET: 读取数据
- PUT:更新数据
- DELETE:删除数据
还有奇特一点的
- OPTIONS
- HEAD
- PATCH
- TRACE
之前讲过数据验证是类型验证,进一步的,甚至可以进行数值验证。
Path
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", gt=0, le=1000),
q: str,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results
async 是异步关键字,异步的概念python很早就有,例如:
@asyncio.coroutine
def smart_fib(n):
async / await 在python3.5+被引入,这里不详述, 对应上面的:
async def smart_fib(n):
Path的第一个参数是默认值,由于路径参数始终需要一个参数,因为它必须是路径的一部分。这种情况下我们可以用...
声明它,以将其标记为必需。实际上我们写任何默认值都不起作用。
数值验证
gt :g
reatert
han
ge:g
reater than ore
qual
lt :l
esst
han
le :l
ess than ore
qual
我们可以将几个参数传递给路径操作装饰器以对其进行配置,请注意,这些参数直接传递给路径操作装饰器,而不是传递给路径操作函数。
from typing import Set
from fastapi import FastAPI, status
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
tags: Set[str] = []
@app.post("/items/", response_model=Item,
status_code=status.HTTP_201_CREATED,
tags=["items"],
summary="Create an item",
description="Create an item with all",
deprecated=False)
async def create_item(*, item: Item):
return item
status_code(响应状态码): 也可以直接传递int代码,例如404
tag(标签): 传递参数tags用list的str(通常只有一个str)
summary(摘要)
description(描述)
deprecated(弃用):True代表在API文档它将被明确标记为不推荐使用,使用删除线
[二]查询参数
声明不属于路径参数的其他功能参数时,它们将自动解释为“查询”参数。
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,short: bool = False):
return fake_items_db[skip : skip + limit]
该查询是?URL中位于关键字之后的一组键值对,以&字符分隔。
http://127.0.0.1:8000/items/?skip=0&limit=10
我们还可以声明bool类型, 1 \true\on\yes 以及它们的大小写变体将被转换为True。
http://127.0.0.1:8000/items/foo?short=yes
当声明非路径参数的默认值时(目前,我们仅看到查询参数),则不需要此值。如果您不想添加特定值,而只是将其设为可选值,则将默认值设置为None。但是,当您需要一个查询参数时,就不能声明任何默认值
例如下面这样一个例子:
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: int = None):
item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
return item
在这种情况下,有3个查询参数:
needy,是必需的str。
skip,int默认值为0。
limit,可选的int。
但是limitL int = None
会面临类型检查错误
于是使用Optional(可选的类型声明),要改写成:
from typing import Optional
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: Optional[int] = None):
item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
return item
Query
类似于路径参数使用Path来声明相同类型的验证, 可以使用Query为查询参数声明更多验证。
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
长度检查:
min_length=3, max_length=50
正则表达式检查:regex="^fixedquery$"
假设您希望参数为item-query。像:http://127.0.0.1:8000/items/?item-query=foobaritems 但是item-query不是有效的Python变量名称。
最接近的是item_query,但是您仍然需要它完全是item-query...
然后,您可以声明一个alias,该别名将用于查找参数值
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(
q: str = Query(
None,
alias="item-query",
title="Query string",
description="Query string for the items to search in the database that have a good match",
min_length=3,
max_length=50,
regex="^fixedquery$",
deprecated=True,
)
):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results
通用验证和元数据:
- alias
- title
- description
- deprecated
特定于字符串的验证:
- min_length
- max_length
- regex
我觉得FastAPI一个非常nice的优势是它基于Pydantic模型来完成request body的许多校验工作,以避免在函数内部写很多类型处理,使代码看起来更简洁。下面一起看看。
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使用该类型声明,FastAPI将:
- 以JSON读取请求的正文。
- 转换相应的类型(如果需要)。
- 验证数据。
上面的该模型声明一个JSON“ object”(或Python dict),例如:
{
"name": "Foo",
"description": "An optional description",
"price": 45.2,
"tax": 3.5
}
由于description和tax是可选的(默认值为None),此JSON“ object”也将有效:
{
"name": "Foo",
"price": 45.2
}
复杂一点的,我们可以同时声明body,path和query参数:
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
app = FastAPI()
@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: str = None):
result = {"item_id": item_id, **item.dict()}
if q:
result.update({"q": q})
return result
功能参数将被识别如下:
- 如果在path中也声明了该参数,它将用作路径参数。
- 如果参数是一个的单一类型(如int,float,str,bool,等等)将被解释为一个查询参数。
- 如果参数声明为Pydantic模型的类型,则它将被解释为请求正文。
Body
和的相同方法Query和Path为查询和路径参数定义额外的数据,FastAPI提供了等效的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(...)
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results
如果这里的importance不使用Body,由于它是一个单一类型int, 它会被当做查询参数。但这里使用Body, 那么我们期望的json是下面这样子:
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
},
"user": {
"username": "dave",
"full_name": "Dave Grohl"
},
"importance": 5
}
注意这里多个body的时候(User、Item),会使用正文中的键作为参数名称,使得User和Item的属性值再往里嵌套一层。但如果是单个Body, 我们也想要使其属性值往里面嵌套一层的话,就要使用embed
。
from fastapi import Body, FastAPI
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, item: Item = Body(..., embed=True)):
results = {"item_id": item_id, "item": item}
return results
这样,期望的body是:
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
}
}
代替:
{
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
}
如果对Item里的price 我们希望进行数据验证的话,也是有办法的,我们使用Field。
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class Item(BaseModel):
name: str
description: str = Field(None, title="The description of the item", max_length=300)
price: float = Field(..., gt=0, description="The price must be greater than zero")
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
另外,example 是配置例子,例子会在docs API显示。
from fastapi import Body, FastAPI
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,
item: Item = Body(
...,
example={
"name": "Foo",
"description": "A very nice Item",
"price": 35.4,
"tax": 3.2,
},
)
):
results = {"item_id": item_id, "item": item}
return results
我们甚至可以嵌套模型:
from typing import Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl
app = FastAPI()
class Image(BaseModel):
url: HttpUrl
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
期望的body 是这样的:
{
"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"
}
}
注意,在Image模型中,我们有一个url字段,我们可以将其声明为Pydantic的HttpUrl,而不是str, 该字符串将被检查为有效的URL,并在JSON Schema / OpenAPI中进行记录。
下是一些您可以使用的其他数据类型:
-
UUID
:- 一个标准的“通用唯一标识符”,在许多数据库和系统中通常作为ID使用。
- 在请求和响应中将以表示
str
。
-
datetime.datetime
:- 一个Python
datetime.datetime
。 - 在请求和响应中,将以
str
ISO 8601格式表示,例如:2008-09-15T15:53:00+05:00
。
- 一个Python
-
datetime.date
:- Python
datetime.date
。 - 在请求和响应中,将以
str
ISO 8601格式表示,例如:2008-09-15
。
- Python
-
datetime.time
:- 一个Python
datetime.time
。 - 在请求和响应中,将以
str
ISO 8601格式表示,例如:14:23:55.003
。
- 一个Python
-
datetime.timedelta
: -
frozenset
:- 在请求和响应中,将与视为相同
set
:- 在请求中,将读取列表,消除重复,并将其转换为
set
。 - 作为响应,
set
将会转换为list
。 - 生成的架构将指定
set
值是唯一的(使用JSON架构的uniqueItems
)。
- 在请求中,将读取列表,消除重复,并将其转换为
- 在请求和响应中,将与视为相同
-
bytes
:- 标准Python
bytes
。 - 在请求和响应中将被视为
str
。 - 生成的模式将指定,这是一个
str
与binary
“格式”。
- 标准Python
-
Decimal
:- 标准Python
Decimal
。 - 在请求和响应中,处理方式与相同
float
。
- 标准Python
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,
}
网友评论