FastAPI教程

编程基础 python  收藏
0 / 1858

前言

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于标准的 Python 类型提示。

它拥有的关键特性:

  • 快速：可与 NodeJS 和 Go 比肩的极高性能（归功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。
  • 高效编码：提高功能开发速度约 200％ 至 300％。*
  • 更少 bug：减少约 40％ 的人为（开发者）导致错误。*
  • 智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间。
  • 简单：设计的易于使用和学习，阅读文档的时间更短。
  • 简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。
  • 健壮：生产可用级别的代码。还有自动生成的交互式文档。
  • 标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

FastAPI相比于其余的Python Web框架来说，主要用于构建后端API方向，它拥有比其余的Python Web框架更好的性能。启动时间十分之快，极大提升了开发进度。并且自动拥有API文档界面，更方便的支持类型输入。

FastAPI是去年2021年Github上年度最佳新兴框架，目前在GitHub上拥有45.5K的Star，被多个国际大公司所认可，十分具有学习、使用价值。

入门

在学习使用FastAPI前，确保运行环境下安装Python3.6以上版本，因为FastAPI基于Python3.6+设计。

按正常流程是安装 FastAPI ，但是这里不建议直接在CMD中使用PIP安装。

这里 以项目形式，使用Pycharm IDE来进行学习。创建一个正常的Python项目，在项目中安装FastAPI、uvicorn，其中uvicorn是一个目前流行的异步服务器。

随后在项目 main.py文件中输入如下内容：

import uvicorn
from fastapi import FastAPI

app = FastAPI()

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

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

点击运行main。成功在127.0.0.1:8000下启动服务器：

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

上面内容就定义了一个首页的API输出，打开浏览器、后者使用第三方工具 访问http://127.0.0.1:8000页面。

会发现成功输出了{"message": "Hello World"}内容，这一切都使用了RestAPI形式。

API文档页面

前面提到了FastAPI中自动生成了API文档页面，这一切都不需要配置。

在服务器启动状态下，访问http://127.0.0.1:8000/docs 页面，你将会看到自动生成的交互式 API 文档，它使用Swagger UI形式。

除了Swagger-UI文档之外，FastAPI还内置了ReDoc形式文档，进入http://127.0.0.1:8000/redoc 即可。

详细学习

启动服务器

从上面的入门中，我们就可以发现，要启动整个FastAPI服务器的话，分3大部分：

  1. 创建FastAPI的实例：

    app = FastAPI()

  2. 创建对应接口：这一部分就是我们需要关注项目的主要部分之一了。

  3. 在Python文件入口中，开启服务器：

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

接口的书写

GET方法请求

FastAPI的接口写法和Java的Spring的Restful API写法十分相像。

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

# 动态路径的请求
@app.get("/param/{item_id}")
async def get_param(item_id: int):
    return {"itemId": item_id,"message":"This is a param content"}

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

# 参数查询的请求
@app.get("/get")
async def get1(id: int = 10):
    return {"id": id, "data": fake_items_db[:3]}

# 可选参数的设置
# 其中Union为typing 包函数
@app.get("/get")
async def get2(id: int = 10,limit:Union[int, None] = None):
    if limit:
        return {"id": id, "data": fake_items_db[:limit]}
    return {"id": id, "data": fake_items_db[:3]}

上面为FastAPI中GET方法的一些使用方法。

其中@app.get()注解，表示其为GET方法的接口，它需要接受一个路径，用作API地址。相应的POST、DELETE、PUT都是同样类似方法。

API接口可以使用async异步，表示此接口为异步步骤，与不带异步关键字的 接口的使用区别：

当你使用到第三方库时，需要调用到它们await时，就需要在接口函数def前使用async，才能实现异步并发效果。

而当您正在使用与某些东西（数据库、API、文件系统等）通信并且不支持使用的第三方库await（目前大多数数据库库都是这种情况），然后声明您的路径操作正常运行，只需def即可。

而当你的项目中，不必与其它东西进行链接通讯时，使用async def可以获得最佳性能。

FastAPI在输入参数可以对其参数类型进行 规定，并且可以设置默认值。当API 输入的参数类型不一致的话，FastAPI会抛出类型错误。

async def get1(id: int = 10):
    ...

默认情况下入参是必输内容。但利用Python3.5+类型的特性（参数类型支持2个），结合Union函数，在另一个参数类型中定义None，并且设置默认值为None，即可将其参数设置为可选参数。

async def get2(id: int = 10,limit:Union[int, None] = None):
    ...

除了查询参数的可选设置，还可以设置响应的检验：

from fastapi import FastAPI, Query

@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, max_length=50)，表示这个参数时一个可选参数并且最大长度为50。除了Query中有 max_length外，还拥有最小值检验min_length。

也可以在正则表达式检验：

q: Union[str, None] = Query(
        default=None, regex="^fixedquery$"
)

上述Query中都带有default，表明这个是可选输入参数。对于必须参数的话，不要带有default：

q: str = Query(min_length=3)
```除此之外，Query还支持下面这些内容定义：

* default：参数的默认值。
* title：参数的标题。用作在 OpenAPI 和自动 API 文档用户界面中作为 API 的标题/名称使用。
* description：参数的说明。用作在 OpenAPI 和自动 API 文档用户界面中对该参数的描述。
* gt：要求参数大于这个值，必须为数字。

#### POST方法请求

如上所描述，FastAPI的POST方法请求使用的`@app.post()`注解。同样PUT方法使用`@app.put()`注解。

```python
@app.post("/items/")
async def create_item(name: str,description: Union[str, None] = None):
    ...
    return ...

请求模型

在项目中通常是将请求内容封装为请求模型，当做请求的模型。

要将其请求参数组装为模型的话，这个请求模型要继承pydantic 的 BaseModel。

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

app = FastAPI()

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

在请求函数内部，还可以利用dict函数提取到对应dict，然后使用update可以向提交的请求体中添加内容。

注意：你不能使用 GET 操作（HTTP 方法）发送请求体。

要发送数据，你必须使用下列方法之一：POST（较常见）、PUT、DELETE 或 PATCH。

对于同时请求链接中拥有路径参数时，FastAPI 将识别出与路径参数匹配的函数参数应从路径中获取，而声明为 Pydantic 模型的函数参数应从请求体中获取。

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

app = FastAPI()

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

我们除了定义请求模型之外，通常情况下还需要对每个参数在接口文档中进行说明，否则接口文档中的参数很难被其他人所明白。

声明效验

对请求模型的参数使用 Pydantic 的 Field 即可对参数进行声明校验和元数据等。

class Item(BaseModel):
    name: str
    description: Union[str, None] = Field(
        default=None, title="说明", max_length=300
    )
    price: float = Field(gt=0, description="这个金额必须大于0!")
    tax: Union[float, None] = None

使用Fileld就和前面提到的Query一样。

高级类型定义

在创建请求体时，除了使用常规的基本类型外，还可以使用List类型、Set类型：

class Item(BaseModel):
    tags1: List[str] = []
    tags2: Set[str] = set()
    tags3: Dict[int, float] = {}

不只是请求体，对于请求函数的单独入参也可以这样声明。

还可以接受另外一个模型，用来嵌套：

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


class Item(BaseModel):
    name: str
    image: Union[Image, None] = None

声明示例

对于接口文档，我们可以对请求体的内容声明一个示例，用作在接口文档上示例。

只需要在请求体模型内定义一个Config类，在其schema_extra中定义“example”：

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }

GET查询的请求体参数拼装

前面提到了创建BaseModel作为请求体可以大幅减少请求重复使用的问题，以及规范化请求。但是也提到了只能使用到POST相关的请求接口上，GET相关请求无法使用。

对于GET请求的请求体创建使用，官方称之为使用依赖注入机制。它有2种实现方式：

  • 第一种实现方法：依赖项就是一个函数，且可以使用与路径操作函数相同的参数，它返回对应的GET查询请求的内容：

    async def common_parameters(
    q: Union[str, None] = 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)):
    return commons

    这里创建了一个common_parameters函数当做依赖项。在请求接口的输入参数中输入dict，使用Depends函数。
    这样这个请求接口输入参数就会有 依赖项返回的内容了。

  • 第二种实现方法：创建一个请求类，继承BaseModel，内部参数遵循查询参数写法：

    class Student(BaseModel):
    name: str
    age: Union[int, None] = None


@app.get("/")
def read_root(student: Student = Depends()):
    return {"name": student.name, "age": student.age}是

    直接在GET请求入参下输入对应请求类，默认值上使用Depends()。这样既可将其请求类以查询参数方式进行工作。

PUT更新

更新数据请用 HTTP PUT 操作。

把输入数据转换为以 JSON 格式存储的数据（比如，使用 NoSQL 数据库时），可以使用 jsonable_encoder。例如，把 datetime 转换为 str。

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded

数据类型

FastAPI利用Pydantic来实现类型检验。它支持多种数据类型，其中数据有一些常用的数据类型：

  • int
  • float
  • str
  • bool

除此之外，还有众多其他数据类型：

  • UUID：一种标准的 “通用唯一标识符” ，在许多数据库和系统中用作ID。在请求和响应中将以 str 表示。
  • datetime.datetime：在请求和响应中将表示为 ISO 8601 时间格式的 str ，比如: 2008-09-15T15:53:00+05:00。
    • datetime.date：在请求和响应中将表示为 ISO 8601 格式的 str ，比如: 2008-09-15。
  • Decimal：在请求和相应中被当做 float 一样处理。

更多额外的内置数据类型，可查询Pydantic文档https://pydantic-docs.helpmanual.io/usage/types/

请求头部

要实现对FastAPI接口的请求头部进行发送相关内容的话，需要导入fastapi包下的Header函数。

在接口参数下，定义需要对头部定义参数即可，如下：

from typing import Union
from fastapi import FastAPI, Header

@app.get("/header/")
async def read_items(admin_token: Union[str, None] = Header(default=None)):
    return {"admin-token": admin_token}

注意：大多数标准的headers用 “连字符” 分隔，也称为 “减号” (-)。

但是像 user-agent 这样的变量在Python中是无效的。

因此, 默认情况下, Header 将把参数名称的字符从下划线 (_) 转换为连字符 (-) 来提取并记录 headers。

同时，HTTP headers 是大小写不敏感的，因此，因此可以使用标准Python样式(也称为 “snake_case”)声明它们。因此，您可以像通常在Python代码中那样使用 user_agent ，而不需要将首字母大写为 User_Agent 或类似的东西。

响应模型

前面我们创建了请求模型，它是用作封装请求接口的。

而对于有些响应返回的类型我们在项目中通常也会将其封装，称为响应模型。

它的创建方法和请求模型一样，都是继承与BaseModel类。

class ResultItem(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5

当它直接被用作返回时，也不会被有检验效果。需要在对应请求接口注解上的response_model进行配置：

@app.post("/items/", response_model=Item)
async def create_item(xxx:xxx):
    ...
    # item是一个Item类的实例对象
    return item

有时，为了简便，通常用一个BaseModel既做 请求模型 又做 响应模型。但出现一个隐私问题，比如某个BaseModel中存在一些隐私数据的话，像用户注册时输入的密码，直接返回整个相同的BaseModel是十分不安全的。

对于隐私数据，可以在接口请求注解上使用response_model_exclude来排除某个指定的参数：

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
    "baz": {
        "name": "Baz",
        "description": "There goes my baz",
        "price": 50.2,
        "tax": 10.5,
    },
}

@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude={"tax"})
async def read_item_public_data(item_id: str):
    return items[item_id]

返回的响应数据不一定需要相同类，FastAPI的响应模型只是输出内部对应的数据：

class UserIn(BaseModel):
    username: str
    password: str
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    username: str
    full_name: Union[str, None] = None


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

上述返回的类型在函数内部是一个 UserIn对象结构，但是返回结果为 UserOut对象结构。

表单数据请求

在有时接口接收的不是Json数据，而是Form表单数据，或者上传文件的接口，这时需要用到Form配置。

要使用表单配置，需预先安装 python-multipart。

from fastapi import FastAPI, Form

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

创建表单（Form）参数的方式与 Body 和 Query 一样。

例如，OAuth2 规范的 “密码流” 模式规定要通过表单字段发送 username 和 password。该规范要求字段必须命名为 username 和 password，并通过表单字段发送，不能用 JSON。使用 Form 可以声明与 Body （及 Query、Path、Cookie）相同的元数据和验证。

可在一个路径操作中声明多个 Form 参数，但不能同时声明要接收 JSON 的 Body 字段。因为此时请求体的编码是 application/x-www-form-urlencoded，不是 application/json。

表单数据的「媒体类型」编码一般为 application/x-www-form-urlencoded。

但包含文件的表单编码为 multipart/form-data。

上传文件请求

要实现上传文件的请求，和表单数据请求一样，需要需预先安装 python-multipart。

从 fastapi 导入 File 和 UploadFile：

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes = File()):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

声明文件体必须使用 File，否则，FastAPI 会把该参数当作查询参数或请求体（JSON）参数。

其中 File()文件作为「表单数据」上传。如果把路径操作函数参数的类型声明为 bytes，FastAPI 将以 bytes 形式读取和接收文件内容。这种方式把文件的所有内容都存储在内存里，适用于小型文件。

而 UploadFile则是使用spooled 文件，它在存储到内存的文件超出最大上限时，FastAPI 会把文件存入磁盘，所以UploadFile 能上传更大文件，并且它可获取上传文件的元数据。

大多数情况下，UploadFile 更好使用。

UploadFile 的属性有如下：

  • filename：上传文件名字符串（str），例如， myimage.jpg；
  • content_type：内容类型（MIME 类型 / 媒体类型）字符串（str），例如，image/jpeg；
  • file： SpooledTemporaryFile（ file-like 对象）。其实就是 Python文件，可直接传递给其他预期 file-like 对象的函数或支持库。

UploadFile 支持以下 async 方法，（使用内部 SpooledTemporaryFile）可调用相应的文件方法：

  • write(data)：把 data （str 或 bytes）写入文件；
  • read(size)：按指定数量的字节或字符（size (int)）读取文件内容；
  • seek(offset)：移动至文件 offset （int）字节处的位置；
    • 例如，await myfile.seek(0) 移动到文件开头；
    • 执行 await myfile.read() 后，需再次读取已读取内容时，这种方法特别好用；
  • close()：关闭文件。

上述方法都是 async 方法，要搭配「await」使用。

例如，在 async 路径操作函数 内，要用以下方式读取文件内容：

contents = await myfile.read()

在普通 def 路径操作函数 内，则可以直接访问 UploadFile.file，例如：

contents = myfile.file.read()

多文件上传

FastAPI 支持同时上传多个文件。

对于多文件上传，将其封装到List内即可：

@app.post("/files/")
async def create_files(files: List[bytes] = File()):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile]):
    return {"filenames": [file.filename for file in files]}

抛出错误

抛出普通浏览器错误

当接口出现输入情况下，往往会出现响应的错误状态，如404Not Find等等。

我们可以手动抛出对应的错误状态，只需要使用导入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]}

其中status_code下可以设置响应的浏览器错误码，detail则是错误时返回的Json内容，它接受的是任何可以被Json格式化的数据类型，不只是str。

抛出自定义错误

大多数情况下，在项目中都是抛出自定义的错误，而不是浏览器错误。

对此需要先定义一个自定义错误模型：

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

fastApi需要定义一个全局错误拦截器，捕捉到这个定义的自定义错误，并进行相应处理：

from fastapi.responses import JSONResponse

@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..."},
    )

上述捕捉到自定义错误后返回了一个指定格式的JSON，并设置了状态码为418。其中上述的Request、JSONResponse都是可以使用starlette包下的内容，它会fastapi中的内容一样。

最后当接口函数中抛出这个自定义异常即可触发相应处理：

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}
```### 路径操作配置

在接口请求注解上，除了定义必须的 请求路径外，还可以配置各种信息，作用于接口。

#### 返回状态

`status_code` 用于定义*路径操作*响应中的 HTTP 状态码。可以直接传递 `int` 代码， 比如 `404`。也可以使用`status` 的快捷常量：

```python
from fastapi import status

@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)

状态码在响应中返回，并会被添加到 OpenAPI 概图。

FastAPI 的fastapi.status 和 starlette.status 一样，只是快捷方式。

接口分类

在接口设置tags 参数，对接口进行分类，规范化接口文档。

tags 参数的值是由 str 组成的 list （一般只有一个 str ）。

@app.post("/items/", response_model=Item, tags=["items"])

接口说明

使用summary、description可以对接口进行声明。

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)

summary： 接口标题 、description：接口说明

当其中description说明描述非常长的时候，可以考虑使用docstring，它支持多段文字，以及支持 Markdown，能正确解析和显示 Markdown 的内容，但要注意文档字符串的缩进。它的用法就是在函数内容开头以三个引号开始“”“，三个引号结束，中间进行描述。

@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

Json转换

在某些情况下，您可能需要将数据类型（如 Pydantic 模型）转换为与 JSON 兼容的类型（如dict、list等）。

FastApi内置了一个JSON转换器 - jsonable_encoder()，使用它可以将其自己的数据类型转换为与JSON兼容的数据类型。它的作用是将其数据内部的不可转换的数据类型（比如datetime）转换为Str格式。

from fastapi.encoders import jsonable_encoder

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data

它不会直接转换为JSON的字符串，而是转换为list、dict这种兼容JSON的数据类型。

自定义响应

对于RestFul的接口，往往会返回一个Result JSON作为请求的响应。

Result的格式一般为请求状态码、响应信息、返回数据这三个内容。

对于FastAPI来说需要定义自定义响应模型，通常在Pydantic模型下文件夹创建一个resp.py用作存储自定义响应模型：

"""

统一响应状态码

"""
from typing import Union

from fastapi import status as http_status
from fastapi.responses import JSONResponse, Response
from fastapi.encoders import jsonable_encoder

class Resp(object):
    def __init__(self, status: int, msg: str, code: int):
        self.status = status
        self.msg = msg
        self.code = code

    def set_msg(self, msg):
        self.msg = msg
        return self


InvalidRequest: Resp = Resp(1000, "无效的请求", http_status.HTTP_400_BAD_REQUEST)
InvalidParams: Resp = Resp(1002, "无效的参数", http_status.HTTP_400_BAD_REQUEST)
BusinessError: Resp = Resp(1003, "业务错误", http_status.HTTP_400_BAD_REQUEST)
DataNotFound: Resp = Resp(1004, "查询失败", http_status.HTTP_400_BAD_REQUEST)
DataStoreFail: Resp = Resp(1005, "新增失败", http_status.HTTP_400_BAD_REQUEST)
DataUpdateFail: Resp = Resp(1006, "更新失败", http_status.HTTP_400_BAD_REQUEST)
DataDestroyFail: Resp = Resp(1007, "删除失败", http_status.HTTP_400_BAD_REQUEST)
PermissionDenied: Resp = Resp(1008, "权限拒绝", http_status.HTTP_403_FORBIDDEN)
ServerError: Resp = Resp(5000, "服务器繁忙", http_status.HTTP_500_INTERNAL_SERVER_ERROR)
    
def ok(*, data: Union[list, dict, str] = None, pagination: dict = None,  msg: str = "success") -> Response:
    return JSONResponse(
        status_code=http_status.HTTP_200_OK,
        content=jsonable_encoder({
            'status': 200,
            'msg': msg,
            'data': data,
            'pagination': pagination
        })
    )

def fail(resp: Resp) -> Response:
    return JSONResponse(
        status_code=resp.code,
        content=jsonable_encoder({
            'status': resp.status,
            'msg': resp.msg,
        })
    )

这里的返回类型为Response，Response 类接受如下参数：

  • content - 一个 str 或者 bytes。
  • status_code - 一个 int 类型的 HTTP 状态码。
  • headers - 一个由字符串组成的 dict。
  • media_type - 一个给出媒体类型的 str，比如 "text/html"。

这里由JSONResponse函数生成相应的返回值。

所以这里的响应体中的 status_code是指的浏览器响应状态码，而content内的status是请求状态码。

在请求接口中 返回使用即可：

@app.get("/users/",summary="查询所有用户")
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = user_dao.get_all_users(db, skip=skip, limit=limit)
    userList = []
    for useronde in users:
        usersd = user.User.from_orm(useronde)
        userList.append(usersd)
    return resp.ok(data=userList)

安全认证

FastAPI 提供了多种工具，可帮助你以标准的方式轻松、快速地处理安全性，而无需研究和学习所有的安全规范。

FastAPI 在 fastapi.security 模块中为每个安全方案提供了几种工具，这些工具简化了这些安全机制的使用方法。

为此我们学习使用OAuth2方式进行安全认证：

实现要实现启动OAuth2认证前，让我们来看一些小的概念：

OAuth2

OAuth2是一个规范，它定义了几种处理身份认证和授权的方法。

它是一个相当广泛的规范，涵盖了一些复杂的使用场景。

它包括了使用「第三方」进行身份认证的方法。

这就是所有带有「使用 Facebook，Google，Twitter，GitHub 登录」的系统背后所使用的机制。

实现OAuth2

使用“表单数据”来发送usernameand password，OAuth2来判断用户输入的账户和密码是否正确。如果正确就返回Token给请求方，请求方再用这个Token给其它需要认证的接口来进行操作。

由于这里涉及使用到表单数据，所以就项目中需要安装python-multipart.

在fastapi.security包中导入OAuth2PasswordBearer：

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

这里的tokenUrl 则是定义其认证头部名，这里定义为token。

这样项目就完成了初步安全加密了，运行后发现：

您的路径操作上右上角多了一个小锁，您可以单击它，它有一个小授权表格来输入一个username和password（和其他可选字段），表示用来授权登录操作。

当然这里由于只设置了 认证要求，但没设置认证方法，所以这里并不能完成认证。

设置认证方法

上面只是展示了开启OAuth2认证的步骤，但是并没有编写认证方法。这里来进行编写认证方法：

所谓认证方法，就是用户认证登录的操作，在FastAPI的OAuth2认证中，默认以/token作为其认证方法路径，也就是说当用户登录时上传的路径其实是这个路径。

OAuth2的认证方法入参参数类型为OAuth2PasswordRequestForm，所以首先，导入 OAuth2PasswordRequestForm，然后在 token 的路径操作中通过 Depends 将其作为依赖项使用。

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    ...

OAuth2PasswordRequestForm 是一个类依赖项，声明了如下的请求表单：

  • username。
  • password。
  • 一个可选的 scope 字段，是一个由空格分隔的字符串组成的大字符串。
  • 一个可选的 grant_type.
  • 一个可选的 client_id
  • 一个可选的 client_secret

一般情况下，用户只需要传入username和password即可。

认证方法中内容则是 对于用户输入的信息进行认证。一般步骤有：

  1. 根据用户输入的username在数据库中寻找是否存在，如果存在则继续，如果不存在就返回输入错误给用户。
  2. 根据用户输入的password来与 数据库查询到的用户进行对比，如果相同则登录成功，返回用户信息。如果不相同，则返回输入错误给用户。

当然这个步骤在实际生产环境下禁止使用，因为里面的信息并不是加密信息，实际环境下数据库中不可能用明文存储密码的。当然这是下部分考虑的，目前先看下简单的操作。

这里我们用dict来创建一个”假的数据库”：

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}

这里的拥有两个用户，其中hashed_password字段存储的是经过加密的密码（当然这里只是在密码前加了fakehashed，实际上不可能这样）。

随后我们创建一个用户模型：

class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None
        
class UserInDB(User):
    hashed_password: str

这个User模型这里主要是用来登录成功返回用户信息的，它跟数据库中的字段是有差别的，没有敏感信息（比如密码），因为我们在登录成功后返回用户信息中，不能直接将密码返回给Response中，这是不安全的。

而UserInDB则是在此阶段多个敏感字段（这里指密码）。

创建完模型后，就可以在认证方法中编写从数据库中获取用户的信息步骤：

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    ...

这儿UserInDB(**user_dict)将其获取到的user_dict用户信息内容设置到了UserInDB对象中。

编写加密方式函数，用作对用户输入的密码进行加密操作：

def fake_hash_password(password: str):
    return "fakehashed" + password

在认证方法中将其用户输入的密码与数据库中的密码进行对比，最后返回指定信息即完成认证方法的编写。

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}

这里返回的响应必须是一个 JSON 对象。它应该有一个 token_type。在我们的例子中，由于我们使用的是「Bearer」令牌，因此令牌类型应为「bearer」。并且还应该有一个 access_token 字段，它是一个包含我们的访问令牌的字符串。这时FastAPI中的OAuth2的定义规范。

当然对于这个简单的示例，我们将极其不安全地返回相同的 username 作为令牌。

设置认证判断

上面我们编写了认证方法，用户在认证中成功认证后，将返回用户的名称作为 认证口令。

但是这只是认证方法，对于需要认证的 接口而言 并不起作用，因为它们不知道怎么判断是否认证。所以我们还需要编写认证判断的代码。

假设，我们编写了一个read_users_me 接口，它需要根据当前认证的用户返回认证用户的信息，如果没有认证，则报错：

@app.get("/users/me")
async def read_users_me(...):
    return current_user

对于这个我们需要编写一个获取当前用户的函数作为依赖项：

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},