FastAPI 入门

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

# 安装与启动#

首先安装 FastAPI

pip install "fastapi[all]"

以上安装还包括了 uvicorn ，可以将其用作运行代码的服务器。

执行代码时，使用一下命令

uvicorn main:app --reload

服务器上后台启动

nohup uvicorn main:app --host 0.0.0.0 --port 8000 > /文件路径/log.log 2>&1 &

接着执行

bg

可以将一个在后台暂停的作业切换到继续在后台执行

最后执行

disown

用于将一个作业从当前终端会话中分离，使其不会受到终端会话的影响，退出终端仍然运行

uvicorn main:app 命令含义如下:

main：main.py 文件（一个 Python「模块」）。
app：在 main.py 文件中通过 app = FastAPI() 创建的对象，也可以是其他名称
--reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

此时访问 localhost:8000 即可调用

# Hello World#

## 新建一个 main.py 文件#

from fastapi import FastAPI

app = FastAPI()

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

代码解释

from fastapi import FastAPI

FastAPI 是一个为你的 API 提供了所有功能的 Python 类。

app = FastAPI()

这里的变量 app 会是 FastAPI 类的一个「实例」。

这个实例将是创建所有 API 的主要交互对象。

这个 app 同样在启动的命令中被 uvicorn 所引用

也可以写成 myapi = FastAPI ()
启动命令就变成 uvicorn main --reload

@app.get("/")

告诉 FastAPI 在它下方的函数负责处理如下访问请求

请求的路径为 /

使用 get 操作调用

同样可以使用 @app.post () 、 @app.put () 、 @app.delete ()

async def root():

定义一个 Python 函数，函数名为 root

async 表示这是一个异步函数

return {"message": "Hello World"}

返回内容

## 执行文件#

uvicorn main:app --reload

执行成功

# 路径参数#

## 路径中传入参数#

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

{item_id} 表示这是一个可以传入的参数

此时请求 localhost:8000/items/1

会返回

{ "item_id": "1"}

## 限制参数类型#

同时，可以给这个参数限制类型

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

只需要在函数入参的后面加上 : 类型

此时如果传入一个字符串，将会返回 422 错误

类型可以使用 str、float、bool 等等

注意！

需要注意顺序问题

比如：

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": user_id}

现在有两个接口，如果 /users/{user_id} 在 /users/me 前面

由于_路径操作_是按顺序依次运行的

那当请求 /users/me 的时候，会执行到 /users/{user_id} 接口，返回 “me” 字符串

所以需要确保 /users/me 在 /users/{user_id} 的前面

## 枚举类型预设值#

### 创建枚举类#

from fastapi import FastAPI
# 引入枚举类型
from enum import Enum

# 创建枚举类
class ModelNameEnum(str, Enum):
      fire = "fire"
      peach = "peach"
      firepeach = "firepeach"

app = FastAPI()


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

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

@app.get("/models/{model_name}")
async def get_model(model_name: ModelNameEnum):
    # 通过枚举类的值来判断
    if model_name is ModelNameEnum.fire:
        return {"model_name": model_name, "message": "Fire Fire Fire"}
    if model_name.value == "peach":
        return {"model_name": model_name, "message": "Peach Peach Peach"}
    return {"model_name": model_name, "message": "FirePeach FirePeach FirePeach"}

代码解释：

async def get_model(model_name: ModelNameEnum):

定义的入参 model_name 是 ModelNameEnum 枚举类型

if model_name is ModelNameEnum.fire:

这里的 is 和 == 的作用是一样的，用来和 枚举成员 进行比较

但是使用 == 的时候，需要用入参 .value 来获取值

model_name.value

### 测试请求#

## 文件路径作为入参#

如果需要传入的入参是一个路径

可以使用

/files/{file_path:path}

作为请求路径

@app.get("/files/{file_path:path}") 
async def read_file(file_path: str): 
	return {"file_path": file_path}

当需要传入 /home/test/text.txt 路径的时候

请求的入参为 localhost:8000/files//home/test/text.txt 注意为两个 /

# 查询参数#

## 默认值#

声明不属于路径参数的其他函数参数时，它们将被自动解释为 "查询字符串" 参数

比如下面的代码

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]

在函数入参中

async def read_item(skip: int = 0, limit: int = 10):

定义了默认值 skip 默认为 0，limit 默认为 10

所以

访问 http://127.0.0.1:8000/items/?skip=0&limit=10 的时候

传入的参数就是 skip = 0 , limit = 10

访问 http://127.0.0.1:8000/items/?skip=1 的时候

传入的参数就是 skip = 1 , limit = 10，因为 limit 不传时，默认为 10

访问 http://127.0.0.1:8000/items/?limit=1 的时候

传入的参数就是 skip = 0 , limit = 1，skip 不传默认为 0

## 可选值#

需要引入 Union

from typing import Union

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

此时，q 就是可传可不穿的参数

还可以加入 bool 类型的入参

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        # 将 q 作为 item 的一个键值对，update 可以将新的键值对合并进去
        item.update({"q": q})
    if not short:
         item.update({"description": "This is an amazing item that has a long description"})
    return item

使用 update ({"XX" : "XX"})，可以在原有的键值对中新增这个键值对拼接上去

测试访问

## 多个路径和查询参数#

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Union[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

可以支持多个参数的请求

## 必须的查询参数#

如果在非路径的参数中指定了一个没有默认值的参数，那么这个参数就是必传的

也就是说不是在 @app.get (/XXX/{ XXX}) 中指定的，是在函数的入参中指定的

而且没有配置 Union[str, None] = None , 那么这个参数就是必传的

@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 参数就是必传

# 请求体#

当你需要将数据从客户端（例如浏览器）发送给 API 时，你将其作为「请求体」发送。

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

你的 API 几乎总是要发送响应体。但是客户端并不总是需要发送请求体。

## 导入 BaseModel#

from pydantic import BaseModel

需要用到 pydantic 模型来声明请求体

## 创建数据模型#

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

这时候就声明了一个名称为 Item 的请求体的结构类型

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

接着添加一个 post 方法，就可以指定入参为 Item 模型类型

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

测试请求

## 使用模型#

在函数的内部，可以直接访问模型对象的所有属性

但是需要通过 模型.dict () 获取模型的字典才可以进行修改

模型.dict () 就是模型中的每一个属性单独拎出来

@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

## 请求体加路径参数#

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

可以同时传入路径参数和请求体

注意返回 item.dict () 的时候要加上 **

可以看出如果返回 item.dict () 就是把模型中的每一个属性，单独的拎出来

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

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

# 参数校验#

查询参数校验#

首先需要导入 Query

from fastapi import FastAPI, Query

### 长度校验#

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

可以在参数后加上 Query ()

由于必须用 Query(default=None) 替换默认值 None，Query 的第一个参数同样也是用于定义默认值。

上面的代码表示 q 这个字段最小长度为 2 最大长度为 5

### 正则表达式校验#

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=2, max_length=5,regex='^query$')):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

regex='^query$'

上面的代码指定了正则表达式， q 的内容必须为 query

### 默认值#

同时默认值也可以是指定的

default = 'query'

## 路径参数的校验#

需要引入 Path

from fastapi import FastAPI,Query,Path

添加方法

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

@app.get("/items4/{item_id}")
async def read_items6(

    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

gt：大于（greater than）
ge：大于等于（greater than or equal）
lt：小于（less than）
le：小于等于（less than or equal）

传递 * 作为函数的第一个参数。

Python 不会对该 * 做任何事情，但是它将知道之后的所有参数都应作为关键字参数（键值对），也被称为 kwargs，来调用。即使它们没有默认值。

也就是说，如果不加 * 作为第一个参数的话，函数入参里不能只填写 q: str
因为他没有默认值

# 请求体中多个 Model 参数#

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class User(BaseModel):
    username: str
    full_name: Union[str, None] = 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

之后的入参就可以使用下面这种形式

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

## 请求体中的单一值#

如果想在同一个请求体里面在新增一个键值对

比如

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

那么可以引入 Body

from fastapi import FastAPI,Query,Path,Body

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

通过添加下面代码来实现

key: str = Body()

同样的 Body () 中也可以和 Query、Path 一样添加校验

key: str = Body(gt=1)

测试接口

## 嵌入单个请求体参数#

如果期望实现的是下面这种请求方式

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

而不是

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

则可以使用 Body () 的 embed 参数

比如

item: Item = Body(embed=True)

from typing import Union

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = 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

# 请求体中的嵌套模型#

## 具有子类型的 List 字段#

首先引入 List

from typing import List, Union

比如需要传入多个标签

可以使用下面代码来表示 List

# 限制内容的类型，只能传入 str 类型
tags: List[str] = []

# 不限制内容的类型，str、int等等都可以
tags:List =[]

代码如下

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


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

## Set 类型#

和 List 不同的是，Set 中的元素不可以重复，所以用于存放标签会更合适

因为重复的元素会被自动合并

一样需要从 typing 中引入 Set

from typing import Set

使用代码

tags: Set[str] = set()

代码如下

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()
  
@app.post("/chat")
async def chat(message: Message, model: str = Body()):
    return {"message":message,"model":model}

## 嵌套模型#

可以在模型中，嵌套另一个模型

例如

{
    "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"
    }
}

先定义子模型

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

接着添加到原来的模型中

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()
    
    image: Union[Image,None] = None

## 特殊的类型和校验#

除了普通的单一值类型（如 str、int、float 等）外，你还可以使用从 str 继承的更复杂的单一值类型。

外部的特殊类型，可以通过 Pydantic 引入，具体可以访问 Pydantic 官网

比如 image 中的 URL 应该是个地址，相对应的 Pydantic 提供了 HttpUrl 的类型用来校验

同样的先引入

from pydantic import BaseModel, HttpUrl

接着修改 image 模型类

class Image(BaseModel):

    url: HttpUrl
    
    name: str

此时如果 URL 传入的不是 http 或者 https 开头的地址，都会报错

## 带有一组子模型的属性#

同样的，还可以将 Pydantic 模型用作 list、set 等的子类型

比如

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()

    image: Union[List[Image],None] = None

# 响应模型#

## 用户注册响应#

可以在任意的_路径操作_中使用 response_model 参数来声明用于响应的模型：

比如用户注册，注册成功后返回用户数据

但是密码是不能直接返回的，所以可以设置一个没有密码的响应模型

用户返回定的模型的出参

首先引入 Any 类型和 EmailStr

from typing import Any

from pydantic import EmailStr

定义注册用户的模型

class User(BaseModel):
	username:str
	full_name: Union[str,None] = None
	email: EmailStr
	password: str

定义注册成功后返回的响应模型

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

实现代码如下

@app.post("/user/" response_model= UserOut):
async def create_user(user User) -> Any:
	return user

这里的 Any 是指返回的 user 可以是任何类型

测试接口

## 预设模型响应#

首先定义一个相应模型

class Item2(BaseModel):
    name: str
    description: Union[str,None] = None
    price: float
    tax: float = 10
    tags: List[str] = []

接着设置默认的返回

item2s = {
    "fire": {"name": "Fire", "price": 3.2},
    "peach": {"name": "Peach", "description": "Peach Peach Peach", "price": 2.2,"tags":["aa","bb","cc"]},
    "firepeach": {"name": "firepeach", "description": "firepeach firepeach firepeach", "price": 2.2}
}

接口实现如下

@app.get("/item2s/{item_id}" response_model=Item2)
async def read_item2(item_id: str):
	return item2s[item_id]

此时请求入参发送 item_id 就会返回 item2s 中对应的 item_id 的那条数据

测试调用

如果不想返回那些带有默认值的参数

可以添加一个参数 response_model_exclude_unset=True

修改代码如下

@app.get("/item2s/{item_id}", response_model=Item2 , response_model_exclude_unset=True)
async def read_item2(item_id: str):
    return item2s[item_id]

此时接口调用，将不会返回 数据内容为默认值，或则内容为空 的数据

与默认值不同的仍然会返回

# 额外的模型#

保存用户数据的时候，需要新增一个存入数据库的模型

因为密码不能通过明文去保存，例如可以用 hash 值去保存

新增 UserDB 模型

class UserDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: Union[str,None] = None

定义一个假的密码加密函数

def fake_password_hasher(raw_password: str):
    return "hashpassword" + raw_password

定义一个拼接 UserDB 模型类型的数据并保存到数据库的函数

def fake_save_user(user_in: User):
    fake_hash_password = fake_password_hasher(user_in.password);
    # 可以使用 ** 将一个字典解包成关键字参数
    # 然后将解包好的 user_in 作为关键字参数传入 user_db
    # 并将密码加密成 hashed_password 一起赋予 user_db
    user_db = UserDB(**user_in.dict(), hashed_password=fake_hash_password);

    print("用户假的保存成功！");
    return user_db```

新增一个注册用户的接口

``` python
@app.post("/user2/" response_model=UserOut)
async def create_user2(user: User):
	user_save = fake_save_user(user)
	return user_save

测试接口

## 关于 `**user.dict()` #

### Pydantic 模型#

user 是一个 User 类的 Pydantic 模型.

Pydantic 模型具有 .dict（） 方法，该方法返回一个拥有模型数据的 dict （字典）。

所以如果创建一个 User 模型的对象是

user_in = UserIn(username="john", password="secret", email="[email protected]")

如果 user.dict ()

user_dict = user.dict()

user_dict 的数据则是

{
    'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
}

也就是所有的属性都被单独出来

### 解包 dict#

如果将 user_dict 这样的 dict 以 **user_dict 形式传递给一个函数（或类），Python 将对其进行「解包」。它会将 user_dict 的键和值作为关键字参数直接传递。

所以如果使用以下代码

UserDB(**user_dict)
# 或者
UserDB(**user.dict())

此时我们得到的 UserDB 则是

UserDB(
	'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
)

或者更确切地，直接使用 user_dict 来表示将来可能包含的任何内容：

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
)

然后添加额外的关键字参数 hashed_password=hashed_password，例如：

UserDB(**user.dict(),fake_hashed_password = fake_hashed_password)

最终的结果就会是

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
    hashed_password = hashed_password,
)

## 关于代码重复#

代码重复会增加出现 bug、安全性问题、代码失步问题（当你在一个位置更新了代码但没有在其他位置更新）等的可能性。

上面的这些模型都共享了大量数据，并拥有重复的属性名称和类型。

因此，基于上面提到的 dict 和解包，可以直接定义一个 UserBase 的基类

class UserBase(BaseModel):
	username: str
	email: EmailStr
	full_name: Union[str,None] = None

而其他的模型只需要在此基础上，只赋予新增的属性

例如

class User(BaseModel):
    password: str


class UserDB(BaseModel):
    hashed_password: str


class UserOut(BaseModel):
    # pass 代表不做任何处理，保持和 UserBase 一致
    pass

因为后面接口的使用中基本所有的字段都会有，只需要解决新增的字段

## Union 模型#

可以将一个响应声明为两种类型的 Union，这意味着该响应将是两种类型中的任何一种。

这将在文档中表示为 anyOf

例如

 response_model=Union[PlaneItem, CarItem]

代码如下

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class BaseItem(BaseModel):
    description: str
    type: str

class CarItem(BaseItem):
    type = "car"

class PlaneItem(BaseItem):
    type = "plane"
    size: int

items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

## 模型列表#

可以用同样的方式声明由对象列表构成的响应。

response_model=List[Item]

代码如下

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str


items = [
    {"name": "Foo", "description": "There comes my hero"},
    {"name": "Red", "description": "It's my aeroplane"},
]


@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

# 响应状态码#

可以通过 status_code 来指定状态码

例如

@app.post("/user2/", response_model=UserOut,status_code=201)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

成功返回时，将返回 201 的状态码

同时 fastapi 也有提供便捷的变量

引入

from fastapi import status

@app.post("/user2/", response_model=UserOut,status_code=status.HTTP_201_CREATED)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

常用的状态码有

有关每个状态代码以及适用场景的更多信息可以参考 HTTP 响应状态码 - HTTP | MDN

# 表单数据#

与 JSON 不同，HTML 表单（<form></form>）向服务器发送数据通常使用「特殊」的编码。

接收的不是 JSON，而是表单字段时，要使用 Form。

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

执行

pip install python-multipart

导入 fastapi 提供的 Form

from fastapi import FastAPI, Form
app = FastAPI()

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

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

例如，OAuth2 规范的 "密码流" 模式规定要通过表单字段发送 username 和 password。

该规范要求字段必须命名为 username 和 password，并通过表单字段发送，不能用 JSON。

使用 Form 可以声明与 Body （及 Query、Path、Cookie）相同的元数据和验证。

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

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

# 请求文件#

# 处理错误#

某些情况下，需要向客户端返回错误提示。

这里所谓的客户端包括前端浏览器、其他应用程序、物联网设备等。

需要向客户端返回错误提示的场景主要如下：

客户端没有执行操作的权限
客户端没有访问资源的权限
客户端要访问的项目不存在
等等 ...

遇到这些情况时，通常要返回 4XX（400 至 499）HTTP 状态码。

4XX 状态码与表示请求成功的 2XX（200 至 299） HTTP 状态码类似。

只不过，4XX 状态码表示客户端发生的错误。

## HTTPException#

向客户端返回 HTTP 错误响应，可以使用 HTTPException。

from fastapi import HTTPException

例如

@app.get("/item2s/{item_id}", response_model=Item2 , response_model_exclude_unset=True)
async def read_item2(item_id: str):
    if item_id not in item2s:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND , detail="没这个文件")
    return item2s[item_id]

测试调用

## 安装自定义异常处理器#

添加自定义处理器，要使用 Starlette 的异常工具。

假设要触发的自定义异常叫作 UnicornException。

且需要 FastAPI 实现全局处理该异常。

此时，可以用 @app.exception_handler() 添加自定义异常控制器：

也可以直接引入 fastapi 封装好的

from fastapi import  Request
from fastapi.response import JSONResponse

代码如下

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

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
	    content={"message": f"哈哈! {exc.name} 这里出错了..."},
    )

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

{"message": f"哈哈! {exc.name} 这里出错了..."} 中的 f
可以获取自定义错误名

测试结果

# JSON 兼容编码器与更新数据#

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

比如，如果您需要将其存储在数据库中。

对于这种要求， FastAPI提供了jsonable_encoder()函数。

## 使用 jsonable_encoder () 更新数据#

假设有一个数据库名为fake_db，它只能接收与 JSON 兼容的数据。

例如，它不接收datetime这类的对象，因为这些对象与 JSON 不兼容。

因此，datetime对象必须将转换为包含ISO 格式化的str类型对象。

同样，这个数据库也不会接收 Pydantic 模型（带有属性的对象），而只接收dict。

对此可以使用 jsonable_encoder。

更新数据请用 HTTP PUT 操作。

它接收一个对象，比如 Pydantic 模型，并会返回一个 JSON 兼容的版本：

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

from datetime import datetime

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

# 假的数据库
fake_db = {
    "fee": {"title": "foo", "timestamp": datetime.now()}
}

# 定义一个ItemJ 模型
class ItemJ(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

# put 用来更新
@app.put("/itemsJ/{id}")
def update_itemJ(id: str, item: ItemJ):
	# 将时间传唤成 str 类型的 Json格式
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data
    return fake_db[id]

测试调用

# 安全性#

假设后端 API 在某个域。

前端在另一个域，或（移动应用中）在同一个域的不同路径下。

并且，前端要使用后端的 username 与 password 验证用户身份。

固然，FastAPI 支持 OAuth2 身份验证。

## 概览#

引入 Depends 依赖和 OAuth2PasswordBearer 身份认证

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

此时在 api 文档 localhost:8000/docs 中可以看到

页面右上角出现了一个「Authorize」按钮。

_路径操作_的右上角也出现了一个可以点击的小锁图标。

点击 Authorize 按钮，弹出授权表单，输入 username 与 password 及其它可选字段：

如果没有，需要安装 python-multipart

执行 pip install python-multipart

是因为 OAuth2 使用 表单数据 发送 `username` 与 `password`。

### 密码流#

Password 流是 OAuth2 定义的，用于处理安全与身份验证的方式（流）。

OAuth2 的设计目标是为了让后端或 API 独立于服务器验证用户身份。

但在本例中，FastAPI 应用会处理 API 与身份验证。

下面是简化的运行流程：

用户在前端输入 username 与password，并点击回车
（用户浏览器中运行的）前端把 username 与password 发送至 API 中指定的 URL（使用 tokenUrl="token" 声明）
API 检查 username 与password，并用令牌（Token）响应（暂未实现此功能）：
令牌只是用于验证用户的字符串
一般来说，令牌会在一段时间后过期
- 过时后，用户要再次登录
- 这样一来，就算令牌被人窃取，风险也较低。因为它与永久密钥不同，在绝大多数情况下不会长期有效
前端临时将令牌存储在某个位置
用户点击前端，前往前端应用的其它部件
前端需要从 API 中提取更多数据：
- 为指定的端点（Endpoint）进行身份验证
- 因此，用 API 验证身份时，要发送值为 Bearer + 令牌的请求头 Authorization
- 假如令牌为 foobar，Authorization 请求头就是： Bearer foobar

### OAuth2PasswordBearer#

FastAPI 提供了不同抽象级别的安全工具。

本例使用 OAuth2 的 Password 流以及 Bearer 令牌（Token）

为此要使用 OAuth2PasswordBearer 类。

创建 OAuth2PasswordBearer 的类实例时，要传递 tokenUrl 参数。该参数包含客户端（用户浏览器中运行的前端）的 URL，用于发送 username 与 password，并获取令牌。

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

在此，tokenUrl="token" 指向的是暂未创建的相对 URL token。这个相对 URL 相当于 ./token。

因为使用的是相对 URL，如果 API 位于 https://example.com/，则指向 https://example.com/token。但如果 API 位于 https://example.com/api/v1/，它指向的就是https://example.com/api/v1/token。

使用相对 URL 非常重要，可以确保应用在遇到使用代理这样的高级用例时，也能正常运行

### 使用#

oauth2_scheme 变量是 OAuth2PasswordBearer 的实例，也是可调用项。

调用方式为

oauth2_scheme(some, parameters)

因此，Depends （依赖）可以调用 oauth2_scheme 变量。

接下来，使用 Depends 把 oauth2_scheme 传入依赖项。

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

该依赖项使用字符串（str）接收_路径操作函数_的参数 token

## 获取当前的用户#

### 创建用户模型#

class UserAuth(BaseModel):
    username: str
    # 先使用 Union[str, None]，后面会改成 EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

### 创建依赖项#

创建一个 get_current_user 依赖项。

get_current_user 将具有一个之前所创建的同一个 oauth2_scheme 作为依赖项。

与之前直接在路径操作中所做的相同，新的依赖项 get_current_user 将从子依赖项 oauth2_scheme 中接收一个 str 类型的 token：

async def get_current_user(token: str = Depends(oauth2_scheme)): 
	user = fake_decode_token(token) 
	return user

### 获取用户信息#

将从 oauth2_scheme 获取到的 token 用来获取用户

get_current_user 将使用创建的（伪）工具函数 fake_decode_token ，该函数接收 str 类型的令牌并返回我们的 Pydantic User 模型

def fake_decode_token(token): 
	return User( username=token + "fakedecoded", email="[email protected]", full_name="John Doe" )

### 注入当前用户#

现在可以在 路径操作 中使用 get_current_user 作为 Depends 了：

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

## 使用密码和 Bearer 的简单 OAuth2#

### 概述#

现在将使用 FastAPI 的安全性实用工具来获取 username 和 password。

OAuth2 规定在使用「password 流程」时，客户端 / 用户必须将 username 和 password 字段作为表单数据发送。

而且规范明确了字段必须这样命名。因此 user-name 或 email 是行不通的。

不过数据库模型也可以使用任何其他名称。

但是对于登录 路径操作，需要使用这些名称来与规范兼容（以具备例如使用集成的 API 文档系统的能力）。

规范还写明了 username 和 password 必须作为表单数据发送（因此，此处不能使用 JSON）。

### 获取 `username` 和 `password`#

#

需要用到 OAuth2PasswordRequestForm

首先引入

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

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

    user_dict = fake_users_db.get(form_data.username)
	# 如果 user_dict 不存在 
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

	# 将 User 的数据 传给 UserDB 模型类
    user = UserDB(**user_dict)
    # 将密码 hash 化
    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")

	# 返回 用户名 token的类型
    return {"access_token": user.username, "token_type": "bearer"}

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

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

#### 密码哈希化#

「哈希」的意思是：将某些内容（在本例中为密码）转换为看起来像乱码的字节序列（只是一个字符串）。

每次你传入完全相同的内容（完全相同的密码）时，你都会得到完全相同的乱码。

但是你不能从乱码转换回密码。

    # 将密码 hash 化
    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")

#### 返回 Token#

token 端点的响应必须是一个 JSON 对象。

它应该有个 token_type。在例子中，使用的是「Bearer」令牌，因此令牌类型应为「bearer」。

并且还应该有一个 access_token 字段，它是一个包含我们的访问令牌的字符串。

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

	# 返回 用户名 token的类型
    return {"access_token": user.username, "token_type": "bearer"}

#### 更新依赖项#

现在将更新依赖项。

我们想要仅当此用户处于启用状态时才能获取 current_user。

因此，创建了一个额外的依赖项 get_current_active_user，而该依赖项又以 get_current_user 作为依赖项。

如果用户不存在或处于未启用状态，则这两个依赖项都将仅返回 HTTP 错误。

因此，在我们的端点中，只有当用户存在，身份认证通过且处于启用状态时，我们才能获得该用户：

# 通过 token 获取用户信息
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"},
        )
    return user

# 获取当前活跃的用户，登陆了之后就活跃了
# 身份认证通过且处于启用状态时，才能获得此用户
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

#### 创建虚假的数据库#

# 模拟数据库

fake_users_db = {
    "bryce": {
        "username": "bryce",
        "full_name": "bryce Yu",
        "email": "[email protected]",
        "hashed_password": "hash123123",
        "disabled": False,
    },

    "yu": {
        "username": "yu",
        "full_name": "yu",
        "email": "[email protected]",
        "hashed_password": "hash123456",
        "disabled": True,
    },
}

### 完整代码（重点！！！！）#

完整的用户登陆到获取当前用户的代码如下

# 创建一个 用户 模型类
class UserAuth(BaseModel):
    username: str
    # 先使用 Union[str, None]，后面会改成 EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

#  创建一个 用户 数据库 模型类
class UserDB(UserAuth):
    hashed_password: str

# 模拟将密码哈希化
def fake_hash_password(raw_password: str):
    # 模拟哈希化 密码，实际上应该使用 bcrypt 或者其他哈希算法
    return "hash" + raw_password

# 通过 token 获取用户信息
def fake_decode_token(token):
    # 通过 token 获取用户信息，现在不安全的使用 用户名 作为 token
    user = get_user(fake_users_db, token)
    return user

# 从 db 数据库中，获取用户名为 username 的用户信息
def get_user(db, username: str):
    # 现在不安全的使用 用户名 作为 token
    if username in db:
        user_dict = db[username]
        return UserDB(**user_dict)



# 先通过通过 oauth2_scheme 获得token ，在使用 token 获取当前登陆的用户信息
# oauth2_scheme 是一个认证方案，它会从请求头中获取 Authorization 字段的值，然后将其解析成 token
async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 通过token 获取用户信息
    user = fake_decode_token(token)
    # 如果用户不存在
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证凭证",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 如果用户存在，返回用户信息
    return user


# 获取当前活跃的用户
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    # 通过 get_current_user 获取当前登陆的用户 current_user
    # 接着判断用户是否被禁用
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户被禁用")
    return current_user


# 通过表单 登陆用户
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):

    # 通过用户名 到 模拟数据库中获取用户
    user_dict = fake_users_db.get(form_data.username)

	# 如果 user_dict 不存在 
    if not user_dict:
        raise HTTPException(status_code=400, detail="用户名或密码错误")

	# 将 User 的数据 传给 UserDB 模型类，获取 UserDB 的实例用来获取密码进行比较
    user = UserDB(**user_dict)

    # 将密码 hash 化，用来和数据库中的密码进行比较
    hashed_password = fake_hash_password(form_data.password)

	# 如果 密码不正确
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    
	# 返回 token 和 token的类型，现在不安全的使用 用户名 作为 token
    return {"access_token": user.username, "token_type": "bearer"}


# 调用这个接口的时候，需要在 headers 中添加 Authorization: Bearer token，token 就是登陆时返回的 token
# 当使用 openApi 测试的时候，会自动在 headers 中添加 Authorization: Bearer token
# 而实际的请求中，需要手动添加 Authorization: Bearer token
@app.get("/users/me")
async def read_users_me(current_user: UserAuth = Depends(get_current_active_user)):
    # 通过 get_current_active_user 获取当前登陆并且没有被禁用的用户 current_user
    return current_user

#### 流程总结#

首先调用 /token 接口，通过表单提交 username 和 password ，接口会返回一个 token

接着需要在 headers 中的 Authorization 中添加 Token 的值，调用 /user/me 获取当前登陆用户

因为本案例中使用的是 Bearer ，所以 Authorization 的值为 "Bearer" + 返回的 Token

而在 openApi 调用测试的时候，为了方便测试，会自动加上 Authorization 的 Token 值

在实际项目的使用中，需要手动添加 headers 中的 Authorization 值

FastAPI 入门

# 安装与启动#

# Hello World#

## 新建一个 main.py 文件#

## 执行文件#

# 路径参数#

## 路径中传入参数#

## 限制参数类型#

## 枚举类型预设值#

### 创建枚举类#

### 测试请求#

## 文件路径作为入参#

# 查询参数#

## 默认值#

## 可选值#

## 多个路径和查询参数#

## 必须的查询参数#

# 请求体#

## 导入 BaseModel#

## 创建数据模型#

## 使用模型#

## 请求体加路径参数#

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

# 参数校验#

查询参数校验#

### 长度校验#

### 正则表达式校验#

### 默认值#

## 路径参数的校验#

# 请求体中多个 Model 参数#

## 请求体中的单一值#

## 嵌入单个请求体参数#

# 请求体中的嵌套模型#

## 具有子类型的 List 字段#

## Set 类型#

## 嵌套模型#

## 特殊的类型和校验#

## 带有一组子模型的属性#

# 响应模型#

## 用户注册响应#

## 预设模型响应#

# 额外的模型#

## 关于 **user.dict() #

### Pydantic 模型#

### 解包 dict#

## 关于代码重复#

## Union 模型#

## 模型列表#

# 响应状态码#

# 表单数据#

# 请求文件#

# 处理错误#

## HTTPException#

## 安装自定义异常处理器#

# JSON 兼容编码器 与 更新数据#

## 使用 jsonable_encoder () 更新数据#

# 安全性#

## 概览#

### 密码流#

### OAuth2PasswordBearer#

### 使用#

## 获取当前的用户#

### 创建用户模型#

### 创建依赖项#

### 获取用户信息#

### 注入当前用户#

## 使用密码和 Bearer 的简单 OAuth2#

### 概述#

### 获取 username 和 password#

#

#### 密码 哈希 化#

#### 返回 Token#

#### 更新依赖项#

#### 创建虚假的数据库#

### 完整代码 （重点！！！！）#

#### 流程总结#

## 关于 `**user.dict()` #

# JSON 兼容编码器与更新数据#

### 获取 `username` 和 `password`#

#### 密码哈希化#

### 完整代码（重点！！！！）#