# fastapi_base

**Repository Path**: linyaofai/fastapi_base

## Basic Information

- **Project Name**: fastapi_base
- **Description**: fastapi 基础项目，案例项目
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 4
- **Created**: 2022-09-27
- **Last Updated**: 2022-09-27

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# FastAPI

中文官网文档：https://fastapi.tiangolo.com/zh/

FastAPI官方文档教程完整翻译：https://blog.csdn.net/xuqingskywolf/category_10022295.html

FastAPI B站讲解：https://www.bilibili.com/video/BV1iN411X72b?p=1

B站视频源码：https://github.com/liaogx/fastapi-tutorial

官方推荐 FastAPI Web项目生成器：https://github.com/tiangolo/full-stack-fastapi-postgresql

FastAPI + MySQL Web项目生成器：https://github.com/CoderCharm/fastapi-mysql-generator

FastPI 相关第三方扩展：https://github.com/mjhea0/awesome-fastapi



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

这短短的一句话提出了FastAPI最重要也是让我最喜欢的三个优点：

**现代**：发布于2018年，可使用最低Python版本3.6

**高性能**：目前Python Web框架中是处理速度最快的，基于 Starlette 和 Pydantic，是 FastAPI 如此高性能的重要原因。

**类型提示**：FastAPI 基于 Pydantic ，Pydantic 主要用来做类型强制检查。参数赋值，不符合类型要求，就会抛出异常。（不强制使用类型提示，但是使用最好）

还有一个特别喜欢的优点：

**标准化**：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。（自动生成API文件）

缺点？ 没有缺点



依赖

Python 3.6 及更高版本

FastAPI 站在以下巨人的肩膀之上：

- Starlette 负责 web 部分。

- Pydantic 负责数据部分。



提前了解：

Python基础语法

Python async/await 新特性

Python 3.6新特性 类型提示：https://fastapi.tiangolo.com/zh/python-types/

Python typing 类型提示 新特性：https://docs.python.org/zh-cn/3/library/typing.html

Python 3.4新特性 枚举类：https://blog.csdn.net/weixin_30765505/article/details/100091862



官方用户指南：

1. 对来自不同地方的参数进行声明，如：**请求头**、**cookies**、**form 表单**以及**上传的文件**。
2. 如何设置**校验约束**如 `max_length` 或者 `regex`。
3. 一个强大并易于使用的 **依赖注入** 系统。
4. 安全性和身份验证，包括通过 **JWT 令牌**和 **HTTP 基本身份认证**来支持 **OAuth2**。
5. 更进阶（但同样简单）的技巧来声明 **多层嵌套 JSON 模型** （借助 Pydantic）。
6. 许多额外功能（归功于 Starlette）比如：
   - **WebSockets**
   - **GraphQL**
   - 基于 `requests` 和 `pytest` 的极其简单的测试
   - **CORS**
   - **Cookie Sessions**
   - ......以及更多



## 上手

安装基础环境：

```
pip install fastapi
pip install uvicorn

或者 

安装fastapi所有依赖：
pip install fastapi[all]
```

![image-20211014134123522](images/image-20211014134123522.png)

创建并进入项目目录：

```
mkdir fastapi_base
cd fastapi_base
```

编写接口代码：

代码文件：main.py
```python
# 基于同步接口代码
from typing import Optional  # 使变量值可以为None
from fastapi import FastAPI

app = FastAPI()   # 创建项目实例，应用程序


# 使用app装饰器将此函数定义为接口
# 路由为：/
@app.get("/")
def read_root():
    return "Hello World"

# 路由中出现{}为路径参数，会直接传入到函数参数中
# item_id 参数为整数类型，必传
# q 参数为字符串类型，使用了Optional表示可以传None，不传则默认值为None
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}
```

```python
# 基于异步接口代码
from typing import Optional
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return "Hello World"


@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}
```

运行项目：
```python
uvicorn main:app --reload
# uvicorn
# main: 表示app应用程序所在的文件名称
# app：FastAPI实例
# reload：debug模式，修改项目内容后会自动重启

# 启动显示：
?[32mINFO?[0m:     Will watch for changes in these directories: ['E:\\study_fastapi']
?[32mINFO?[0m:     Uvicorn running on ?[1mhttp://127.0.0.1:8000?[0m (Press CTRL+C to quit)
?[32mINFO?[0m:     Started reloader process [?[36m?[1m6836?[0m] using ?[36m?[1mstatreload?[0m
?[32mINFO?[0m:     Started server process [?[36m10692?[0m]
?[32mINFO?[0m:     Waiting for application startup.
?[32mINFO?[0m:     Application startup complete.

```


访问：
```
使用浏览器访问接口 http://127.0.0.1:8000
显示：
"Hello World"

使用浏览器访问接口 http://127.0.0.1:8000/items/5?q=somequery
显示：
{"item_id":5,"q":"somequery"}

使用浏览器访问交互式 API 文档：http://127.0.0.1:8000/docs

使用浏览器访问ReDoc API 文档：http://127.0.0.1:8000/redoc
```

## 问题集合

### 发现FastApi自动生成文档问题

启动FastApi后，访问接口正常，启动正常，但是打不开文档，有js报错
FastApi版本：0.68.1

问题解决：无法使用版本过低的浏览器打开


![1632922968(1).jpg](http://file.ktianc.top/files/images/61546dadf1b94d421274c031/)

![1632922903(1).jpg](http://file.ktianc.top/files/images/61546dabaec2c0a2ef90a8b6/)

![1632923146(1).jpg](http://file.ktianc.top/files/images/61546e22c689c02d68788e8c/)

## 官方推荐

官方推荐ujson JSON解析器：https://github.com/ultrajson/ultrajson

### ujson

可用作 Python 的大多数其他 JSON 解析器的替代品:

```
>>> import ujson
>>> ujson.dumps([{"key": "value"}, 81, True])
'[{"key":"value"},81,true]'
>>> ujson.loads("""[{"key": "value"}, 81, true]""")
[{'key': 'value'}, 81, True]
```



## 简单示例

```python
from typing import Optional
from pydantic import BaseModel

# FastAPI 是直接从 Starlette 继承的类。
# 你可以通过 FastAPI 使用所有的 Starlette 的功能。
from fastapi import FastAPI

# 这里的变量 app 会是 FastAPI 类的一个「实例」。
# 这个实例将是创建你所有 API 的主要交互对象。
# 这个 app 同样在启动命令中被 uvicorn 所引用
app = FastAPI()


# 定义用户基本数据类型
class User(BaseModel):
    username: str
    age: int
    is_active: Optional[bool] = True  # 使用Optional后，表示当前字段可以为None可以传入None，默认为True

# @app.get("/") 告诉 FastAPI 在它下方的函数负责处理如下访问请求：
# 请求路径为 /
# 使用 get 请求方式
@app.get("/")
async def read_root():
    return "Hello World"


@app.post("/user/{id}")
async def create_user(id: int, user: User):
    return {"id": id, "username": user.username, "age": user.age, "is_active": user.is_active}

# 启动命令：uvicorn main:app --reload
```

测试接口：

打开交互式接口文档：http://127.0.0.1:8000/docs

![1632925601(1).jpg](http://file.ktianc.top/files/images/615477ada349a546ab0d0665/)

![1632925715(1).jpg](http://file.ktianc.top/files/images/6154781daec2c0a2ef90a8b9/)

![1632925811(1).jpg](http://file.ktianc.top/files/images/6154787c1be57911deb58ed3/)

## 搭建基础项目结构

1. 在项目下创建run主启动文件，以及Python包（每一个应用为一个包，用来做区分，类似django中的app）study

2. study包下创建case01.py文件，并配置一下内容：

   ```python
   # -*- coding: utf-8 -*-
   # @version        : 1.0
   # @Creaet Time    : 2021/9/29 23:03 
   # @File           : case01.py
   # @IDE            : PyCharm
   # @desc           : 案例01，请求操作
   
   from fastapi import APIRouter
   
   case01 = APIRouter()
   
   
   @case01.get("/", summary="案例1初始接口")
   def case01_root():
       return "case01_root"
   ```

3. 并将此文件引入到应用下的`__init__`.py文件中：

   ```python
   # -*- coding: utf-8 -*-
   # @version        : 1.0
   # @Creaet Time    : 2021/9/29 23:02 
   # @File           : __init__.py
   # @IDE            : PyCharm
   # @desc           : 初始化
   
   from study.case01 import case01
   ```
   
4. 配置项目根目录下的run主启动文件

   ```python
   #!/usr/bin/python
   # -*- coding: utf-8 -*-
   # @version        : 1.0
   # @Creaet Time    : 2021/9/29 22:39 
   # @File           : run.py
   # @IDE            : PyCharm
   # @desc           : fastapi 学习项目
   
   from fastapi import FastAPI
   import uvicorn
   from study import case01  # 引入应用
   
   app = FastAPI()
   
   # 引入应用中的路由
   app.include_router(case01, prefix="/study/case01", tags=["案例1"])
   
   if __name__ == '__main__':
       # 启动项目
       # reload：自动重载项目
       # debug：调试
       # workers：启动几个进程
       uvicorn.run('run:app', host="127.0.0.1", port=8000, reload=True, debug=True, workers=1)
   ```

## 路由参数解析与验证

### 请求操作

```python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/9/29 23:03 
# @File           : case01.py
# @IDE            : PyCharm
# @desc           : 案例01，请求操作


"""
请求操作：
路径参数，路径顺序，枚举类型参数，文件路径类型参数，路径参数验证，定义查询参数，bool类型转换
查询参数验证，定义请求体与验证，多种参数混合，请求体嵌套

额外数据类型：https://fastapi.tiangolo.com/zh/tutorial/extra-data-types/
"""

"""
七种类型参数：
路径参数：fastapi.Path
查询参数：fastapi.Query
请求体参数：pydantic.BaseModel（pydantic.Field），请求体中不能添加路径参数（Path） 与 查询参数（Query）
Header参数：fastapi.Header，fastapi.Cookie
Request参数：fastapi.Request
文件参数：fastapi.File，fastapi.UploadFile
表单参数：fastapi.Form
"""

from datetime import datetime
from enum import Enum
from fastapi import APIRouter, Path, Query
from typing import List
from pydantic import BaseModel, Field

case01 = APIRouter()


@case01.get("/", summary="案例1初始接口")
def case01_root():
    return "case01_root"


@case01.get("/user/{id}", summary="路径参数")
def get_user(id: int):
    """
    路径参数id的值会直接作为实参id传递给函数get_user。<br>
    并指定了id参数类型为int整数，如果传入的参数不为整数，那么会报错。<br>
    注意：如果传入的参数为"3"，那么会将字符串3自动转换为整数3<br>
    也就是说，借助类型声明，FastAPI可以对Request内容自动进行数据解析和数据转换。<br>
    """
    return {"id": id}


@case01.get("/path/params/validate/{num}", summary="路径参数验证")
def path_params_validate(
        num: int = Path(..., title="整数字段", description="大于1，小于10", ge=1, le=10),
):
    """
    使用fastapi.Path函数可以自定义路径参数验证规则
    """
    return {"num": num}


@case01.get("/user/admin", summary="路径顺序，固定路径")
def hello_user():
    """
    如果存在固定路径与参数路径可能发生一致的情况，并且他们的请求方法一致 <br>
    那么应该确保固定路径应该放在参数路径前面 <br>
    否则将一直无法访问到固定路径接口函数
    """
    return "hello admin"


@case01.get("/user/{name}", summary="路径顺序，参数路径")
def welcome_user(name: str):
    return f"welcome {name}"  # Python3.6 版本开始出现了此新的格式化字符串语法，性能提升


# 定义性别枚举类
class Sex(str, Enum):
    man = "man"
    woman = "woman"


@case01.get("/sex/{sex}", summary="枚举类型参数")
def get_sex(sex: Sex):
    """
    定义枚举类型参数后，在接口文档中，<br>
    在传递这个参数时，会显示选择框，选择项为定义的枚举类属性值
    如果传递的参数值不为枚举类中的数据值，那么就会报错：“value”不是有效的枚举成员
    """
    return {"sex": sex}


@case01.get("/file/{filepath: path}", summary="文件路径类型参数")
def download_file(filepath: str):
    """
    在传递路径参数时，如果参数值为一个文件路径，例如：/usr/local/README.md<br>
    那么系统不会认为是这是一整个参数，而是会将它当做路由的一部分进行解析，则会报错："detail": "Not Found"<br>
    所以我们需要在这种文件路径参数后面指定为path类型，这样系统就会将后面这一部分解析为字符串
    """
    return f"download {filepath}"


@case01.get("/users/", summary="定义查询参数")
def get_users(page: int, limit: int = None):
    return {"page": page, "limit": limit}


@case01.get("/bool/type/conversion", summary="bool类型转换")
def bool_type_conversion(is_active: bool = False):
    """
    bool类型参数，在接口文档中同样会自动显示为选择框
    但是可以直接通过访问URL来测试：http://127.0.0.1:8000/study/case01/bool/type/conversion?is_active=false
    参数值中：yes，on，1，True，true会被自动转换为True
    """
    return {"is_active": is_active}


@case01.get("/query/params/validate", summary="定义查询参数")
def query_params_validate(
        page: int = Query(..., title="页码", description="大于1，小于10", ge=1, le=10),
        name: str = Query(..., title="名称", description="大于1，小于10", min_length=1, max_length=10),
        roles: List[str] = Query(default=["admin", "quest"], alias="角色列表", description="添加了别名，与默认值")
):
    return {"page": page, "name": name, "roles": roles}


class User(BaseModel):
    """
    定义请求体数据模型<br>
    请求体中不能添加路径参数（Path） 与 查询参数（Query）<br>
    注意：GET请求不支持传输请求体参数，只能使用POST, PUT, DELETE等<br>
    GET请求报错：TypeError: Failed to execute 'fetch' on 'Window': Request with GET/HEAD method cannot have body.
    """
    username: str = Field(..., title="用户名", min_length=1, max_length=20)
    password: str = Field(default="123456", title="密码", min_length=6, max_length=16)
    create_time: datetime = datetime.now()

    class Config:
        # 示例参数值，会默认显示在接口文档中
        schema_extra = {
            "example": {
                "username": "admin",
                "password": "admin123",
                "create_time": "2020-01-01 00:01:02"
            }
        }


@case01.post("/request/data/validate", summary="定义请求体参数并验证")
def request_data_validate(user: User):
    return {"username": user.username, "password": user.password, "create_time": user.create_time}


@case01.post("/params/validate/{id}", summary="多种参数混合")
def params_validate(
        id: int,
        user: User,
        page: int = Query(..., title="查询参数")
):
    return {"id": id,"username": user.username, "password": user.password, "create_time": user.create_time, "page": page}


class Users(BaseModel):
    users: List[User] = None  # 请求体嵌套
    create_user_id: int = Field(..., title="整数类型", ge=0)
    desc: str = None


@case01.post("/request/data/nesting", summary="请求体嵌套")
def request_data_nesting(users: Users):
    return {"create_user_id": users.create_user_id, "desc": users.desc, "user": users.users}
```

### 特殊请求操作

新添加视图文件流程

1. 应用目录中新建视图文件 `case02.py`

   ```python
   # -*- coding: utf-8 -*-
   # @version        : 1.0
   # @Creaet Time    : 2021/10/12 22:27 
   # @File           : case02.py
   # @IDE            : PyCharm
   # @desc           : 案例02，特殊请求操作
   
   from fastapi import APIRouter
   
   case02 = APIRouter()
   
   
   @case02.get("/", summary="案例2初始接口")
   def case02_root():
       return "case02_root"
   ```

2. 在应用 `__init__.py`文件中引入路由对象

   ```python
   from study.case02 import case02
   ```

3. 在 `run.py`主文件中引入路由

   ```python
   app.include_router(case02, prefix="/study/case02", tags=["案例2，特殊请求操作"])
   ```

4. 完成

```python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/12 22:27 
# @File           : case02.py
# @IDE            : PyCharm
# @desc           : 案例02，特殊请求操作


"""
特殊请求操作：
1. 定义 Cookie 参数
2. 定义 Header 参数
3. 定义 Request 参数
4. 接收表单数据
5. 单文件、多文件参数上传
"""

"""
使用Form接收表单数据时，需要执行安装：pip install python-multipart
"""

from fastapi import APIRouter, Cookie, Header, Request, Form, File, UploadFile
from typing import Optional, List

case02 = APIRouter()


@case02.get("/", summary="案例2初始接口")
def case02_root():
    return "case02_root"


@case02.get("/cookie/param", summary="定义Cookie参数")
def cookie_param(cookie_id: Optional[str] = Cookie(None)):
    """
    使用 Cookie 函数定义参数，可以直接获取到Cookie中的对应参数<br>
    例子：Cookie: cookie_id=xxxxxxxx
    如果需要直接获取到Cookie中的所有值，可以使用Header函数
    """
    return {"cookie_id": cookie_id}


@case02.get("/header/param", summary="定义Header参数")
def header_param(user_agent: Optional[str] = Header(None, convert_underscores=True), x_token: List[str] = Header(None)):
    """
    使用 Header 函数定义参数，可以直接获取到请求体中的对应参数<br>
    convert_underscores：如果请求体中为user-agent，则会将-转换为_匹配
    """
    return {"user_agent": user_agent, "x_token": x_token}


@case02.get("/request/param", summary="定义 Request 参数")
def request_param(request: Request, item: str):
    """
    使用 Request 类型参数，为请求体，可获取到所有请求内容<br>
    """
    print(item)
    print(request)
    print(request.url)
    print(request.client.host)
    print(request.headers)
    print(request.cookies)
    return {"url": request.url}

@case02.post("/reqeust/form", summary="接收表单数据")
def request_form(
        username: str = Form(..., title="用户名"),
        message: str = Form(..., title="留言内容")
):
    """
    接收表单数据，使用 Form 函数，其中基本属性与其他字段函数一致<br>
    请求时会发现'Content-Type: application/x-www-form-urlencoded' 不再为json格式

    """
    return {"username": username, "message": message}


@case02.post("/file", summary="File，上传单个文件")
def file(file: bytes = File(...)):
    """
    传递的参数类型为 bytes 字节数据类型，会以 bytes 形式写入内存<br>
    所以只能获取到数据内容，无法获取到文件信息，适合小文件上传<br>
    """
    return {"file_size": len(file)}


@case02.post("/files", summary="File，上传多个文件")
def files(files: List[bytes] = File(...)):
    return {"file_size": len(files)}


@case02.post("/upload/file", summary="UploadFile，上传单个文件")
async def upload_file(file: UploadFile = File(...)):
    """
    传递的参数类型为 UploadFile 类型，是直接可以通过Python操作的文件对象，能获取到文件信息<br>
    UploadFIle 具有以下 async 异步方法：<br>
    write(data)：写入data ( str 或 bytes ) 到文件<br>
    read(size)：读取文件的 size (int) 个字节/字符<br>
    seek(offset)：转到文件中的字节位置 offset(int)，如： await myfile.seek(0)  将转到文件的开头<br>
    close()：关闭文件<br>
    异步接口，所以需要在接口函数前加上 async<br>
    适合上传大文件
    """
    result = {
        "filename": file.filename,
        "content-type": file.content_type,
        "read": len(await file.read())
    }
    return result


@case02.post("/upload/files", summary="UploadFile，上传多个文件")
async def upload_files(files: List[UploadFile] = File(...)):
    for file in files:
        print(file.filename)
        print(file.content_type)
        contents = await file.read()
        print(contents)
    return "upload files"


@case02.post("/upload/files/save", summary="UploadFile，上传多个文件，并保存到本地")
async def upload_files_save(files: List[UploadFile] = File(...)):
    """
    首先先在项目根目录下创建一个static目录，用来存在静态资源，我们吧上传的文件都保存到这个目录中<br>
    然后我们使用异步方式读取上传过来的文件内容，读取后写入到本地文件中即可。

    异步方法原理：
    读取文件内容时是一个异步协程操作，执行流程：
    1. 先读取一个文件内容，这个属于IO操作，比较耗时，但是fastapi不会在这里等待读取完成
    2. 而是会再去获取到一个文件对象，继续执行读取操作
    """
    for file in files:
        try:
            data = await file.read()
            with open('static/' + file.filename, "wb") as f:
                f.write(data)
            return "success"
        except Exception as e:
            return str(e)
```

## 响应操作

额外响应模型：https://fastapi.tiangolo.com/zh/tutorial/extra-models/

```python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/13 09:00
# @File           : case03.py
# @IDE            : PyCharm
# @desc           : 案例03，响应操作


"""
1. 声明响应模型
2. 额外响应模型：https://fastapi.tiangolo.com/zh/tutorial/extra-models/
3. JSON 兼容编码器：https://blog.csdn.net/xuqingskywolf/article/details/106383755
4. 直接返回响应
5. 返回状态码
6. 错误处理：https://fastapi.tiangolo.com/zh/tutorial/handling-errors/
"""
from datetime import datetime

"""
EmailStr字段需要单独下载：pip install pydantic[email]
"""

from fastapi import APIRouter, Response, status, HTTPException
from typing import Optional, Union, List, Dict
from pydantic import BaseModel, Field, EmailStr
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, ORJSONResponse, HTMLResponse

case03 = APIRouter()


@case03.get(path="/",
            summary="案例3",
            description="初始接口",
            status_code=status.HTTP_200_OK,
            name="案例三",
            response_description="响应描述")
def case03_root():
    return "case03_root"


# =========================================================声明响应模型=================================================


class User(BaseModel):
    id: int
    username: str = Field(..., title="用户名")
    password: str = Field(..., title="密码")
    nickname: Optional[str] = Field(None, title="昵称")
    email: EmailStr = Field(None, title="邮箱")
    address: Optional[str] = Field(None, title="地址")
    telephone: str = Field(default=None, min_length=11, max_length=11, title="手机号")


@case03.post("/response/all", summary="声明响应模型", response_model=User)
def response_all(user: User):
    """
    response_model：表示声明响应的模型
    """
    return user


class UserOut(BaseModel):
    id: int
    username: str = Field(..., title="用户名")
    nickname: Optional[str] = Field(None, title="昵称")
    telephone: str = Field(default=None, min_length=11, max_length=11, title="手机号")


@case03.post("/response/model", summary="重新定义声明响应模型", response_model=UserOut)
def response_model(user: User):
    """
    只会返回 response_model 响应模型中定义的数据字段
    """
    return user


@case03.post("/response/exclude/unset", summary="响应模型，只返回用户传递的数据字段", response_model=User, response_model_exclude_unset=True)
def response_exclude_unset(user: User):
    """
    response_model_exclude_unset=True，表示在响应模型中只返回用户传递的数据字段
    """
    return user


@case03.post("/response/model/include", summary="响应模型，指定返回数据字段", response_model=User, response_model_include=["id", "username"])
def response_model_include(user: User):
    """
    response_model_include：用于指定响应模型中返回的数据字段
    """
    return user


@case03.post("/response/model/exclude", summary="响应模型，指定不返回数据字段，排除字段", response_model=User, response_model_exclude=["password", "telephone"])
def response_model_exclude(user: User):
    """
    response_model_exclude：用于指定响应模型不返回数据字段，排除字段
    """
    return user


# =========================================================额外响应模型=================================================


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


class AbstractOut(UnionItem):
    type = "car"


class DetailOut(UnionItem):
    type = "plane"
    size: int


union_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,
    },
}


@case03.post("/response/model/union/{item}", summary="Union响应模型", response_model=Union[DetailOut, AbstractOut])
def response_model_union(item: str):
    """
    你可以将一个响应，声明为两种类型的 Union，这意味着该响应将是两种类型中的任何一种。

    定义一个 Union 类型时，首先包括最详细的类型，然后是不太详细的类型。
    在下面的示例中，更详细的 DetailOut 位于 Union[DetailOut，AbstractOut] 中的 AbstractOut 之前。
    """
    return union_items[item]


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


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


@case03.post("/response/model/list", summary="响应模型列表", response_model=List[Item])
def response_model_list():
    """
    声明返回一个响应模型列表
    """
    return items


@case03.post("/response/model/dict", summary=" dict 构成的响应", response_model=Dict[str, float])
def response_model_dict():
    """
    还可以使用一个任意的普通 dict 声明响应，仅声明键和值的类型，而不使用 Pydantic 模型。
    如果你事先不知道有效的字段/属性名称（对于 Pydantic 模型是必需的），这将很有用。
    在这种情况下，你可以使用 typing.Dict：
    """
    return {"foo": 2.3, "bar": 3.4}


# =========================================================JSON 兼容编码器==============================================


json_db = {}


class JsonItem(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None


@case03.post("/json/item/{id}")
def create_json_item(id: str, item: JsonItem):
    """
    在某些情况下，您可能需要将数据类型（例如 Pydantic 模型）转换为与 JSON 兼容的数据（例如 dict、list 等）。

    假设您有一个仅接收 JSON 兼容数据的数据库 json_db。

    这个数据库不直接接收 datetime 对象，和 Pydantic 模型（具有属性的对象）。

    因此，需要将 datetime 对象转换为包含 ISO格式 数据的 str。Pydantic 模型转换为 dict

    在这个例子中，jsonable_encoder 将 Pydantic 模型转换为 dict，将 datetime 转换为 str。其值和子值都与 JSON 兼容。

    不使用 jsonable_encoder 的结果：得到的是一个 Pydantic 模型<br>
    {'test': JsonItem(title='string', timestamp=datetime.datetime(2021, 10, 15, 0, 0), description='string')}

    使用 jsonable_encoder 的结果：得到的是一个 dict<br>
    {'test': {'title': 'string', 'timestamp': '2021-10-15T00:00:00', 'description': 'string'}}

    如果没有return 则 response.body 默认为null
    """
    json_compatible_item_data = jsonable_encoder(item)
    json_db[id] = json_compatible_item_data
    print(json_db)


# =========================================================直接返回响应=================================================


@case03.post("/json/response", summary="直接返回响应，JSONResponse")
def json_response(item: JsonItem):
    """
    当你创建一个 FastAPI 路径操作 时，你可以正常返回以下任意一种数据：dict，list，Pydantic 模型，数据库模型等等。

    FastAPI 默认会使用 jsonable_encoder 将这些类型的返回值转换成 JSON 格式，

    然后，FastAPI 会在后台将这些兼容 JSON 的数据（比如字典）放到一个 JSONResponse 中，该 JSONResponse 会用来发送响应给客户端。

    但是你可以在你的 路径操作 中直接返回一个 JSONResponse。

    直接返回响应可能会有用处，比如返回自定义的响应头和 cookies。

    你可以返回任意 Response 或者任意 Response 的子类。，JSONResponse 本身是一个 Response 的子类。

    当你返回一个 Response 时，FastAPI 会直接传递它，不会做任何转换。
    """
    json_compatible_item_data = jsonable_encoder(item)
    return JSONResponse(content=json_compatible_item_data)


@case03.get("/custom/response", summary="直接返回响应，自定义 Response")
def custom_response():
    """
    上面的例子说明了FastAPI的默认响应方式，以及 JSONResponse 例子

    现在，让我们看看你如何才能返回一个自定义的响应。

    假设你想要返回一个 XML 响应。

    你可以把你的 XML 内容放到一个字符串中，放到一个 Response 中，然后返回。

    其他全部的响应都继承自主类 Response。

    Response 类接受如下参数：

    - content - 一个 str 或者 bytes。
    - status_code - 一个 int 类型的 HTTP 状态码。
    - headers - 一个由字符串组成的 dict。
    - media_type - 一个给出媒体类型的 str，比如 "text/html"。
    """
    data = """<?xml version="1.0"?>
        <shampoo>
        <Header>
            Apply shampoo here.
        </Header>
        <Body>
            You'll have to use soap here.
        </Body>
        </shampoo>
        """
    return Response(content=data, media_type="application/xml")


@case03.get("/orjson/response", summary="声明响应，ORJSONResponse", response_class=ORJSONResponse)
def orjson_response():
    """
    如果你需要压榨性能，你可以安装并使用 orjson 并将响应设置为 ORJSONResponse。

    导入你想要使用的 Response 类（子类）然后在 路径操作装饰器 中声明它。

    安装：pip install orjson
    """
    return [{"item_id": "Foo"}]


@case03.get("/html/response", response_class=HTMLResponse, summary="声明响应，HTMLResponse")
async def html_response():
    return """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """


@case03.get("/custom/html/response", response_class=HTMLResponse, summary="自定义响应后，声明响应，HTMLResponse")
async def custom_html_response():
    """
    路径操作函数 直接返回的 Response 不会被接口文档记录（比如，Content-Type 不会被文档记录），并且在接口文档中也是不可见的。而是显示默认的 JSONResponse

    如果你想要在函数内重载响应，但是同时想在接口文档中显示所使用的 Media type，你可以使用 response_class 参数并返回一个 Response 对象。

    接着 response_class 参数只会被用来在接口文档中生成 Media type，你的 Response 用来返回响应。
    """
    html_content = """
        <html>
            <head>
                <title>Some HTML in here</title>
            </head>
            <body>
                <h1>Look ma! HTML!</h1>
            </body>
        </html>
        """
    return HTMLResponse(content=html_content, status_code=200)


"""
HTMLResponse
如上文所述，接受文本或字节并返回 HTML 响应。

PlainTextResponse
接受文本或字节并返回纯文本响应。

JSONResponse
接受数据并返回一个 application/json 编码的响应。
如上文所述，这是 FastAPI 中使用的默认响应。

ORJSONResponse
如上文所述，ORJSONResponse 是一个使用 orjson 的快速的可选 JSON 响应。
ORJSONResponse 可能是一个更快的选择。

UJSONResponse
UJSONResponse 是一个使用 ujson 的可选 JSON 响应。
在处理某些边缘情况时，ujson 不如 Python 的内置实现那么谨慎。

RedirectResponse
返回 HTTP 重定向。默认情况下使用 307 状态代码（临时重定向）。

StreamingResponse
采用异步生成器或普通生成器/迭代器，然后流式传输响应主体。
对类似文件的对象使用 StreamingResponse
如果您有类似文件的对象（例如，由 open() 返回的对象），则可以在 StreamingResponse 中将其返回。

FileResponse
异步传输文件作为响应。
与其他响应类型相比，接受不同的参数集进行实例化：
- path - 要流式传输的文件的文件路径。
- headers - 任何自定义响应头，传入字典类型。
- media_type - 给出媒体类型的字符串。如果未设置，则文件名或路径将用于推断媒体类型。
- filename - 如果给出，它将包含在响应的 Content-Disposition 中。
文件响应将包含适当的 Content-Length，Last-Modified 和 ETag 的响应头。
"""


# =========================================================状态码=================================================


@case03.post("/response/code", summary="指定默认响应状态码", status_code=200)
def response_code(user: User):
    """
    status_code：表示响应状态码
    """
    return user


@case03.post("/response/code/status", summary="指定默认响应状态码，使用预定义状态码", status_code=status.HTTP_200_OK)
def response_code_status(user: User):
    """
    status 中包含了很多已经定义好的状态码，会更加直观
    """
    return user


@case03.get("/response/code/condition/{id}", summary="自定义响应状态码")
def response_code_condition(id: int, response: Response):
    if id < 10:
        response.status_code = status.HTTP_404_NOT_FOUND

    return id


# =========================================================错误处理=================================================


@case03.get("/response/exception/{id}", summary="错误处理")
def response_exception(id: int):
    if id < 10:
        # 注意：这里使用的是raise抛出异常，而不是return
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="未找到此ID")

    return id
```

## 项目文档配置

```python
class FastAPI(Starlette):
    def __init__(
        self,
        *,
        debug: bool = False,
        routes: Optional[List[BaseRoute]] = None,
        title: str = "FastAPI",
        description: str = "",
        version: str = "0.1.0",
        openapi_url: Optional[str] = "/openapi.json",
        openapi_tags: Optional[List[Dict[str, Any]]] = None,
        servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        docs_url: Optional[str] = "/docs",
        redoc_url: Optional[str] = "/redoc",
        swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
        swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,
        middleware: Optional[Sequence[Middleware]] = None,
        exception_handlers: Optional[
            Dict[
                Union[int, Type[Exception]],
                Callable[[Request, Any], Coroutine[Any, Any, Response]],
            ]
        ] = None,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        terms_of_service: Optional[str] = None,
        contact: Optional[Dict[str, Union[str, Any]]] = None,
        license_info: Optional[Dict[str, Union[str, Any]]] = None,
        openapi_prefix: str = "",
        root_path: str = "",
        root_path_in_servers: bool = True,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        **extra: Any,
    ) -> None:
```

使用：

```python
from fastapi import FastAPI

app = FastAPI(
    title="FastAPI 学习项目",
    description="主要讲解了 FastAPI 的基本使用，并完整开发一个项目。",
    version="1.0.0",
)

"""
其他配置：
docs_url：配置交互文档的路由地址，如果禁用则为None，默认为 /docs
redoc_url： 配置Redoc文档的路由地址，如果禁用则为None，默认为 /redoc
openapi_url：配置接口文件json数据文件路由地址，如果禁用则为None，默认为/openapi.json
"""
```

![image-20211013163134280](images/image-20211013163134280.png)

## 请求方法参数配置

1. get方法：获取数据
2. post方法：创建数据
3. put方法：更新数据
4. delete方法：删除数据

请求方法基类：

```python
def api_route(
        self,
        path: str,
        *,
        response_model: Optional[Type[Any]] = None,
        status_code: Optional[int] = None,
        tags: Optional[List[str]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        response_description: str = "Successful Response",
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        methods: Optional[List[str]] = None,
        operation_id: Optional[str] = None,
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_by_alias: bool = True,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = Default(JSONResponse),
        name: Optional[str] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        openapi_extra: Optional[Dict[str, Any]] = None,
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
```

使用：

```python
@app.get(path="/",
            summary="案例3",
            description="初始接口",
            status_code=status.HTTP_200_OK,
            name="案例三",
            response_description="响应描述")
def app_root():
    return "app_root"

"""
其他路径操作参数：
include_in_schema：使用参数 include_in_schema 并将其设置为 False ，来从生成的 OpenAPI 方案中排除一个 路径操作（这样一来，就从自动化文档系统中排除掉了）。
"""
```

![image-20211013165729606](images/image-20211013165729606.png)

## 静态文件访问

在项目主文件`run.py`中添加配置：

```python
from starlette.staticfiles import StaticFiles

# 挂载静态目录，并添加路由访问，此路由不会在接口文档中显示
app.mount("/static", app=StaticFiles(directory="./static"))

# 访问：127.0.0.1:8000/static/123.png
```

## 异常处理

官方文档：https://fastapi.tiangolo.com/zh/tutorial/handling-errors/

使用默认异常处理器

```python
from fastapi import status, HTTPException

@case03.get("/response/exception/{id}", summary="异常处理")
def response_exception(id: int):
    if id < 10:
        # 注意：这里使用的是raise抛出异常，而不是return
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="未找到此ID")

    return id
```

访问后返回结果：

```python
"""
访问：http://127.0.0.1:8000/study/case03/response/exception/2

HTTP异常结果：
{
  "detail": "未找到此ID"
}
"""
-----------------------------------------------------------------------------------------
"""
访问：http://127.0.0.1:8000/study/case03/response/exception/a

验证异常结果：
{
    "detail":[
        {
            "loc":[
                "path",
                "id"
            ],
            "msg":"value is not a valid integer",
            "type":"type_error.integer"
        }
    ]
}
"""
```

### 自定义异常处理函数

在`run.py`项目主文件中添加 自定义异常处理器

```python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.exceptions import RequestValidationError
from starlette import status


app = FastAPI()


# 重写HTTPException异常处理器
@app.exception_handler(StarletteHTTPException)
async def unicorn_exception_handler(request: Request, exc: StarletteHTTPException):
    """
    request 参数不能省略不写
    """
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "message": exc.detail,
        }
    )


# 重写请求验证异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """
    request 参数不能省略不写
    """
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "message": str(exc),
        }
    )
```

访问后返回结果：

```python
"""
访问：http://127.0.0.1:8000/study/case03/response/exception/2

HTTP异常结果：
{
  "message": "未找到此ID"
}
"""
-----------------------------------------------------------------------------------------
"""
访问：http://127.0.0.1:8000/study/case03/response/exception/a

验证异常结果：
{"message":"1 validation error for Request\npath -> id\n  value is not a valid integer (type=type_error.integer)"}
"""
```

## 依赖注入

```python
#!/usr/bin/python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/14 22:49 
# @File           : case04.py
# @IDE            : PyCharm
# @desc           : 案例04，依赖注入


"""
依赖注入官方文档：https://fastapi.tiangolo.com/zh/tutorial/dependencies/

官方介绍：
"依赖注入"首先意味着在程序中我们的代码可以声明一些它必须依赖的项：我们称之为dependencies，也就是依赖项。
然后，在实际运行中，FastAPI会把所有需要的依赖项提供给你的代码，称之为"注入"依赖项。

"依赖注入"非常适用于以下使用场景：
1、业务逻辑复用
2、共享数据库连接
3、安全机制、权限校验、角色权限等等
4、其他使用场景
所有上述使用场景，借助于"依赖注入"可以明确的提高代码复用，减少代码重复。
"""

"""
依赖项主要分为：
1. 函数依赖项
2. 类依赖项
3. 子依赖项
4. 路径操作装饰器依赖项
5. 全局依赖项
6. 使用 yield 的依赖和子依赖

依赖项统一使用 fastapi.Depends 声明

依赖注入主要为后面讲解安全机制，共享数据库链接做铺垫。
"""


from fastapi import APIRouter, Response, status, HTTPException, Depends, Header
from typing import List, Optional


# =========================================================5.全局依赖项=================================================


"""
有时，我们要为整个应用 或 整个项目添加依赖项。

通过与定义 路径装饰器依赖项 类似的方式，可以把依赖项添加至整个 FastAPI 项目，或 APIRouter 整个应用。
"""


async def app_verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def app_verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


case04 = APIRouter(dependencies=[Depends(app_verify_token), Depends(app_verify_key)])


@case04.get(path="/", summary="案例4")
def case04_root():
    return "case04_root"


# =========================================================1.函数依赖项=================================================


async def page_parameters(page: int = 1, limit: int = 10):
    print("函数依赖项 公共执行体")
    return {"page": page, "limit": limit}


@case04.get("/get/books", summary="函数依赖项")
async def get_books(page: dict = Depends(page_parameters)):
    """
    依赖项就是一个函数，且可以使用与 路径操作函数 相同的参数：例如：Path，Query

    依赖项使用 Depends 声明

    这里只能传给 Depends 一个参数。且该参数必须是可调用对象，比如函数。

    接收到新的请求时，FastAPI 执行如下操作：

    - 用正确的参数调用依赖项函数（「可依赖项」）
    - 获取函数返回的结果
    - 把函数返回的结果赋值给路径操作函数的参数
    """
    return page


@case04.get("/get/users", summary="函数依赖项")
async def get_users(page: dict = Depends(page_parameters)):
    return page


# =========================================================2.类依赖项=================================================


class CommonQueryParams:
    def __init__(self, page: int = 1, limit: int = 10):
        self.page = page
        self.limit = limit

    def public(self):
        print("类依赖项 公共执行体")


@case04.get("/get/class/depends", summary="类依赖项")
# async def get_class_depends(commons: CommonQueryParams = Depends(CommonQueryParams)):  # 默认语法
async def get_class_depends(commons: CommonQueryParams = Depends()):
# 也可以省略 Depends 中的参数不写，是一种 FastAPI 提供的一种捷径，会直接使用参数变量的 类型 作为 Depends 的参数
# async def get_class_depends(commons: Depends(CommonQueryParams)):
# 也可以省略声明类型不写，不过这样写编辑器不知道该参数是什么类型，不会有类属性和方法提示，建议使用上面方式声明类型
    """
    在 FastAPI 中，您可以使用 Python 类作为依赖项。

    FastAPI 实际检查的是它是否为『可调用的』（函数、类或者其他任何东西）。

    如果您在 FastAPI 中将『可调用的』作为依赖项传递，它将分析该『可调用的』参数，并以与路径操作函数的参数相同的方式处理它们。包括子依赖项。

    这也适用于根本没有参数的可调用对象。与没有参数的路径操作函数相同。
    """
    commons.public()
    return {"page": commons.page, "limit": commons.limit}


# =========================================================3.子依赖项=================================================


"""
FastAPI 支持创建含子依赖项的依赖项。

并且，可以按需声明任意深度的子依赖项嵌套层级。

FastAPI 负责处理解析不同深度的子依赖项。
"""


department_table = {
    "IT": ["Li", "Feng", "Ding"],
    "UI": ["Fang", "Zhang"]
}


def query_department(department: str):
    return department_table[department]


def query_user(username: str, department: List[str] = Depends(query_department)):
    if username not in department:
        return None

    return username


@case04.get("/son/depends/func", summary="子依赖项")
async def son_depends_func(user: Optional[str] = Depends(query_user)):
    """
    执行逻辑：<br>
    - 用户输入部门，和用户名
    - 根据用户输入的部门，查询出部门下有哪些员工
    - 再判断部门员工中有没有用户输入的用户名，有则返回此员工，无则返回空
    """
    return {"user": user}


@case04.get("/son/depends/func/no/cache", summary="子依赖项，不使用缓存值")
async def son_depends_func_no_cache(user: Optional[str] = Depends(query_user, use_cache=False)):
    """
    如果在同一个路径操作 多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项。

    FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。

    在高级使用场景中，如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False
    """
    return {"user": user}


# =========================================================4.路径操作装饰器依赖项=================================================


"""
有时，我们并不需要在路径操作函数中使用依赖项的返回值。

或者说，有些依赖项不返回值。

但仍要执行或解析该依赖项。

对于这种情况，不必在声明路径操作函数的参数时使用 Depends，而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list。
"""


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@case04.get("/path/depends/func", dependencies=[Depends(verify_token), Depends(verify_key)], summary="路径操作装饰器依赖项")
async def path_depends_func():
    """
    路径操作装饰器依赖项的执行或解析方式和普通依赖项一样，但就算这些依赖项会返回值，它们的值也不会传递给路径操作函数。

    路径装饰器依赖项与正常的依赖项一样，可以 raise 异常
    """
    return [{"item": "Foo"}, {"item": "Bar"}]


# =========================================================6.使用 yield 的依赖和子依赖====================================


"""
FastAPI 支持在完成后会执行一些额外步骤的依赖项。

为此，请使用 yield 而不是 return，然后再写一些额外的步骤。
"""


class DBSession(object):

    def close(self):
        pass


async def func_get_db():
    """
    例如，您可以使用它来创建数据库会话并在完成后关闭它。

    发送响应之前，仅执行 yield 语句之前的代码（包括该代码）：
    """
    db = DBSession()
    try:
        yield db    # 产生的值是注入到路径操作和其他依赖项中的值
    finally:
        db.close()  # yield响应传递后执行语句后面的代码


"""
在 yield 依赖项中同样可以创建任意深度的子依赖项嵌套层级。并且它们中的任何一个或全部都可以使用 yield。

FastAPI 将确保每个带有 yield 的依赖项中的『退出代码』以正确的顺序运行。

这与 Python 使用 yield 方式一致，不再过多讲解

您可以拥有所需的任何依赖项组合。

FastAPI将确保一切以正确的顺序运行。
"""


class MySuperContextManager:
    """
    Python 中，您可以通过 使用类中的两个方法：__enter__() 和 __exit__() 来创建上下文管理器。
    """
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def class_get_db():
    """
    在具有 yield 的依赖项中使用上下文管理器
    """
    with MySuperContextManager() as db:
        yield db
```

## 安全认证

```python
#!/usr/bin/python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/16 12:57 
# @File           : case05.py
# @IDE            : PyCharm
# @desc           : 案例05，安全认证


"""
安全性官方文档：https://fastapi.tiangolo.com/zh/tutorial/security/

FastAPI 安全认证默认使用 OAuth2 规范，它定义了几种处理身份认证和授权的方法（这里主要使用密码授权模式）。

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

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

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

并且这些工具已经自动地被集成到交互式文档系统中
"""


"""
1. OAuth2 认证示例
2. 基于 Password 和 Bearer token 的 OAuth2 认证
3. 使用（哈希）密码和 JWT Bearer 令牌的 OAuth2
"""


from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional


case05 = APIRouter()


@case05.get("/", summary="案例5初始接口")
def case05_root():
    return "case05_root"


"""
指明客户端用来请求 Token 的URL地址
使用 OAuth2PasswordBearer 需要安装 python-multipart，这是因为 OAuth2 使用 Form 表单发送 username 和 password。
OAuth2PasswordBearer 是接收URL参数的一个类，客户端会向该URL发送 username 和 password，然后得到一个 Token 值
OAuth2PasswordBearer 并不会创建响应的URL路径操作，只是指明了客户端用来请求 Token 的URL地址

tokenUrl URL地址前不会默认添加引入路由时的前缀 prefix，所以需要手动加上

当请求 OAuth2PasswordBearer 提供的URL地址时，FastAPI会检查请求头信息，如果没有找到Authorization头信息，或者头信息的内容不是 Bearer token，他会返回 401 未授权
由于在fastapi交互文档中不能修改请求头, 可以使用postman调试工具, 在postman中, 提供了一个专门的认证选项(Authorization), 提供了一些常见的认证可供选择
选择Bearer Token认证, 在右边会出现Token的输入框, 我们随便输入一个值, 点击发送, 可以看到返回的不再是401, 而是接口的正常返回值
"""
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/study/case05/token")


# ====================================1. OAuth2 认证====================================


@case05.get("/oauth2", summary="OAuth2 认证")
async def oauth2(token: str = Depends(oauth2_scheme)):
    """
    使用 OAuth2 认证依赖项后，在交互式接口文档中，您的路径操作在右上角有一个小锁，您可以单击它

    如果单击它，您会得到授权，可以输入 username 和 password（以及其他可选字段）

    如果未授权直接请求此接口，会返回 401 Not authenticated 未授权

    我们随便输入用户名密码后, 点击Authorize按钮, 发现它报错说Auth Error Error: Not Found, 这是因为我们还没有实现这个表单格式的登录接口

    也就是说这个参数tokenUrl只是为了方便fastapi的文档网页的认证使用, 在前后端实际项目中并不会起到作用
    """
    return {"token": token}


# ====================================2. 基于 Password 和 Bearer token 的 OAuth2 认证====================================


"""
OAuth2PasswordRequestForm 是一个类依赖项，声明了如下的请求表单：
- username。
- password。
- 一个可选的 scope 字段，是一个由空格分隔的字符串组成的大字符串。
- 一个可选的 grant_type.
- 一个可选的 client_id（我们的示例不需要它）。
- 一个可选的 client_secret（我们的示例不需要它）。

Tip：
OAuth2 规范实际上要求 grant_type 字段使用一个固定的值 password，但是 OAuth2PasswordRequestForm 没有作强制约束。
如果你需要强制要求这一点，请使用 OAuth2PasswordRequestFormStrict 而不是 OAuth2PasswordRequestForm。

Info：
OAuth2PasswordRequestForm 并不像 OAuth2PasswordBearer 一样是 FastAPI 的一个特殊的类。
OAuth2PasswordBearer 使得 FastAPI 明白它是一个安全方案。所以它得以通过这种方式添加到 OpenAPI 中。
但 OAuth2PasswordRequestForm 只是一个你可以自己编写的类依赖项，或者你也可以直接声明 Form 参数。
但是由于这是一种常见的使用场景，因此 FastAPI 出于简便直接提供了它。
"""


# 模仿数据库用户表
fake_users_db = {
    "admin": {
        "username": "admin",
        "full_name": "Admin",
        "email": "guest@example.com",
        "hashed_password": "fakehashedadmin",
        "disabled": False,
    },
    "guest": {
        "username": "guest",
        "full_name": "Guest",
        "email": "guest@example.com",
        "hashed_password": "fakehashedguest",
        "disabled": True,
    },
}


def fake_hash_password(password: str):
    """
    模仿 密码加密
    """
    return "fakehashed" + password


class User(BaseModel):
    """
    用户模型
    """
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    """
    输入模型
    """
    hashed_password: str


def get_user_info(db, username: str):
    """
    获取用户详细信息，如果不存在则返回 None
    """
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)  # UserInDB(**user_dict) 表示：直接将 user_dict 的键和值作为关键字参数传递


def fake_decode_token(token):
    """
    获取用户详细信息，如果不存在则返回 None
    这里目前没有提供任何安全性
    详见下一个版本
    """
    user = get_user_info(fake_users_db, token)
    return user


async def get_current_user(token: str = Depends(oauth2_scheme)):
    """
    获取当前登录的用户，没有则返回 401 无效的身份验证凭据

    我们在此处返回的值为 Bearer 的额外响应头 WWW-Authenticate 也是规范的一部分。

    任何的 401「未认证」HTTP（错误）状态码都应该返回 WWW-Authenticate 响应头。

    对于 bearer 令牌（我们的例子），该响应头的值应为 Bearer。

    实际上你可以忽略这个额外的响应头，不会有什么问题。

    但此处提供了它以符合规范。

    而且，（现在或将来）可能会有工具期望得到并使用它，然后对你或你的用户有用处。

    这就是遵循标准的好处...
    """
    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: User = Depends(get_current_user)):
    """
    判断当前登录用户是否可用，不可用则返回 400 非活动用户，可用则返回
    """
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户不可用")
    return current_user


@case05.post("/token", summary="用户认证")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """

    我们在上面使用 OAuth2PasswordBearer 提供的 tokenUrl 只是用于让 FastAPI 明白这是一个安全方案。所以它得以通过这种方式添加到 OpenAPI 中。

    而我们需要重写 tokenUrl 编写用户认证逻辑，路由路径需要与 tokenUrl URL地址一致，
    注意：tokenURL 是不会默认添加路由前缀的，而这里会默认添加路由前缀

    token 接口的响应必须是一个 JSON 对象。

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

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

    对于这个简单的示例，我们将极其不安全地返回相同的 username 作为令牌。
    """
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="用户名不存在")
    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="密码不正确")

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


@case05.get("/get/user/me", summary="获取当前登录的用户")
async def get_user_me(current_user: User = Depends(get_current_active_user)):
    """
    获取当前登录的用户

    注意：交互接口文档刷新后，登录后的状态会取消
    """
    return current_user


# ====================================2. 使用（哈希）密码和 JWT Bearer 令牌的 OAuth2====================================


"""
JWT 表示 「JSON Web Tokens」。https://jwt.io/

它是一个将 JSON 对象编码为密集且没有空格的长字符串的标准。

通过这种方式，你可以创建一个有效期为 1 周的令牌。然后当用户第二天使用令牌重新访问时，你知道该用户仍然处于登入状态。
一周后令牌将会过期，用户将不会通过认证，必须再次登录才能获得一个新令牌。

我们需要安装 python-jose 以在 Python 中生成和校验 JWT 令牌：pip install python-jose[cryptography]

PassLib 是一个用于处理哈希密码的很棒的 Python 包。它支持许多安全哈希算法以及配合算法使用的实用程序。推荐的算法是 「Bcrypt」：pip install passlib[bcrypt]
"""


"""安全的随机密钥，该密钥将用于对 JWT 令牌进行签名，生成随机密钥，linux下执行：openssl rand -hex 32"""
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
"""用于设定 JWT 令牌签名算法"""
ALGORITHM = "HS256"
"""令牌过期时间，30分钟"""
ACCESS_TOKEN_EXPIRE_MINUTES = 30


"""模仿数据库用户表"""
jwt_fake_users_db = {
    "admin": {
        "username": "admin",
        "full_name": "Admin",
        "email": "admin@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # 密码：secret
        "disabled": False,
    },
}


"""定义一个将在令牌端点中用于响应的 Pydantic 模型"""
class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Optional[str] = None


"""PassLib 上下文处理。这将用于哈希和校验密码。"""
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

"""指明客户端用来请求 JWT 的URL地址"""
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/study/case05/jwt/token")


def verify_password(plain_password: str, hashed_password: str):
    """验证密码"""
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    """密码加密"""
    return pwd_context.hash(password)


def jwt_get_user(db, username: str):
    """获取用户，不存在则然后 None"""
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    """验证用户密码"""
    user = jwt_get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """
    创建一个生成新的访问令牌的工具函数。
    """
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


@case05.post("/jwt/token", response_model=Token, summary="JWT 验证登录方式")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    重写 JWT 验证登录 的 tokenUrl

    JWT 验证登录流程：

    - 验证用户提交的用户名和密码，成功后获取到用户，不成功则返回 401 异常

    - 创建一个 timedelta 对象，用于设置令牌的过期时间

    - 创建一个真实的 JWT 访问令牌并返回它。

    信息：

    JWT 的规范中提到有一个 sub 键，值为该令牌的主题。

    使用它并不是必须的，但这是你放置用户标识的地方，所以我们在示例中使用了它。

    除了识别用户并允许他们直接在你的 API 上执行操作之外，JWT 还可以用于其他事情。

    例如，你可以识别一个 「汽车」 或 「博客文章」。

    然后你可以添加关于该实体的权限，比如「驾驶」（汽车）或「编辑」（博客）。

    然后，你可以将 JWT 令牌交给用户（或机器人），他们可以使用它来执行这些操作（驾驶汽车，或编辑博客文章），甚至不需要有一个账户，只需使用你的 API 为其生成的 JWT 令牌。

    使用这样的思路，JWT 可以用于更复杂的场景。

    在这些情况下，几个实体可能有相同的 ID，比如说 foo（一个用户 foo，一辆车 foo，一篇博客文章 foo）。

    因此，为了避免 ID 冲突，当为用户创建 JWT 令牌时，你可以在 sub 键的值前加上前缀，例如 username:。所以，在这个例子中，sub 的值可以是：username:johndoe。

    要记住的重点是，sub 键在整个应用程序中应该有一个唯一的标识符，而且应该是一个字符串。
    """
    user = authenticate_user(jwt_fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


async def jwt_get_current_user(token: str = Depends(oauth2_scheme)):
    """
    更新 get_current_user 以接收 JWT 令牌。

    解码接收到的令牌，对其进行校验，然后返回当前用户。

    如果令牌无效，立即返回一个 HTTP 错误。
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = jwt_get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)):
    """获取当前登录的用户，并判断用户是否可用，不可用则返回 400 用户不可用"""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户不可用")
    return current_user


@case05.get("/jwt/get/user/me", response_model=User)
async def jwt_get_user_me(current_user: User = Depends(jwt_get_current_active_user)):
    """获取当前登录的用户"""
    return current_user
```

## 后台任务

```python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/19 15:47
# @File           : case08.py
# @IDE            : PyCharm
# @desc           : 案例08，后台任务


"""
您可以定义要在返回响应后运行的后台任务。

这对于在请求之后需要进行的操作很有用，但是客户端实际上并不需要在收到响应之前就等待操作完成。

这包括，例如：
- 执行操作后发送的电子邮件通知：
    由于连接到电子邮件服务器并发送电子邮件的过程通常很慢（几秒钟），因此您可以立即返回响应并在后台发送电子邮件通知。

- 处理数据：
    例如，假设您收到的文件必须经过缓慢的处理，您可以返回『已接受』（HTTP 202）响应，并在后台对其进行处理。
"""
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, Depends

case08 = APIRouter()


@case08.get("/", summary="案例8初始接口")
def case08_root():
    return "case08_root"


def write_notification(email: str, message=""):
    with open("logs/notification.txt", mode="a") as email_file:
        content = f"发送通知到 {email}: {message}\n"
        email_file.write(content)


"""首先，导入 BackgroundTasks 并在路径操作函数中定义一个带有 BackgroundTasks 类型声明的参数"""
@case08.post("/send-notification/{email}", summary="后台任务")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    """在您的路径操作函数内部，使用 .add_task() 方法将您的任务函数传递给后台对象："""
    background_tasks.add_task(write_notification, email, message="发送内容")
    return {"message": "正在后台发送通知"}


# ====================================依赖注入=========================================================================


"""
使用 BackgroundTasks 还可以与依赖项注入系统一起使用，您可以在多个级别上声明 BackgroundTasks 类型的参数：
- 在路径操作函数中
- 在依赖项中（被依赖项）
- 在子依赖项中
- 等等。

FastAPI 知道每种情况下的操作以及如何重用同一对象，以便所有后台任务合并在一起并在后台运行

如果您需要执行大量的后台计算，而不必一定要在同一进程中运行它（例如，您不需要共享内存，变量等），则可能会受益于使用其他更大的工具，例如 Celery。
它们往往需要更复杂的配置，例如 RabbitMQ 或 Redis 之类的消息/作业队列管理器，但是它们允许您在多个进程（尤其是多个服务器）中运行后台任务。
但是，如果您需要从同一个 FastAPI 应用程序访问变量和对象，或者需要执行一些小的后台任务（例如发送电子邮件通知），则只需使用 BackgroundTasks 即可。
"""


def write_log(message: str):
    with open("logs/log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, desc: Optional[str] = None):
    if desc:
        desc = f"具体描述: {desc}\n"
        background_tasks.add_task(write_log, desc)
    return desc


@case08.get("/send-notification/{email}", summary="后台任务，依赖注入")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)):
    message = f"发送消息给 {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "消息正在发送中"}
```

## 数据库操作

## 中间件

## 测试

```python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Creaet Time    : 2021/10/19 15:47
# @File           : case09.py
# @IDE            : PyCharm
# @desc           : 案例09，接口测试


# 导入 TestClient
from fastapi.testclient import TestClient
# 导入 FastAPI 项目实例
from run import app

# 创建一个 TestClient 传递给您的 FastAPI
client = TestClient(app)


# 创建名称以 test_ 开头的函数（这是标准的 pytest 约定）
def test_case01_root():
    # 使用 TestClient 对象的方式与处理 requests 的方式相同
    response = client.get("/study/case01")
    assert response.status_code == 200
    assert response.json() == "case01_root"


"""
详细文档：
https://blog.csdn.net/xuqingskywolf/article/details/106630271

安装：
pip install pytest

项目根目录下运行：
pytest
"""
```

## 数据库迁移

FastAPI项目官网是直接使用的SqlAlchemy ORM，SqlAlchemy官方本身就是使用alembic实现表迁移的

[官网](https://alembic.sqlalchemy.org/en/latest/) https://alembic.sqlalchemy.org/en/latest/

博客：https://blog.51cto.com/u_15075523/4217138

### alembic

Alembic 使用 SQLAlchemy 作为底层引擎，为关系数据库提供变更管理脚本的创建、管理和调用。本教程将全面介绍该工具的理论和用法。

首先，确保按照 安装中的说明安装Alembic 。如链接文档中所述，通常最好将 Alembic 安装在与目标项目相同的模块/Python 路径中，通常使用 Python 虚拟环境，以便在alembic 运行命令时，调用的 Python 脚本by alembic，即您的项目env.py脚本，将可以访问您的应用程序模型。这并非在所有情况下都严格必要，但在绝大多数情况下通常是首选。

下面的教程假设alembic命令行实用程序存在于本地路径中，并且在调用时将可以访问与目标项目相同的 Python 模块环境。

先安装`alembic`

```shell
pip install alembic
```

然后生成初始化文件, 在项目外一层使用以下初始化命令

```shell
alembic init alembic
```

就会创建一个名为`alembic`的文件和一个`alembic.ini`文件
如下所示：

```
Copyalembic/
|____version/            // 一个空的文件夹，保存迁移中间文件                      
|____env.py              // 迁移配置文件 (主要设置文件)
|____script.py.mako                             
|____READEME             // 配置文件                                            
alembic.ini              // alembic 自动生成的迁移配置文件
app/                     // 项目文件夹
```

### 配置

配置`alembic/env.py`文件

```python
import os
import sys

# 当前项目路径
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

sys.path.insert(0, BASE_DIR)

from app.api.db.base import Base  # noqa

target_metadata = Base.metadata
```

## 事件

## 异步数据库

fastapi 异步数据库 和 非异步数据库的区别

## 数据库测试

## websocket