学习教程：
高性能 FastAPI 框架入门精讲（比B站的全）
或【独家新技术】从0到1学习 FastAPI 框架的所有知识点（B站，缺少第7章内容）

学习时间20220308-20220313
老师提供的教程代码

一个绝好的文档和教程: https://fastapi.tiangolo.com/zh/tutorial/

课程简介

体验新一代高性能 Python Web 框架，本课程将从 Hellow World 开始引导同学们学习 FastAPI 框架的所有知识点。从框架特性一览，到 ASGI 服务介绍，以全球新冠病毒感染数据查询为功能场景，依次讲解 FastAPI 的 API 交互文档使用，如何使用 Pydantic 定义和规范数据格式、类型，各种请求参数和验证，Jinja2 模板渲染和 Static 静态文件配置，FastAPI 的响应处理和配置，通过代码示例讲解依赖注入系统的所有知识，数据库配置与 SQLAlchemy 的使用，大型工程应该如何设计目录结构。 框架的安全、认证、授权，中间件开发，跨域资源共享的实现，后台任务和测试用例的编写。

第1章 课程介绍

介绍本课程的学习内容和目标，如何学习本课程，展示最终代码和效果，引导同学们对 FastAPI 框架有一个基本的了解，并能够在课程结束后独立使用 FastAPI 完成 RESTful API 接口开发

我的笔记：
前置知识：掌握Web开发基本知识、关系型数据库（MySQL/SQLite），熟练Python语法
课程代码/资料：https://github.com/liaogx/fastapi-tutorial
课程学习内容：FastAPI的所有核心模块
课程效果：完成查询页面和FastAPI的SwaggerUI


第2章 FastAPI介绍和项目准备

在学习 FastAPI 之前，先给大家介绍其相关的内容，如框架的优点，起源（与Pydantic/Starlette的关系），部署用到的服务。然后再讲项目开发环境的搭建，最后是Pydantic包用法的讲解。

FastAPI主要特点

  • 性能优越
  • 开发效率高，bug少
  • 代码风格直观
  • 易学易用
  • 代码精简，代码重复率低
  • 自带API交互文档，开发成果随时可以交付给前端对接
  • API开发标准化

Starlette、Pydantic、FastAPI的关系

Python3.6+的类型提示Type hints

定义形参数据类型

class Name:
    pass

def xxx(name: Name, age: int): 
    # 这里指定了形参的数据类型，调用时候必须传入对应的类型，指定的类型也可以是自定义类型
    print(name,age)



# 形参类型可以指定多层嵌套，比如定义形参必须是某种类型，且这个形参内的某个属性必须是某种类型...
from typing import List,Set,Tuple,Dict
def process_time(items: List[str], dic: Dict[str,float]):
    # 定义第一个形参的类型是列表，且列表内的元素必须是str类型
    # 定义第二个形参的类型是字典，且字典的键是str类型，值是float类型
    pass

Starlette、Pydantic、FastAPI的关系

Pydantic: 是一个基于Python类型提示（Type hints）来定义函数(API/接口)数据类型验证，序列化和文档(使用JSON模式)库
Starlette（翻译过来就是小星星）: 是一种基于Python3.6+的轻量级异步Web框架(ASGI 框架/工具包)，是构建高性能 Async io服务的理想选择，支持WebSocket、GrapQL、后台任务、启动或关闭时触发事件、
FastAPI主要使用了Pydantic库和Starlette框架及一些其他内容组成的。


ASGI 服务 Uvicorn 和 Hypercorn 介绍

Starlette是一个轻量级的ASGI框架，Django是WSGI框架，ASGI和WSGI都是一种规范，这些规范是基于服务器（Nginx或Apache）与 服务器应用（Django或Flask）之间的一种通讯约定，即这些规范是服务器与代理服务器的一种通讯协议，类似于http协议。

  • 基于ASGI服务: Uvicorn、Hypercorn、Daphne是可以通过pip安装的包，他们都属于ASGI规范的服务，有的文档会翻译成ASGI服务器，指的也是ASGI服务，相当于Nginx服务/Apache服务，只不过Uvicorn、Hypercorn、Daphne是通过pip安装Python包的方式去实现ASGI服务。Uvicorn、Hypercorn、Daphne这三种是Python异步Web框架部署的时候需要用到的服务，任选一个使用即可，使用FastAPI也从中任选一个即可，本课程选用的是Uvicorn（用的最多）
  • 基于WSGI服务: uWSGI、Gunicorn，这两个是部署Python同步Web应用（Django/Flask）使用的，uWSGI用的最多

搭建 FastAPI 项目开发环境

搭建Python运行虚拟环境

本课程使用到以下的包，保存到项目目录/requirements.txt：

aiofiles==0.6.0
atomicwrites==1.4.0
attrs==20.3.0
bcrypt==3.2.0
certifi==2020.12.5
cffi==1.14.4
chardet==4.0.0
click==7.1.2
colorama==0.4.4
cryptography==3.3.1
dnspython==2.0.0
ecdsa==0.14.1
email-validator==1.1.2
fastapi==0.63.0
h11==0.11.0
idna==2.10
importlib-metadata==3.3.0
iniconfig==1.1.1
Jinja2==2.11.2
MarkupSafe==1.1.1
packaging==20.8
passlib==1.7.4
pluggy==0.13.1
py==1.10.0
pyasn1==0.4.8
pycparser==2.20
pydantic==1.7.3
pyparsing==2.4.7
pytest==6.2.1
python-jose==3.2.0
python-multipart==0.0.5
requests==2.25.1
rsa==4.6
six==1.15.0
SQLAlchemy==1.3.22
starlette==0.13.6
toml==0.10.2
typing-extensions==3.7.4.3
urllib3==1.26.2
uvicorn==0.13.2
zipp==3.4.0
  1. 创建一个文件夹作为项目目录（比如d:/myFastApi），然后进入项目目录cd d:/myFastApi执行命令行virtualenv venv，执行这个命令默认使用自己系统已经装好的Python版本，执行命令后，会生成一个venv文件夹，这个文件夹就是虚拟环境的目录，如果激活了虚拟环境，那么安装的软件包就会安装在venv文件夹目录底下。注意：一定要使用Python3.6以上的版本。
  2. 激活虚拟环境：cd 项目目录/venv/Scripts，然后执行activate.bat(Windows)/activate(Linux)
  3. 安装指定的安装包pip install -r requirements.txt

Pydantic基础教程

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = "__Jack__"

from datetime import datetime, date
from pathlib import Path
from typing import List
from typing import Optional

from pydantic import BaseModel, ValidationError
from pydantic import constr
from sqlalchemy import Column, Integer, String
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.declarative import declarative_base

"""
Data validation and settings management using python type annotations.
使用Python的类型注解来进行数据校验和settings管理

pydantic enforces type hints at runtime, and provides user friendly errors when data is invalid.
Pydantic可以在代码运行时提供类型提示，数据校验失败时提供友好的错误提示

Define how data should be in pure, canonical python; validate it with pydantic.
定义数据应该如何在纯规范的Python代码中保存，并用Pydantic验证它
"""





print("\033[31m1. --- 知识点1：Pydantic的基本用法。Pycharm可以安装Pydantic插件 ---\033[0m") # 前面和后面加了"\033[31m"和"\033[0m"，Pycharm输出的是红色的字符串

class User(BaseModel):
    id: int  # 如果定义了一个字段但是没有给默认值，那么该字段就是必填字段。传给该字段的类型必须是int类型，或可以直接转成int类型的类型（比如'123'这种数字的字符串）
    name: str = "John Snow"  # 有默认值，选填字段，实例化时可以不给该字段赋值
    signup_ts: Optional[datetime] = None # 选填字段的第二种写法，使用Optional包起来，然后给一个默认值，这里的默认值是None
    friends: List[int] = []  # 要求列表中元素是int类型或者可以直接转换成int（比如'123'这种数字的字符串）类型

external_data = {
    "id": "123",
    "signup_ts": "2020-12-22 12:22",
    "friends": [1, 2, "3"],  # "3"是可以int("3")的
}
user = User(**external_data) # 字典前面使用**，是解包的意思，将字典的每个字段分解处理，当参传给User。同理也有打包的概念，比如形参用def xx(*args)，意思收集所有的位置参数到一个新的元组，并将这个元组赋值给变量args
print(user.id, user.friends)  # 实例化后调用属性
print(repr(user.signup_ts)) # repr可以将任意对象转为易读的字符串，比如传一个数据对象，就会返回一个时间字符串，类似于str()但比str()好用
print(user.dict()) # 将对象转为dict类型





print("\033[31m2. --- 知识点2：校验失败处理 ---\033[0m")
try:
    User(id=1, signup_ts=datetime.today(), friends=[1, 2, "not number"])
except ValidationError as e:
    print(e.json()) # 将错误内容JSON化





print("\033[31m3. --- 知识点3：模型类的的属性和方法 ---\033[0m")
print(user.dict()) # 对象转为dict
print(user.json()) # 对象转为json格式
print(user.copy()) # 这里是浅拷贝
print(User.parse_obj(external_data)) # 解析dict对象来实例化User，User.parse_obj(external_data)等同于上面的User(**external_data)
print(User.parse_raw('{"id": "123", "signup_ts": "2020-12-22 12:22", "friends": [1, 2, "3"]}')) # 也可以解析原生的数据，就是解析key:value类型的字符串来实例化User

# 解析文件中的内容来实例化User
path = Path('pydantic_tutorial.json') # 实例化文件，然后就可以对文件操作了
path.write_text('{"id": "123", "signup_ts": "2020-12-22 12:22", "friends": [1, 2, "3"]}') # 先将数据写入文件
print(User.parse_file(path)) # 解析文件
print(user.schema()) # 相比于user.dict()会输出更多更详细的信息。title是类名，数据在properties中，然后每个数据名称、类型、默认值都有详细说明
print(user.schema_json()) # # 相比于user.json()会输出更多更详细的信息

# 不校验数据类型实例化对象
user_data = {"id": "error", "signup_ts": "2020-12-22 12 22", "friends": [1, 2, 3]}  # id是字符串 是错误的
print(User.construct(**user_data))  # 解析数据并实例化对象，但不检验数据类型直接实例化User，不会报错，但是不符合类中的数据类型的定义的要求，这种方式不常用也不推荐

# 获得对象所有字段
print(User.__fields__.keys())  # 定义模型类的时候，只要给所有字段都注明类型，通过User.__fields__.keys()方式获得的字段顺序就和定义时候一样





print("\033[31m4. --- 知识点4：递归模型 ---\033[0m")

# 递归模型/嵌套模型：在一个类里调使用另一个类
class Sound(BaseModel):
    sound: str # key: value形式

class Dog(BaseModel):
    birthday: date
    weight: float = Optional[None] # 选填字段，实例化时可以不给该字段赋值，类型是float，默认值是None
    sound: List[Sound]  # 不同的狗有不同的叫声。递归模型（Recursive Models）就是指一个嵌套一个，List嵌套Sound，Sound嵌套str

dogs = Dog(birthday=date.today(), weight=6.66, sound=[{"sound": "wang wang ~"}, {"sound": "ying ying ~"}]) # {"sound": "ying ying ~"}可以直接赋值给Sound类，所以可以这么写
print(dogs.dict())





print("\033[31m5. --- 知识点5：ORM模型：从ORM类实例创建符合pydentic对象  ---\033[0m")
# 这里使用 sqlalchemy 和 pydantic 进行讲解
Base = declarative_base()
class CompanyOrm(Base):
    """
    公司类。该类是与数据库的表一一对应的模型类
    """
    __tablename__ = 'companies'
    id = Column(Integer, primary_key=True, nullable=False) # primary_key是不是主键字段，nullable是否能为空
    public_key = Column(String(20), index=True, nullable=False, unique=True) # index是否要给该字段建立索引，unique该字段是否唯一
    name = Column(String(63), unique=True)
    domains = Column(ARRAY(String(255))) # 一个公司可以有多个域名，所以域名字段需要放在一个元组的类型里，每个元组内的一个元素，就是一个字符串类型String(255)的域名

class CompanyModel(BaseModel):
    """
    使用 pydantic 的方法定义一个和上面CompanyOrm类对应的类（要求数据类型、字符串长度...等一致）
    """
    id: int
    public_key: constr(max_length=20) # contrict(限制/约束)字符串，使用constr来定义有限制长度的字符串
    name: constr(max_length=63)
    domains: List[constr(max_length=255)]

    class Config: # 这里要这么定义，且里面的变量要这么定义，才能使用  from_orm()
        orm_mode = True

# 使用 sqlalchemy 方式创建数据表模型类的实例 co_orm
co_orm = CompanyOrm(
    id=123,
    public_key='foobar',
    name='Testing',
    domains=['example.com', 'foobar.com'],
)

# 解析ORM对象的数据，实例化为符合pydantic规范的类（CompanyModel）
print(CompanyModel.from_orm(co_orm))

print("\033[31m6. --- Pydantic支撑的字段类型  ---\033[0m")  # 官方文档：https://pydantic-docs.helpmanual.io/usage/types/

第3章 请求参数和验证

本章先讲 FastAPI 程序的编码方式，它自带的 API 交互文档 Swagger UI，然后讲解 FastAPI 框架如何给后端传递参数，包括：路劲参数、参数参数、Body请求体、Cookie和Header参数。还有不同参数类型的组合使用。

FastAPI的hello world

hello_world.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()  # 这里的变量名不一定是app，随意自定义即可。如果这里变量名是app，下面也要用@app，也就是这里变量名如果是xxx，下面也得@xxx

class CityInfo(BaseModel):
    province: str
    country: str
    is_affected: Optional[bool] = None  # 是否有感染病例。选填字段，默认是None

# 如果想要哪个函数变成api供客户端调用，使用 @实例化对象.请求方式('url地址')的方式，将普通函数变成接口
# @app.get('/')
# def hello_world():
#     return {'hello': 'world'}
#
#
# @app.get('/city/{city}') # 使用{}包起来的就是传过来的参数，介于斜杠之间的参数称之为路径参数
# def result(city: str, query_string: Optional[str] = None):  # 形参的city就是上面{city}接收到的参数。在请求的url后面使用形如"?参数名=参数值"的方式比如"http://www.baidu.com/img?imgtype=jpg"接收传的参数叫查询参数，查询参数可以直接使用形参接收，这里的形参query_string就是查询参数的参数名。本函数写了2个形参接收参数，如果调用时候传的参数数量不对或者不对，就会报错，如果要将参数设置为选填参数，使用Optional为参数设定类型和默认值即可，比如query_string: Optional[str] = None。如果写成"query_string: str"，"query_string" 这个就不是选填参数了
#     return {'city': city, 'query_string': query_string}
# 测试请求：127.0.0.1:8000/city/BeiJing/query_string=aaa
# 查看接口文档：127.0.0.1:8000/doc
#
#
# @app.put('/city/{city}') # 浏览器不能直接发送put请求，访问接口文档（127.0.0.1:8000/doc）测试请求
# def result(city: str, city_info: CityInfo):
#     return {'city': city, 'country': city_info.country, 'is_affected': city_info.is_affected}
#

@app.get('/')
async def hello_world(): # 只要在函数前面加上async，就可以使得该函数变成异步函数了
    return {'hello': 'world'}


@app.get('/city/{city}')
async def result(city: str, query_string: Optional[str] = None):
    return {'city': city, 'query_string': query_string}


@app.put('/city/{city}')
async def result(city: str, city_info: CityInfo):
    return {'city': city, 'country': city_info.country, 'is_affected': city_info.is_affected}

# 启动命令(uvicorn py文件名:FastAPI实例的变量名)：uvicorn hello_world:app --reload # 如果使用--reload参数启动，可以实现只要代码有更改，就重启。

FastAPI 的 API 交互文档 - Swagger UI 和 ReDoc

Swagger UI使用详解

访问127.0.0.1:8000/docs使用Swagger UI

  • 请求

    1. 如果要给接口传递数据，要点一下Try itout
    2. 点击Excute就是发送请求
    3. name列就是字段名，请求字段名底部有"(path)"表示该字段就是路径参数，"(query)"就是查询参数

  • 响应

    1. 请求之后的Responses会有一个Curl，表示如果在Linux使用curl发送该请求，就复制这个输入框的内容到Linux执行即可
    2. 返回的第一个Responses是服务器返回给我们的数据。第二个Responses是我们自己对这个响应的数据进行描述，然后给前端工程师使用，类似于对响应数据的一个备注吧。
    3. 可以给Responses一个Example Value，即写一个例子给前端工程师参考，可以写明每个字段是什么含义，响应的数据结构是怎么样的
    4. Validation Error：错误校验。FastAPI有比较好的减少Bug是因为有这个数据校验

ReDoc使用详解

访问127.0.0.1:8000/redoc。不好用，使用Swagger UI就可以了。


其他内容

本小结包括以下内容

  1. 路径参数和数据的解析、验证
  2. 查询参数和数据的解析、验证
  3. 请求体以及混合参数
  4. 如何定义数据格式嵌套的请求体
  5. 设置 Cookie 和 Header 参数

本课程的项目目录：

C:.
│ coronavirus.sqlite3
│ hello_world.py
│ pydantic_tutorial.json
│ pydantic_tutorial.py
│ README.md
│ requirements.txt
│ run.py
│
├─coronavirus
│ │ city.json
│ │ crud.py
│ │ data.json
│ │ database.py
│ │ main.py
│ │ models.py
│ │ schemas.py
│ │ init.py
│ │
│ ├─static
│ │ │ chapter02.png
│ │ │ chapter03.png
│ │ │ chapter04.png
│ │ │ chapter05.png
│ │ │ chapter06.png
│ │ │ chapter07.png
│ │ │ chapter08.png
│ │ │ fastapi-tutorial.png
│ │ │ preview01.png
│ │ │ preview02.png
│ │ │ preview03.png
│ │ │ preview04.png
│ │ │ preview05.png
│ │ │ preview06.png
│ │ │ preview07.png
│ │ │ preview08.png
│ │ │ preview09.png
│ │ │ semantic.css
│ │ │ semantic.js
│ │ │ semantic.min.css
│ │ │ semantic.min.js
│ │ │
│ │ └─jquery-3.5.1
│ │ jquery-3.5.1.js
│ │ jquery-3.5.1.min.js
│ │ jquery-3.5.1.min.map
│ │ jquery-3.5.1.slim.js
│ │ jquery-3.5.1.slim.min.js
│ │ jquery-3.5.1.slim.min.map
│ │
│ └─templates
│ home.html
│
└─tutorial
chapter03.py
chapter04.py
chapter05.py
chapter06.py
chapter07.py
chapter08.py
test_chapter08.py
init.py

在Pychram中的项目根目录右键Python Package，然后创建名为coronavirus(用于疫情跟踪器)和tutorial(教程的知识点)python包(或者先创建普通文件夹，然后在创建的文件夹中创建文件__init__.py也是一样效果，也会让文件夹变为Python的一个包，比如需要将chapter04.py文件中的app04这个变量导入run.py，就在该__init__.py中写from .chapter04 import app04，然后再run.py中使用from tutorial import app04，就可以在主文件中使用app04了 )，然后在项目目录创建一个主文件run.py(执行这个主文件，这个项目就跑起来了)

run.py内容如下：

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

import time
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

from coronavirus import application

# 导入多个应用，然后下面再将这些应用合并到主应用中
from tutorial import app03, app04, app05, app06, app07, app08

# 导入错误处理的包
# from fastapi.exceptions import RequestValidationError
# from fastapi.responses import PlainTextResponse
# from starlette.exceptions import HTTPException as StarletteHTTPException


# FastAPI 应用的常见配置项
app = FastAPI(
    title='FastAPI Tutorial and Coronavirus Tracker API Docs', # Swagger UI的顶部的标题
    description='FastAPI教程 新冠病毒疫情跟踪器API接口文档，项目代码：https://github.com/liaogx/fastapi-tutorial',# Swagger UI的顶部的标题下方的描述
    version='1.0.0',
    docs_url='/docs', # 设置访问地址。http://127.0.0.1:8000/docs
    redoc_url='/redocs', # 设置redocs访问地址
)

# 静态文件的配置
# mount表示将某个目录下一个完全独立的应用挂载过来，这个不会在API交互文档中显示
app.mount(path='/static', app=StaticFiles(directory='./coronavirus/static'), name='static')  # .mount()不要在分路由APIRouter().mount()调用，模板会报错
# 参数详解
# path是路由的路径
# directory是真实的静态文件所在目录的路径
# name
# 注意：APIRouter()的实例.mount(...)是无法挂载成功的，只能挂载在主应用下面，也就是FastAPI()实例下面才能.mount(...)

# 重写HTTPException异常处理器
# @app.exception_handler(StarletteHTTPException)
# async def http_exception_handler(request, exc):
#     """
#     :param request: 这个参数不能省
#     :param exc:
#     :return:
#     """
#     return PlainTextResponse(str(exc.detail), status_code=exc.status_code) # 使用PlainTextResponse将原来响应的json格式的内容改为普通文本方式响应给前端，status_code还是使用原来的exc.status_code
#
#
# 重写请求验证异常处理器
# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     """
#     :param request: 这个参数不能省
#     :param exc:
#     :return:
#     """
#     return PlainTextResponse(str(exc), status_code=400)


# 给主应用定义一个中间件，拦截所有http请求，该请求中完成了计算一个接口业务逻辑
# 的请求过程的用时并响应给前端
@app.middleware('http')
async def add_process_time_header(request: Request, call_next):  # call_next将接收request请求做为参数
    start_time = time.time()
    response = await call_next(request) # 继续执行业务代码
    process_time = time.time() - start_time # 算出执行业务代码的时长
    response.headers['X-Process-Time'] = str(process_time)  # 添加自定义的以“X-”开头的请求头，这是自定义请求头的一个不成文的习惯
    return response

# 定义中间件用于解决跨域问题
app.add_middleware(
    CORSMiddleware,
    allow_origins=[ # 允许通过的域名
        "http://127.0.0.1",
        "http://127.0.0.1:8080"
    ],
    allow_credentials=True, # 允许使用证书
    allow_methods=["*"], # 允许通过的请求方法，比如post或get，也可以用通配符
    allow_headers=["*"], # 允许通过的请求头，也可以用通配符
)

# 给将子应用合并到主应用中来，并指定子应用的路由。比如主应用路由是/，那么子应用路由就是/chapter03，tags就是应用的标题，如果没设置tags，在Swagger UI中显示的应用名就是Default,设置了tags，就将tags显示为应用名
app.include_router(app03, prefix='/chapter03', tags=['第三章 请求参数和验证'])
app.include_router(app04, prefix='/chapter04', tags=['第四章 响应处理和FastAPI配置'])
app.include_router(app05, prefix='/chapter05', tags=['第五章 FastAPI的依赖注入系统'])
app.include_router(app06, prefix='/chapter06', tags=['第六章 安全、认证和授权'])
app.include_router(app07, prefix='/chapter07', tags=['第七章 FastAPI的数据库操作和多应用的目录结构设计'])
app.include_router(app08, prefix='/chapter08', tags=['第八章 中间件、CORS、后台任务、测试用例'])
app.include_router(application, prefix='/coronavirus', tags=['新冠病毒疫情跟踪器API'])

if __name__ == '__main__':
    uvicorn.run('run:app', host='0.0.0.0', port=8000, reload=True, debug=True, workers=1) # 使用这个方式就不用执行命令行uvicorn xxx.py的方式来启动FastAPI项目了。这行等同于 uvicorn hello_word:app --reload。
    # 参数说明
    # 'run:app':app就是当前文件实例化出来的FastAPI的变量名
    # host='0.0.0.0'就是在本机运行
    # reload就是代码发生变动后自动重启更新代码
    # workers就是进程的数量

chapter03.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from datetime import date
from enum import Enum
from typing import Optional, List

from fastapi import APIRouter, Query, Path, Body, Cookie, Header
from pydantic import BaseModel, Field

app03 = APIRouter()

"""Path Parameters and Number Validations 路径参数和数字验证"""


# 没有传参数
@app03.get("/path/parameters")
def path_params01():
    return {"message": "This is a message"}

# 传递路径参数。如果请求"http://127.0.0.1:8000/path/parameters"，那么会先从本文件从上到下匹配路由，因为请求的url既能匹配@app03.get("/path/parameters")也能匹配@app03.get("/path/{parameters}")，所以谁在上面就先匹配谁，于是匹配到了@app03.get("/path/parameters")然后执行了path_params01()，如果是@app03.get("/path/{parameters}")在@app03.get("/path/parameters")的前面，那么就执行然后执行了path_params02()
@app03.get("/path/{parameters}")  # 函数的参数顺序就是路由的顺序
def path_prams02(parameters: str):
    return {"message": parameters}


class CityName(str, Enum):
    Beijing = "BeijingChina"
    Shanghai = "ShanghaiChina"

# 接收枚举类型的参数。即传过来的参数只能是枚举类中定义好的数据
# 请求例子：http://127.0.0.1:8000/chapter03/enum/BeijingChina/
@app03.get("/enum/{city}")
async def latest(city: CityName):
    if city == CityName.Shanghai:# 对比提交的值是否和枚举中的对应的key的相等
        return {"city_name": city, "confirmed": 1492, "death": 7}
    if city == CityName.Beijing:
        return {"city_name": city, "confirmed": 971, "death": 9}
    return {"city_name": city, "latest": "unknown"}


# 通过path传递文件路径。file_path是传递参数的键名，path表示的是传过来的参数如果有"/"则不将斜杠当成路径的一部分，而是将斜杠作为参数的一部分（即作为字符串中的字符去理解）
@app03.get("/files/{file_path:path}")  
def filepath(file_path: str):
    return f"The file path is {file_path}"


@app03.get("/path_/{num}")
def path_params_validate(
    num: int = Path(..., title="Your Number", description="不可描述", ge=1, le=10),
):
    # fastapi包中的Path类就是校验【路径参数】使用的，“ge=1, le=10”在这里的意思是接收的num参数是int类型且必须要大于等于1，小于等于10。description的值是用于Swagger UI中对字段的描述的。这里的第一个参数传的"..."等同于传的None，也可以传None，只不过传了None使用Swagger UI测试就不好测试，因为这个参数类型已经给了是int类型，所以没法输入None，就传了"..."。fastapi包中的Path还可以定义别名，最大长度、最小长度、正则匹配等
    return num








"""Query Parameters and String Validations 查询参数和字符串验证"""

# 模拟提交分页查询，page页码，limit每页的数据量
@app03.get("/query")
def page_limit(page: int = 1, limit: Optional[int] = None):  # 给了默认值就是选填的参数，没给默认值就是必填参数
    if limit: # 如果limit不是None，则进入if
        return {"page": page, "limit": limit}
    return {"page": page}


# 如果接收的是bool类型，传的值是yes、on、1、True、true都会自动转为true，其他值都会转为false或报错
@app03.get("/query/bool/conversion")  # bool类型转换：yes on 1 True true会转换成true, 其它为false
def type_conversion(param: bool = False):
    return param


# 校验参数，即限制参数的类型，大小，长度等
# fastapi包中的Query就是校验【查询参数】使用的
@app03.get("/query/validations")  # 长度+正则表达式验证，比如长度8-16位，以a开头。其它校验方法看Query类的源码
def query_params_validate(
    value: str = Query(..., min_length=8, max_length=16, regex="^a"),  # ...换成None就变成选填的参数。regex="^a"正则匹配传来的值必须要以a开头
    values: List[str] = Query(["v1", "v2"], alias="alias_name") # Query的第一个参数是默认值，即默认值是["v1", "v2"]。别名就是请求时候传参的key，或Swagger UI中接口的提交字段名
):  # 多个查询参数的列表。参数别名
    return value, values








"""Request Body and Fields 请求体和字段"""

# 继承pydantic的BaseModel之后，可以定义请求体的数据格式和类型。在使用pydantic的类（pydantic的BaseModel的类）中要对字段进行校验就使用Field这个类
class CityInfo(BaseModel):
    name: str = Field(..., example="Beijing")  # example是给前端人员用Swagger UI时展示的该字段的一个例子
    country: str
    country_code: str = None  # 给一个默认值
    country_population: int = Field(default=800, title="人口数量", description="国家的人口数量", ge=800) # 设置默认值是800

    class Config:
        schema_extra = {
            "example": { # 定义前端请求时要提交的数据的一个例子，给前端在使用Swagger UI时参考
                "name": "Shanghai",
                "country": "China",
                "country_code": "CN",
                "country_population": 1400000000,
            }
        }


@app03.post("/request_body/city")
def city_info(city: CityInfo):
    print(city.name, city.country)
    return city.dict()








"""Request Body + Path parameters + Query parameters 多参数混合"""

# 这里使用【路径参数】、【请求体参数】、【查询参数】三种方式接收请求提交的信息
@app03.put("/request_body/city/{name}")
def mix_city_info(
    name: str, # 上面路径参数的name参数
    city01: CityInfo,
    city02: CityInfo,  # 请求体，请求体（Body）可以是多个的。当然多个请求体也可以是多个不同的类型不同结构的数据
    confirmed: int = Query(ge=0, description="确诊数", default=0),
    death: int = Query(ge=0, description="死亡数", default=0),
):
    if name == "Shanghai":
        return {"Shanghai": {"confirmed": confirmed, "death": death}}
    return city01.dict(), city02.dict()


@app03.put("/request_body/multiple/parameters")
def body_multiple_parameters(
    city: CityInfo = Body(..., embed=True),  # 当只有一个Body参数的时候，embed=True表示请求体参数嵌套。多个Body参数默认就是嵌套的
    confirmed: int = Query(ge=0, description="确诊数", default=0),
    death: int = Query(ge=0, description="死亡数", default=0),
):
    print(f"{city.name} 确诊数：{confirmed} 死亡数：{death}")
    return city.dict()








"""Request Body - Nested Models 数据类型嵌套的请求体"""


class Data(BaseModel):
    city: List[CityInfo] = None  # 这里就是定义数据格式嵌套的请求体。即Data类型使用了List，List又使用了另一种自定义类型。就是一个数据类型中嵌套了另一个类型的数据。
    date: date  # 额外的数据类型，还有uuid datetime bytes frozenset等，参考：https://fastapi.tiangolo.com/tutorial/extra-data-types/
    confirmed: int = Field(ge=0, description="确诊数", default=0) # 使用pydantic的Field对字段的数据进行校验
    deaths: int = Field(ge=0, description="死亡数", default=0)
    recovered: int = Field(ge=0, description="痊愈数", default=0)


@app03.put("/request_body/nested") # nested意思是嵌套
def nested_models(data: Data):
    return data








"""Cookie 和 Header 参数"""

@app03.get("/cookie")  # 效果只能用Postman测试。header中定义"Cookie": "cookie_id=xxx",cookie_id就是这里的后端接口的接收参数名称cookie_id
def cookie(cookie_id: Optional[str] = Cookie(None)):  # 定义接收cookie_id参数，类型是str，默认值是None。定义Cookie参数需要使用Cookie类，否则（比如不用Cookie类，直接写成cookie_id: Optional[str] = None）就是普通的获取【查询参数】。
    return {"cookie_id": cookie_id}


@app03.get("/header")
def header(
    user_agent: Optional[str] = Header(None, convert_underscores=True), # 接收Header参数，默认值为None。convert_underscores：是否转换下划线
    x_token: List[str] = Header(None)
):
    """
    有些HTTP代理和服务器是不允许在请求头中带有下划线的，所以Header提供convert_underscores属性让设置
    user_agent: convert_underscores=True 会把提交的 user_agent 变成 user-agent
    x_token: x_token是包含多个值的列表
    """
    return {"User-Agent": user_agent, "x_token": x_token}

第4章 响应处理和FastAPI配置

本章讲解 FastAPI 框架的响应 Response，包括响应模型类和状态码。还有表单数据处理、上传文件、路径操作配置。FastAPI 应用的配置，错误处理。

chapter04.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import Optional, List

from fastapi import APIRouter, status, Form, File, UploadFile, HTTPException
from pydantic import BaseModel, EmailStr

app04 = APIRouter()

"""Response Model 响应模型"""

# 定义请求体的数据结构
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    mobile: str = "10086"
    address: str = None
    full_name: Optional[str] = None

# 定义响应体的数据结构
class UserOut(BaseModel):
    username: str
    email: EmailStr  # 用 EmailStr 需要 pip install pydantic[email]，使用EmailStr可以自动校验Email的地址是否是合法的（校验提交的数据是否符合email规范）
    mobile: str = "10086"
    address: str = None
    full_name: Optional[str] = None


users = {
    "user01": {"username": "user01", "password": "123123", "email": "user01@example.com"},
    "user02": {"username": "user02", "password": "123456", "email": "user02@example.com", "mobile": "110"}
}

# response_model 在文档里叫做路径操作(path operation)，response_model规定响应体的格式
@app04.post("/response_model/", response_model=UserOut, response_model_exclude_unset=True)
async def response_model(user: UserIn):
    """response_model_exclude_unset=True表示默认值不包含在响应中，响应给前端的数据仅包含在接口的业务逻辑过程中实际赋的值，如果实际给的值与默认值相同也会包含在响应中。比如在上面的UserOut中mobil有默认值，如果在响应给前端之前没有给mobile赋值，那么它响应的数据就没有mobile这个字段（即不会把默认值响应给前端），retrun之前给mobile赋值了，无论赋的值是否和mobile的默认值相同，都会响应mobile这个字段给前端。response_model_exclude_unset=False，则会如果没有提供mobile值，就把默认值也相应回去，如果提供了mobile值就把提供的值响应回去"""
    print(user.password)  # password不会被返回
    # return user
    return users["user01"] # users的user01只有username、password、email。而响应体的格式是UserOut，UserOut没有password，所以只给前端响应数据了sername、email。
# 注意:如果响应的数据值是None，前端收到的就是null


@app04.post(
    "/response_model/attributes",
    response_model=UserOut,
    # response_model=Union[UserIn, UserOut], # 规定响应的数据结构是列表里的数据字段的并集
    # response_model=List[UserOut],
    response_model_include=["username", "email", "mobile"],  # 必须要响应的字段（即必须响应哪些字段）
    response_model_exclude=["mobile"] # 要排除响应的字段（即哪些字段不要响应）。如果定义了response_model_include中有mobile，response_model_exclude又排除了mobile字段，那么不会有冲突不会报错，最终结果是排除mobile字段。
)
async def response_model_attributes(user: UserIn):
    """response_model_include列出需要在返回结果中包含的字段；response_model_exclude列出需要在返回结果中排除的字段"""
    # del user.password  # Union[UserIn, UserOut]后，也可以删除 password 这个字段，这样前端就收不到 password 字段了
    return user
    # return [user, user] # response_model=List[UserOut]时用






"""Response Status Code 响应状态码"""

@app04.post("/my_status_code", status_code=200)
async def status_code():
    return {"my_status_code": 200}

# 响应状态码的另一种写法，使用status中的属性比如"status.HTTP_200_OK"来指定要响应的状态码
@app04.post("/status_attribute", status_code=status.HTTP_200_OK)
async def status_attribute():
    print(f"{type(status.HTTP_200_OK)}-{status.HTTP_200_OK}") # int-200
    return {"status_code": status.HTTP_200_OK}








"""Form Data 表单数据处理"""

@app04.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):  # 定义表单参数
    """用Form类需要pip install python-multipart; Form类的元数据和校验方法类似Body/Query/Path/Cookie，参考它们去使用就行了"""
    return {"username": username}








"""Request Files 单文件、多文件上传及参数详解"""

# 上传单个小文件（不推荐）
@app04.post("/file")
async def file_(file: bytes = File(...)):  # 如果要上传多个文件 files: file: List[bytes] = File(...)
    """使用File类 文件内容会以bytes的形式读入内存 适合于上传小文件，如果文件大的话内存会爆掉"""
    return {"file_size": len(file)} # len(file)可以获取file共有多少bytes


# 上传多个大文件
@app04.post("/upload_files")
async def upload_files(files: List[UploadFile] = File(...)):  # 如果要上传单个文件 file: UploadFile = File(...)
    """
    使用UploadFile类的优势:
    1.文件存储在内存中，使用的内存达到阈值后，将被保存在磁盘中
    2.适合于图片、视频大文件
    3.可以获取上传的文件的元数据（如文件名，创建时间等）
    4.有文件对象的异步接口
    5.上传的文件是Python文件对象，可以使用write(), read(), seek(), close()操作
    """
    for file in files:
        contents = await file.read()
        # 假设有两个异步函数async a，async b。a中的某一步有await，当程序碰到关键字await b()后，异步程序挂起后去执行另一个异步b程序，就是从函数内部跳出去执行其他函数，当挂起条件消失后，不管b是否执行完，要马上从b程序中跳出来，回到原程序执行原来的操作。如果await后面跟的b函数不是异步函数，那么操作就只能等b执行完再返回，无法在b执行的过程中返回。如果要在b执行完才返回，也就不需要用await关键字了，直接调用b函数就行。所以这就需要await后面跟的是异步函数了。在一个异步函数中，可以不止一次挂起，也就是可以用多个await。
        print(contents)
    return {"filename": files[0].filename, "content_type": files[0].content_type} # 返回第一个文件的文件名和文件类型。








"""【见run.py】FastAPI项目的静态文件配置"""







"""Path Operation Configuration 路径操作配置"""

@app04.post(
    "/path_operation_configuration",
    response_model=UserOut,
    # tags=["Path", "Operation", "Configuration"], # 和主文件app.py中的tags一样
    summary="This is summary", # 在Swagger UI 中，每条接口的url右侧的备注信息。如果没给这个参数，则默认都是首字母大写的接口所在的函数的名字
    description="This is description", # 在 Swagger UI 中，对本接口的描述
    response_description="This is response description", # 请求结果的描述（相当于对响应数据的一个说明内容）
    deprecated=True, # 此字段默认为False，设置成True则表示这个接口已废弃（Swagger UI 中会变成划横线并变成灰色），但是依然可以使用
    status_code=status.HTTP_200_OK
)
async def path_operation_configuration(user: UserIn):
    """
    Path Operation Configuration 路径操作配置
    :param user: 用户信息
    :return: 返回结果
    """
    return user.dict()






"""【见run.py】FastAPI 应用的常见配置项"""






"""Handling Errors 错误处理"""

@app04.get("/http_exception")
async def http_exception(city: str):
    if city != "Beijing":
        raise HTTPException(status_code=404, detail="City not found!", headers={"X-Error": "Error"}) # "detai"和"City not found!"就是响应给客户端的json的键值对。headers是要添加到响应头里面给客户端的数据键值对
    return {"city": city}


# 如果要重写异常处理器，则参考上面的主文件run.py的异常处理器
# 在主文件run.py重写了异常处理器后，调用这个接口，传参city_id=1时，抛出异常给前端，因为重写的异常处理器逻辑是将响应的json改为普通文本，所以响应体内容不再是{"detail": "Nope! I don't like 1."}，而是响应的普通文本:Nope! I don't like 1.
@app04.get("/http_exception/{city_id}")
async def override_http_exception(city_id: int):
    if city_id == 1:
        raise HTTPException(status_code=418, detail="Nope! I don't like 1.")
    return {"city_id": city_id}

第5章 FastAPI的依赖注入系统

本章专门讲解 FastAPI 的依赖注入系统，从概念、原理到开发套路。

“依赖注入”是指在编程中，为保证代码成功运行，先导入或声明其所需要的 “依赖”，如多个地方都会用到的同一个函数、数据库连接等。好处：

  1. 提高代码的复用率
  2. 共享数据库的连接
  3. 增强安全、认证和角色管理：比如一个网站需要注册才能查看某些文章，然后付费用户能看到的内容是依赖于注册用户，就是如果想看付费用户的内容，必须得先注册，然后管理员能看到的内容是依赖于注册用户，超级管理员是依赖于管理员
#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import Optional

from fastapi import APIRouter
from fastapi import Depends, HTTPException, Header

app05 = APIRouter()

"""Dependencies 创建、导入和声明依赖"""

# 这里定义一个函数作为依赖，供下面的接口去使用
async def common_parameters(q: Optional[str] = None, page: int = 1, limit: int = 100):
    return {"q": q, "page": page, "limit": limit}



# 上面定义的def common_parameters函数相当于创建了一个依赖common_parameters，在这里使用Depends(common_parameters)来导入上面的common_parameters依赖。
# 这里的注入依赖不区分同步异步，可以在同步接口调用异步的依赖，也可以异步接口调用同步依赖
@app05.get("/dependency01")
async def dependency01(commons: dict = Depends(common_parameters)): # 这里使用dict是因为依赖common_parameters返回的是字典
    return commons
# commons: dict = Depends(common_parameters)表示对接口的依赖进行了一个声明，表示的是接口参数请求依赖于common_parameters的函数。当接口被调用的时候，会调用 common_parameters 函数进行请求处理。相当于把 common_parameters 当做一个回调函数传给接口，回调函数的参数就是接口可以接收的参数。


# 这里的函数和上面的区别是没有 async 修饰，是个同步的方法，调用异步的依赖也没有问题
@app05.get("/dependency02")
def dependency02(commons: dict = Depends(common_parameters)):  # 可以在async def中调用def依赖，也可以在def中导入async def依赖
    return commons







"""Classes as Dependencies 类作为依赖项"""

# 这里定义一个列表，模拟从数据库读取出来的数据
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, page: int = 1, limit: int = 100):
        self.q = q
        self.page = page
        self.limit = limit

# 将类作为依赖注入接口参数有3种写法，下面注释的2种也是。3种写法作用一模一样
@app05.get("/classes_as_dependencies")
# async def classes_as_dependencies(commons: CommonQueryParams = Depends(CommonQueryParams)):
# async def classes_as_dependencies(commons: CommonQueryParams = Depends()):
async def classes_as_dependencies(commons=Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.page: commons.page + commons.limit] # 使用切片方式，截取列表fake_items_db，模拟分页查询数据库的数据
    response.update({"items": items})
    return response








"""Sub-dependencies 子依赖"""
def query(q: Optional[str] = None):
    return q


# sub_query的q依赖于上面的query
def sub_query(q: str = Depends(query), last_query: Optional[str] = None):
    if not q:
        return last_query
    return q


@app05.get("/sub_dependency")
async def sub_dependency(final_query: str = Depends(sub_query, use_cache=True)):
    """use_cache默认是True, 表示当多个依赖有一个共同的子依赖时，每次request请求只会调用子依赖一次，多次调用将从缓存中获取"""
    return {"sub_dependency": final_query}








"""Dependencies in path operation decorators 路径操作装饰器中的多依赖"""

# 定义没有返回值的子依赖
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


# 在路径操作中调用多个不同的依赖：就是获取【路径参数】并将参数传给依赖的函数去处理。这种方式就算依赖有返回值，该返回值也不会传给接口函数。
@app05.get("/dependency_in_path_operation", dependencies=[Depends(verify_token), Depends(verify_key)])
async def dependency_in_path_operation():
    return [{"user": "user01"}, {"user": "user02"}]










"""Global Dependencies 全局依赖"""
# 比如给所有app05的所有接口添加校验token和校验key的依赖，那就定义verify_token函数去校验token，verify_key函数校验key，然后将这些函数作为依赖给app05去使用就行了，app05的每个接口都能校验校验token和key了。如果要将依赖作为整个FastAPI项目的接口使用，那就将依赖注入到主应用中去，也就是在run.py中给FastAPI(....dependencies=[Depends(verify_token), Depends(verify_key))

# app05 = APIRouter(dependencies=[Depends(verify_token), Depends(verify_key)])









"""Dependencies with yield 带yield的依赖"""

# 这个需要Python3.7才支持，Python3.6需要pip install async-exit-stack async-generator

# 以下都是伪代码（下面的代码没看懂）
async def get_db():
    db = 获取数据库连接
    try:
        yield db
    finally:
        db.关闭连接


async def dependency_a():
    dep_a = "generate_dep_a()"
    try:
        yield dep_a
    finally:
        dep_a.endswith("db_close")


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = "generate_dep_b()"
    try:
        yield dep_b
    finally:
        dep_b.endswith(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = "generate_dep_c()"
    try:
        yield dep_c
    finally:
        dep_c.endswith(dep_b)

第6章 安全、认证和授权

本章会先图文讲解 OAuth 2.0 的概念和原理，然后 FastAPI 框架中实现 OAuth 2.0 的密码模式认证和 JWT 认证。

OAuth2.0的授权模式：

  1. 授权码授权模式（Authorization Code Grant）
  2. 隐式授权模式（Implicit Grant）
  3. 密码授权模式（Resource Owner Password Credentials Grant）
    1. 用户输入用户名和密码，然后提交用户名和密码到授权服务器（就是可以校验用户名和密码的服务器）
    2. 服务器校验用户名和密码通过以后，返回token给客户端
    3. 客户端拿到token
    4. 此后的每次请求资源或接口，客户都携带该token进行请求，就可以通过验证了
  4. 客户端凭证授权模式（Client Credentials Grant）

本教程只讲解密码授权模式:

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from datetime import datetime, timedelta
from typing import Optional
from pydantic import BaseModel
from fastapi import APIRouter, Depends, HTTPException, status

# 密码授权模式需要用到这两个包
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

# JWT加密要使用的包
from jose import JWTError, jwt
from passlib.context import CryptContext


app06 = APIRouter()

"""OAuth2 密码模式和 FastAPI 的 OAuth2PasswordBearer"""

"""
OAuth2PasswordBearer是接收URL作为参数的一个类：客户端会向该URL发送username和password参数，然后得到一个Token值
OAuth2PasswordBearer并不会创建相应的URL路径操作，只是指明客户端用来请求Token的URL地址
当请求到来的时候，FastAPI会检查请求的Authorization头信息，如果没有找到Authorization头信息，或者头信息的内容不是Bearer token，它会返回401状态码(UNAUTHORIZED)
"""

oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/token")  # 定义请求Token的URL地址 http://127.0.0.1:8000/chapter06/token

# 引用oauth2_schema作为依赖，然后把从OAuth2PasswordBearer获取到的token直接返回。就是如果客户端请求用户名和密码过来，就给客户端返回一个token
@app06.get("/oauth2_password_bearer")
async def oauth2_password_bearer(token: str = Depends(oauth2_schema)):
    return {"token": token}


"""基于 Password 和 Bearer token 的 OAuth2 认证"""

# 模拟数据库的2条用户的数据
fake_users_db = {
    "john snow": {
        "username": "john snow",
        "full_name": "John Snow",
        "email": "johnsnow@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False, # 目前该用户是否已经激活，激活的用户和未激活用户的权限不一致，可以模拟角色管理
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "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

# 登录，如果登录成功，就返回用户名。否则抛出异常
@app06.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=status.HTTP_400_BAD_REQUEST, 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=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    return {"access_token": user.username, "token_type": "bearer"}

# 获取所有用户
def get_user(db, username: str):
    if username in db: # 如果在db这个字典中存在key==username
        user_dict = db[username]
        return UserInDB(**user_dict)

# 模拟解码校验token
def fake_decode_token(token: str):
    user = get_user(fake_users_db, token) # 这行的token就是上面一行的形参token
    return user

# 获取当前用户
async def get_current_user(token: str = Depends(oauth2_schema)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},  # 如果认证失败，请求头中返回“WWW-Authenticate”。加不加headers参数不强制。但使用该参数是OAuth2的规范，建议使用。
        )
    return user

# 获取激活的用户
async def get_current_active_user(current_user: User = Depends(get_current_user)):
    # 如果是未激活用户，则抛出异常
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user


# 返回当前登录用户的信息。只有当前用户是活跃用户时才返回当前用户，否则会在依赖get_current_active_user中抛出错误。需要先登录，获取token后再调用本接口，才能获取到用户信息
@app06.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user















"""OAuth2 with Password (and hashing), Bearer with JWT tokens 开发基于JSON Web Tokens的认证"""
# JWT(JSON Web Tokens)的验证流程:
# 1. 浏览器给服务器发送用户名和密码
# 2. 服务器接收到用户名以后会创建一个JWT并返回给浏览器
# 3. 浏览器接收到JWT后，之后每次向服务器发送请求的认证请求头都会带上这个JWT
# 4. 每次接收到浏览器的请求时，都会验证浏览器请求头里携带的JWT的签名，然后从JWT里面获取到用户的信息，再把数据响应给浏览器
# 相当于登录时使用明文用户名和密码，如果正确就生成一个JWT作为token，它里面还加密了一些用户信息比如用户名，下次登录时带上该token就可以校验该token是否有效，且如果有效，下一步就使用该token解析出里面携带的信息比如用户名，然后直接根据用户名获取用户数据。JW

# 从上面流程可以看出，基于JSON Web Tokens的认证需要以下功能：
# 1. 需要有一个接口可以接收用户名和密码
# 2. 需要有一个可以创建JWT的函数
# 3. 还要有对请求头里的JWT认证信息校验的功能，包括校验JWT的签名和解析用户信息
fake_users_db.update({
    "john snow": {
        "username": "john snow",
        "full_name": "John Snow",
        "email": "johnsnow@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
})

 # 在linux中生成密钥的命令：openssl rand -hex 32。在Windows中可以手动自定义一个字符串也行
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" # 密钥
ALGORITHM = "HS256"  # 生成JWT时的加密的算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 设定JWT令牌过期时间(分钟)


# 定义后端要返回给前端的JWT数据格式
class Token(BaseModel):
    """返回给用户的Token"""
    access_token: str
    token_type: str

# 调用CryptContext，对用户传过来的密码进行加密，这里使用bcrypt加密算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 定义获取token的路径的依赖，下面要用。可以参考上面的oauth2_schema也是一样的作用。
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/jwt/token")


# 对密码进行校验：plain_password-明文密码，hashed_password-加密后的密码。返回True或False
def verity_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

# 获取用户
def jwt_get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

# 校验用户并获取用户信息
def jwt_authenticate_user(db, username: str, password: str):
    user = jwt_get_user(db=db, username=username)
    if not user or :
        return False
    if not verity_password(plain_password=password, hashed_password=user.hashed_password):
        return False
    return user

# 创建token。data是个dict，该dict里面有两个键值对，一个是用户名，一个是token过期时间，这里的过期时间使用上面定义的变量ACCESS_TOKEN_EXPIRE_MINUTES，该过期时间可以更改，如果传了过期时间expires_delta，则以过期时间为准
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy() # 浅拷贝，拷贝第一层级数据，更深层级的数据还是引用原来的对象
    # 如果传入了过期时间，就使用当前时间+传入的过期时间作为token的有效时间，否则使用当前时间+默认时间（这里为15分钟）token的有效时间
    expire = datetime.utcnow() + (expires_delta if expires_delta else timedelta(minutes=15))
    to_encode.update({"exp": expire}) # 更新token的过期时间(有效时间)
    encoded_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM) # 生成JWT。claims：要进行编码的字符，algorithm：生成JWT的算法
    return encoded_jwt

# 定义接口接收前端传过来的用户名和密码（也就是登录接口），response_model声明了要返回的数据结构
@app06.post("/jwt/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # 校验并获取用户
    user = jwt_authenticate_user(db=fake_users_db, username=form_data.username, password=form_data.password)
    if not user: # 如果用户不存在或者密码不正确，则返回401状态码
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            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
    )  # 如果登录成功就更新用户的token过期时间，并加入一些数据比如用户名之类的信息，生成token。然后将token返回前端。data的key是sub，这个key是这里使用的jwt的默认key，用sub作为key就行了
    return {"access_token": access_token, "token_type": "bearer"}


# 使用jwt的方式获取当前用户
async def jwt_get_current_user(token: str = Depends(oauth2_schema)):
    # 定义一个异常，下面再用
    credentials_exception = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM]) # 获取token，并解析token，这里的key的值和algorithms的值需要和 login_for_access_token 中生成token时用的一致才能解析成功。
        username = payload.get("sub") # 获取到登录时（login_for_access_token()）生成token时放入的数据，上面的登录接口写的是放入了用户名
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = jwt_get_user(db=fake_users_db, username=username) # 获取用户信息
    if user is None:
        raise credentials_exception
    return user # 返回用户的信息


# 使用jwt的方式获取当前已激活的用户
async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user

# 使用jwt的方式获取当前登录的用户信息
@app06.get("/jwt/users/me")
async def jwt_read_users_me(current_user: User = Depends(jwt_get_current_active_user)):
    return current_user

第7章 FastAPI的数据库操作和多应用的目录结构设计

这章代码量较多，主要是开发 Coronavirus 应用，实现课程开头部分大家看到的效果。过程中会讲解到数据库的配置，SQLAlchemy ORM的开发，Jinja2模板的开发。最后还有一个关于工程知识点，项目目录有结构的优化。

本章节会用到以下文件

  1. city.json
  2. crud.py: 对数据库的操作封装在这里，比如增删查改
  3. data.json
  4. database.py：如何在FastAPI项目中配置使用数据库，这里使用的是 SQLite
  5. main.py: 业务接口，供前端调用的接口都写在这里
  6. models.py: 与数据库表字段一一对应的模型类
  7. schemas.py: 响应体的数据规范，基本和models.py中的模型类一一对应

database.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 这里使用 SQLite
SQLALCHEMY_DATABASE_URL = 'sqlite:///./coronavirus.sqlite3'
# SQLALCHEMY_DATABASE_URL = "postgresql://username:password@host:port/database_name"  # MySQL或PostgreSQL的连接方法

engine = create_engine(
    # echo=True表示引擎将用repr()函数记录所有语句及其参数列表到日志
    # 由于SQLAlchemy是多线程，指定check_same_thread=False来让建立的对象任意线程都可使用。这个参数只在用SQLite数据库时设置
    SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True, connect_args={'check_same_thread': False}
)
# echo设置为True，在执行SQL语句时候会将这些SQL语句输出到终端中，否则静默不输出
# 因为sqlalchemy是多线程的，设置了'check_same_thread': False之后，那么任意线程都可以使用sqlalchemy创建的对象（此处也就是任意线程都可以使用这里创建出来的engine）


# 在SQLAlchemy中，CRUD都是通过会话(session)进行的，所以我们必须要先创建会话，每一个SessionLocal实例就是一个数据库session
# flush()是指发送数据库语句到数据库，但还没有真正执行并持久化；commit()是指提交事务，将执行SQL语句变更保存到数据库完成持久化
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=True)

# 创建基本映射类
Base = declarative_base(bind=engine, name='Base')

models.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from sqlalchemy import Column, String, Integer, BigInteger, Date, DateTime, ForeignKey, func
from sqlalchemy.orm import relationship

from .database import Base


class City(Base):
    __tablename__ = 'city'  # 数据表的表名
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    province = Column(String(100), unique=True, nullable=False, comment='省/直辖市')
    country = Column(String(100), nullable=False, comment='国家')
    country_code = Column(String(100), nullable=False, comment='国家代码')
    country_population = Column(BigInteger, nullable=False, comment='国家人口')
    data = relationship('Data', back_populates='city')  # 'Data'是关联的类名，Data类定义在下面，往下看；back_populates来指定反向访问的属性名称，可以通过父表查询到子表
    created_at = Column(DateTime, server_default=func.now(), comment='创建时间')
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='更新时间') # 当数据更新的时候，这个字段也会自动更新

    __mapper_args__ = {"order_by": country_code}  # 当获取这个表的数据是，指定排序的方式的字段。默认是正序，倒序就写country_code.desc()

    def __repr__(self): # __repr__函数相当于java中类里面的toString()
        return f'{self.country}_{self.province}'


class Data(Base):
    __tablename__ = 'data'
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    city_id = Column(Integer, ForeignKey('city.id'), comment='所属省/直辖市')  # 外键字段，参数是外接的表名.字段名，这里的city是上面City类的表名，id是表中的要连接的
    date = Column(Date, nullable=False, comment='数据日期')
    confirmed = Column(BigInteger, default=0, nullable=False, comment='确诊数量')
    deaths = Column(BigInteger, default=0, nullable=False, comment='死亡数量')
    recovered = Column(BigInteger, default=0, nullable=False, comment='痊愈数量')
    city = relationship('City', back_populates='data')  # 'City'是关联的类名；back_populates来指定反向访问的属性名称

    created_at = Column(DateTime, server_default=func.now(), comment='创建时间')
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='更新时间')

    __mapper_args__ = {"order_by": date.desc()}  # 按日期倒序排序

    def __repr__(self):
        return f'{repr(self.date)}：确诊{self.confirmed}例'


""" 附上三个SQLAlchemy教程

SQLAlchemy的基本操作大全 
    http://www.taodudu.cc/news/show-175725.html

Python3+SQLAlchemy+Sqlite3实现ORM教程 
    https://www.cnblogs.com/jiangxiaobo/p/12350561.html

SQLAlchemy基础知识 Autoflush和Autocommit
    https://zhuanlan.zhihu.com/p/48994990
"""

schemas.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from datetime import date as date_
from datetime import datetime

from pydantic import BaseModel


class CreateData(BaseModel):
    date: date_
    confirmed: int = 0
    deaths: int = 0
    recovered: int = 0


class CreateCity(BaseModel):
    province: str
    country: str
    country_code: str
    country_population: int


class ReadData(CreateData):
    id: int
    city_id: int
    updated_at: datetime
    created_at: datetime

    class Config: # 设置Config内部类并设置属性orm_mode = True是为了方便使用 from_orm 直接将ORM类中的数据转为当前ReadData类的数据格式，然后返回。如果不理解，可以搜前面的 from_orm 再学习一下
        orm_mode = True


class ReadCity(CreateCity):
    id: int
    updated_at: datetime
    created_at: datetime

    class Config:
        orm_mode = True

crud.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from sqlalchemy.orm import Session

from coronavirus import models, schemas

# 通过城市的id查询城市的数据
def get_city(db: Session, city_id: int):
    return db.query(models.City).filter(models.City.id == city_id).first()

# 通过城市的名称查询城市的数据
def get_city_by_name(db: Session, name: str):
    return db.query(models.City).filter(models.City.province == name).first()

# 分页查询多个城市的数据
def get_cities(db: Session, skip: int = 0, limit: int = 10):
    return db.query(models.City).offset(skip).limit(limit).all()

# 创建城市数据并持久化。限制提交的数据是schemas.CreateCity这种数据结构
def create_city(db: Session, city: schemas.CreateCity):
    db_city = models.City(**city.dict())
    db.add(db_city)
    db.commit()
    db.refresh(db_city)
    return db_city

# 获取新冠感染的数据(data表)，如果传了城市名称，就分页查询该城市的数据，否则分页查询所有城市的数据
def get_data(db: Session, city: str = None, skip: int = 0, limit: int = 10):
    if city:
        return db.query(models.Data).filter(models.Data.city.has(province=city))  # 外键关联查询，这里不是像Django ORM那样使用对象.属性.属性(Data.city.province)
    return db.query(models.Data).offset(skip).limit(limit).all() # 如果没有传递城市名称，则查询所有城市，按分页查询

# 创建城市感染的数据
def create_city_data(db: Session, data: schemas.CreateData, city_id: int):
    db_data = models.Data(**data.dict(), city_id=city_id)
    db.add(db_data)
    db.commit()
    db.refresh(db_data)
    return db_data

main.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import List

import requests
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Request
from fastapi.templating import Jinja2Templates
from pydantic import HttpUrl
from sqlalchemy.orm import Session

from coronavirus import crud, schemas
from coronavirus.database import engine, Base, SessionLocal
from coronavirus.models import City, Data

application = APIRouter()

# 设置Jinja2模板路径
templates = Jinja2Templates(directory='./coronavirus/templates')

Base.metadata.create_all(bind=engine) # 创建数据库的表/创建表

# 创建数据库连接的依赖，供下面的其他接口使用
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 创建城市信息，如果城市已存在就报错，否则创建城市信息到city表
@application.post("/create_city", response_model=schemas.ReadCity)
def create_city(city: schemas.CreateCity, db: Session = Depends(get_db)):
    db_city = crud.get_city_by_name(db, name=city.province)
    if db_city:
        raise HTTPException(status_code=400, detail="City already registered")
    return crud.create_city(db=db, city=city)


# 读取城市信息（city表）
@application.get("/get_city/{city}", response_model=schemas.ReadCity)
def get_city(city: str, db: Session = Depends(get_db)):
    db_city = crud.get_city_by_name(db, name=city)
    if db_city is None:
        raise HTTPException(status_code=404, detail="City not found")
    return db_city

# 获取多个城市信息（分页查询city表）
@application.get("/get_cities", response_model=List[schemas.ReadCity])
def get_cities(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    cities = crud.get_cities(db, skip=skip, limit=limit)
    return cities

# 创建城市对应的新冠感染的数据
@application.post("/create_data", response_model=schemas.ReadData)
def create_data_for_city(city: str, data: schemas.CreateData, db: Session = Depends(get_db)):
    db_city = crud.get_city_by_name(db=db, name=city)
    data = crud.create_city_data(db=db, data=data, city_id=db_city.id)
    return data

# 获取新冠感染的数据，如果指定了是哪个城市，就查询该城市的新冠感染的数据
@application.get("/get_data")
def get_data(city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    data = crud.get_data(db, city=city, skip=skip, limit=limit)
    return data

# 后台任务，本项目是用于后台将其他平台的数据同步过来。将该函数定义好后，下面再使用。
# 后台任务的其他使用场景：注册时候给用户发邮件时，会将发邮件这个任务添加到后台任务
# 中去执行。免得如果用户等待太久这是不合理的
def bg_task(url: HttpUrl, db: Session):
    """这里注意一个坑，不要在后台任务的参数中db: Session = Depends(get_db)这样导入依赖会导入失败"""
    city_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=false")
    if 200 == city_data.status_code:
        db.query(City).delete()  # 同步数据前先清空原有的数据
        for location in city_data.json()["locations"]:
            city = {
                "province": location["province"],
                "country": location["country"],
                "country_code": "CN",
                "country_population": location["country_population"]
            }
            crud.create_city(db=db, city=schemas.CreateCity(**city))

    coronavirus_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=true")
    if 200 == coronavirus_data.status_code:
        db.query(Data).delete()
        for city in coronavirus_data.json()["locations"]:
            db_city = crud.get_city_by_name(db=db, name=city["province"])
            for date, confirmed in city["timelines"]["confirmed"]["timeline"].items():
                data = {
                    "date": date.split("T")[0],  # 把'2020-12-31T00:00:00Z' 变成 ‘2020-12-31’
                    "confirmed": confirmed,
                    "deaths": city["timelines"]["deaths"]["timeline"][date],
                    "recovered": 0  # 每个城市每天有多少人痊愈，这种数据没有
                }
                # 这个city_id是city表中的主键ID，不是coronavirus_data数据里的ID
                crud.create_city_data(db=db, data=schemas.CreateData(**data), city_id=db_city.id)

# 接收本请求后，开启后台任务，拉取数据（同步数据）
@application.get("/sync_coronavirus_data/jhu")
def sync_coronavirus_data(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    """从Johns Hopkins University同步COVID-19数据"""
    background_tasks.add_task(bg_task, "https://coronavirus-tracker-api.herokuapp.com/v2/locations", db)
    return {"message": "正在后台同步数据..."}


# 获取城市数据（后端渲染再返回给前端，不是前后端分离）
@application.get("/")
def coronavirus(request: Request, city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    data = crud.get_data(db, city=city, skip=skip, limit=limit)
    return templates.TemplateResponse("home.html", {
        "request": request, # 固定写法
        "data": data, # 将数据传到模板里去使用
        "sync_data_url": "/coronavirus/sync_coronavirus_data/jhu"
    })

前端home.html：略


第8章 中间件、CORS、后台任务、测试用例

这章把4个零散（中间件、跨域资源共享CORS、后台任务、测试用例）的知识点放在一起，都分别实现一些的简单的示例，便于大家理解。后台任务的部分，会把 Coronavirus 应用中的后台同步数据功能完成。

中间件的开发

Snipaste_2022-03-10_15-07-12

如果要拦截所有请求，就把中间件写在主应用(run.py)中，参考run.py内容中的给主应用添加中间件，然后在中间件中计算执行业务代码时长，然后添加到响应头中返回给前端。

跨域资源共享CORS的问题是请求是否符合同源策略的问题，解决方法写在了run.py中，使用中间件处理即可，在run.py中的内容如下：

# 定义中间件用于解决跨域问题
app.add_middleware(
    CORSMiddleware,
    allow_origins=[ # 允许通过的域名
        "http://127.0.0.1",
        "http://127.0.0.1:8080"
    ],
    allow_credentials=True, # 允许使用证书
    allow_methods=["*"], # 允许通过的请求方法，比如post或get，也可以用通配符
    allow_headers=["*"], # 允许通过的请求头，也可以用通配符
)

后台任务Demo：chapter08.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import Optional

from fastapi import APIRouter, BackgroundTasks, Depends

app08 = APIRouter()

"""【见run.py】Middleware 中间件"""

# 注：带yield的依赖的退出部分的代码 和 后台任务 会在中间件之后运行

"""【见run.py】CORS (Cross-Origin Resource Sharing) 跨源资源共享"""

# 域的概念：协议+域名+端口
# 如果协议、域名、端口有其中一个不一样，就是不符合同源策略，就会有跨域问题
"""Background Tasks 后台任务"""


def bg_task(framework: str):
    with open("README.md", mode="a") as f:
        f.write(f"## {framework} 框架精讲")

# 调用本接口就执行后台任务
@app08.post("/background_tasks")
async def run_bg_task(framework: str, background_tasks: BackgroundTasks):
    """
    :param framework: 被调用的后台任务函数的参数
    :param background_tasks: FastAPI.BackgroundTasks
    :return:
    """
    background_tasks.add_task(bg_task, framework) # 将回调函数传给BackgroundTasks去执行
    return {"message": "任务已在后台运行"}


def continue_write_readme(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        background_tasks.add_task(bg_task, "\n> 整体的介绍 FastAPI，快速上手开发，结合 API 交互文档逐个讲解核心模块的使用\n")
    return q


# 使用依赖注入实现后台任务
@app08.post("/dependency/background_tasks")
async def dependency_run_bg_task(q: str = Depends(continue_write_readme)):
    if q:
        return {"message": "README.md更新成功"}

测试用例：

test_chapter08.py
执行测试用例：cd到项目目录，然后执行pytest即可。测试用例会把所有的测试用例执行一遍。

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from fastapi.testclient import TestClient

from run import app

"""Testing 测试用例"""

client = TestClient(app)  # 先pip install pytest


def test_run_bg_task():  # 函数名用“test_”开头是 pytest 的规范。注意不是async def
    response = client.post(url="/chapter08/background_tasks?framework=FastAPI")
    assert response.status_code == 200
    assert response.json() == {"message": "任务已在后台运行"}


def test_dependency_run_bg_task():
    response = client.post(url="/chapter08/dependency/background_tasks")
    assert response.status_code == 200
    assert response.json() is None


def test_dependency_run_bg_task_q():
    response = client.post(url="/chapter08/dependency/background_tasks?q=1")
    assert response.status_code == 200
    assert response.json() == {"message": "README.md更新成功"}

项目参考：
https://github.com/tiangolo/full-stack-fastapi-postgresql前端：Vue后端：FastAPI数据库：PostgreSQL 部署：Docker不一定用得到 Docker，不过代码风格和 FastAPI 文件结构如何组织值得借鉴。

补充知识点

helloword

参考文章
pip install fastapi uvicorn

第1种运行FastAPI的方式:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def hello_world():
    return {"hello fast api"}

在命令行运行：uvicorn main:app --reload
之后默认访问：http://127.0.0.1:8000/即可看到["hello fast api"]的结果

第2种运行FastAPI的方式:
直接将下面内容保存为文件，比如hellword.py然后直接python helloword.py即可，然后访问：http://127.0.0.1:5555/即可得到结果

# -*- coding:utf-8 -*-

from fastapi import FastAPI
import uvicorn

app = FastAPI()


@app.get("/")
async def hello_world():
    return {"hello fast api"}


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

请求及参数获取

参考文章

# -*- coding:utf-8 -*-

from fastapi import FastAPI, Query, Body
import uvicorn

app = FastAPI()

# 路径参数的获取：http://127.0.0.1:5555/name=aa/age=423
@app.get("/name={n}/age={ag}")
async def server1(n, ag):  
    return {
        "name": n,
        "age": ag,
    }
# 注意: 获取路径参数的函数中的参数，需要与大括号里({})的参数名相同，比如上面的server1(n, ag)的n和ag必须要和get("/name={n}/age={ag}")相同

# get参数的获取：http://127.0.0.1:5555/get/?name=aa&age=423
@app.get("/get/")
async def server2(name=Query('default_name'), age=Query(None)):  # Query可以设置Get参数的默认值，如果不设置就是None
    return {
        "name": name,
        "age": age,
    }


# post参数：http://127.0.0.1:5555/post/
# post时body里设置：{"name":"aaa","age":123}
@app.post("/post/")
async def server3(name=Body(None), age=Body(None)):  # Body可以设置Post参数的默认值，如果不设置就是None
    return {
        "name": name,
        "age": age,
    }


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

使用依赖注入校验token

参考文章
方式1: 以函数的形式创建依赖项

async def verify_token(request: Request):
    token = request.headers.get("token")
    if not token:
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail="token 不正确"
        )
    # 假设到这一步，进行验证、解析token，假设解析出user_id···
    user_id = 'fcj'
    return user_id

@app.get("/get_info")
async def get_info(user_id: str = Depends(verify_token)):
    # 通过user_id获取用户信息
    user_info = {
        "user_id": user_id,
        "name": "tigeriaf"
    }
    return user_info

方式1: 以类的形式创建依赖项

class UserInfo:
    def __init__(self, request: Request):
        token = request.headers.get("token")
        if not token:
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="token 不正确"
            )
        # 验证、解析token，假设解析出user_id···
        self.user_id = 111
        self.name = "tigeriaf"


@app.get("/get_info")
async def get_info(user_info: UserInfo = Depends(UserInfo)):
    user_info = {
        "user_id": user_info.user_id,
        "name": user_info.name
    }
    return user_info

报错

获取不到参数

报错："状态代码: 422 Unprocessable Entity"
获取不到参数/获取不到post参数

# POST 数据:
# {username: "admin", password: "1e64a7c3d7bb7cbe6cd22bd68cfecd9e"}

# (错误的)响应:
# {
#     "detail":[
#         {
#             "loc":[
#                 "query",
#                 "password"
#             ],
#             "msg":"field required",
#             "type":"value_error.missing"
#         },
#         {
#             "loc":[
#                 "query",
#                 "username"
#             ],
#             "msg":"field required",
#             "type":"value_error.missing"
#         }
#     ]
# }

# FastAPI代码
from fastapi import APIRouter, status, Form, File, UploadFile, HTTPException, Body

# 错误代码
@app.post("/login")
async def login(password: str, username: str):  # 定义表单参数，如果是提交的表单而不是json(即请求头是Content-Type: application/x-www-form-urlencoded，不是"Content-Type: application/json;charset=UTF-8"，必须使用Form(...)接收
    """用Form类需要pip install python-multipart; Form类的元数据和校验方法类似Body/Query/Path/Cookie，参考它们去使用就行了"""
    print(f"{username}\t-\t{password}")
    return {"username": username, 'token': '123TokenOk'}

# 正确代码
@app.post("/login")
async def login(password=Body('abc'), username=Body('cba')):  # 定义表单参数，如果是提交的表单而不是json(即请求头是Content-Type: application/x-www-form-urlencoded，不是"Content-Type: application/json;charset=UTF-8"，必须使用Form(...)接收
    """用Form类需要pip install python-multipart; Form类的元数据和校验方法类似Body/Query/Path/Cookie，参考它们去使用就行了"""
    print(f"{username}\t-\t{password}")
    return {"username": username, "password": password, 'token': '123TokenOk'}


# 正确写法是使用Body('默认值')来接收，如果没有默认值那就是Body(None)
# 错误写法：async def login(password: str = 'abc', username: str = 'cba'):  # 定义表单参数，如果是提交的表单而不是json(即请求头是Content-Type: application/x-www-form-urlencoded，不是"Content-Type: application/json;charset=UTF-8"，必须使用Form(...)接收

我的FastAPI项目模板

Q.E.D.

做一个热爱生活的人