FastAPI框架

FastAPI 文档

目录

• FastAPI 文档
  • 1 前置基础介绍
    • 1.1 Web开发模式
    • 1.2 基于RESTful的FastAPI的设计风格
    • 1.3 REST接口开发的核心任务
  • 2 FastAPI介绍
    • 2.0 概述
    • 为什么选择 FastAPI ?
      • FastAPI 是Python领域(3.6+)用来构建 API 服务的一个高性能框架。
      • 一、快速
      • 二、简单易懂，易于上手
      • 三、健壮性强
      • 四、标准统一
      • 性能测试参考：
    • 2.1 环境安装与使用
    • 2.2 FastAPI特性
    • 3.1 使用虚拟环境
  • 3 FastAPI 基础内容
    • 3.1 启动项目 main.py
    • 3.2 路径参数
    • 3.3 查询参数
    • 3.4 FastAPI请求体
    • 3.5 查询参数和字符串验证
    • 3.6 路径参数和数值验证
    • 3.7 声明主体多参数
    • 3.8 body 字段
    • 3.9 body嵌套模型
    • 3.10 额外的数据类型
    • 3.11 cookie参数和header参数
    • 3.12 响应模型 Response Model
    • 3.13 扩展模型
    • 3.14 响应状态码
    • 3.15 表单数据
    • 3.16 请求文件
    • 3.17 错误处理
    • 3.18 路径操作配置
    • 3.20 JSON兼容编码
    • 3.21 body更新
    • 3.22 FastAPI 依赖部分
    • 3.23 FastAPI 中间件
    • 3.24 FastAPI Cors 跨越请求
    • 3.25 FastAPI 大应用-多个文件
  • 4 FastAPI进阶内容
    • 4.1 自定义响应
    • 4.2自定义响应cookie
    • 4.3 自定义设置响应头
    • 4.4 HTTP的基本验证
    • 4.5 Request 对象
    • 4.6 高级中间件
    • 4.7 事件启动与结束
    • 4.8 WebSocket
  • 5 FastAPI 认证
    • 5.1 安全介绍
    • 5.2 Jwt 介绍
    • 5.3 使用密码和JTW凭证进行认证
  • 6 数据库
    • 6.1 SQLAlchemy 数据库
    • 6.2 alembic 数据库迁移篇
    • 6.3 SQLAlchemy 案例
    • 6.4 tortoise-orm (数据库)
  • 7 FastAPI 日志管理
  • 8 FastAPI 缓存
  • 9 FastAPI Users 用户管理
  • 10 FastAPI 定期任务
  • 11 FastAPI后台接口文档
    • 1 接口文档说明：
    • 2 登录
      • 2.1 登录验证接口
    • 3 用户管理
      • 3.1 添加用户
    • 4 首页
      • 4.1 首页文章
      • 4.2 首页分类
      • 4.3 管理员
    • 4 后台管理
      • 4.1 文章管理
      • 4.2 分类管理
  • 9 docker + Nginx 部署FastAPI
    • 1 使用Pycharm 将本地项目上传到远程服务器
    • 2 docker部署项目
    • 4 nginx 转发

1 前置基础介绍

1.1 Web开发模式

FastAPI 适用于前后端分离模式

注：

​ 1 后端仅返回前端所需的数据

​ 2 API: 前后端分离模式中,每一个函数或试图称之为接口或API

1.2 基于RESTful的FastAPI的设计风格

在进行API接口设计时，应规定统一的接口设计方式采用的RESTful API设计风格

请求方式

不同的URL 采用不同的请求方式，代表不同的执行操作

常用的HTTP请求方式有下面四个：

请求方式	说明
GET	获取资源数据(单个或多个)
POST	新增资源数据
PUT	修改资源数据
DELETE	删除资源数据

响应数据格式

服务器返回的响应数据格式，应该尽量使用JSON

响应的状态码

服务器向客户端返回的状态码和提示信息

200 OK - [GET]：服务器成功返回用户请求的数据
201 CREATED - [POST/PUT/PATCH]：用户新建或修改数据成功
202 Accepted - [*]：表示一个请求已经进入后台排队（异步任务）
204 NO CONTENT - [DELETE]：用户删除数据成功
400 INVALID REQUEST - [POST/PUT/PATCH]：用户发出的请求有错误，服务器没有进行新建或修改数据的操作
401 Unauthorized - [*]：表示用户没有权限（令牌、用户名、密码错误）
403 Forbidden - [*] 表示用户得到授权（与401错误相对），但是访问是被禁止的
404 NOT FOUND - [*]：用户发出的请求针对的是不存在的记录，服务器没有进行操作
406 Not Acceptable - [GET]：用户请求的格式不可得
410 Gone -[GET]：用户请求的资源被永久删除，且不会再得到的
422 Unprocesable entity - [POST/PUT/PATCH] 当创建一个对象时，发生一个验证错误
500 INTERNAL SERVER ERROR - [*]：服务器发生错误，用户将无法判断发出的请求是否成功

1.3 REST接口开发的核心任务

将请求的数据（如JSON格式）转换为模型类对象

操作数据库

将模型类对象转换为响应的数据（如JSON格式）

序列化 ：将数据库数据序列化为前端所需要的格式 如（字典、JSON、XML等）

反序列化： 将前端发送的数据反序列化保存到模型类对象，并保存到数据库中

2 FastAPI介绍

2.0 概述

为什么选择 FastAPI ?

FastAPI 是Python领域(3.6+)用来构建 API 服务的一个高性能框架。

一、快速

性能极高，可与 NodeJS, Go 媲美。(得益于Starlette和Pydantic)。

Starlette 是一个轻量级 ASGI 框架/工具包。它非常适合用来构建高性能的 asyncio 服务，并支持 HTTP 和 WebSockets。

官方网址：https://www.starlette.io/

Pydantic 是一个使用Python类型提示来进行数据验证和设置管理的库。Pydantic定义数据应该如何使用纯Python规范用并进行验证。

官方网址：https://pydantic-docs.helpmanual.io/

二、简单易懂，易于上手

1、设计的初衷就是易于学习和使用。不需要阅读复杂的文档。

2、良好的编辑器支持。支持自动完成，可以花更少的时间用于调试。

3、代码复用性强。

4、方便的 API 调试，生成 API 文档。

5、能提高开发人员两到三倍的开发速度。减少40%的人为出错几率。

三、健壮性强

企业级代码质量。

四、标准统一

遵循以下API解决方案的开放标准：OpenAPI (也叫Swagger) and JSON Schema。

性能测试参考：

https://www.techempower.com/benchmarks/#section=test&runid=7464e520-0dc2-473d-bd34-dbdfd7e85911&hw=ph&test=query&l=zijzen-7

2.1 环境安装与使用

FastAPI是基于python3.6+和标准python类型的一个现代化的，快速的(高性能) 的异步框架，有自动式交互文档，非常适合构建RESTful API。

查看官方文档: https://fastapi.tiangolo.com

2.2 FastAPI特性

详情查看官方文档:

https://fastapi.tiangolo.com/features/

FastAPI的特点：

基于开放的标准

• OpenAPI的创建，包含了路径操作，参数，请求体和安全等的声明。
• 通过JSON Schema自动的数据模型文档。
• 围绕着这些标准设计，通过严谨的学习，而不是在上面加一层后记。
• 也允许在很多种语言中使用自动的客户端代码生成器

3.1 使用虚拟环境

使用虚拟环境安装fastapi

引用：https://www.cnblogs.com/chjxbt/p/10517952.html

3 FastAPI 基础内容

本章案例在 test_fastapi/FastAPI基础内容目录下运行

3.1 启动项目 main.py

from fastapi import FastAPI
import uvicorn
# 创建app应用
app = FastAPI()

# 路径装饰器
@app.get("/")
async def root():
    return {"message": "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}
   
# 现测试环境采用调式模式运行
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

进入交互式API文档

现在进入 http://127.0.0.1:8000/docs.

你会看到自动生成的交互式api文档，可以进行操作

3.2 路径参数

我们可以声明路径“参数”或“变量”与Python格式字符串使用相同的语法:

# 路径参数
from fastapi import FastAPI

app = FastAPI()


@app.get("/me/xx")
async def read_item_me():
    return {"me": 'me'}


@app.get("/me/{item_id}")  # item_id 为路径参数
async def read_item(item_id: str):
    return {"item_id": item_id}


@app.get("/")
async def main():
    return {"message": "Hello，FastAPI"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

路径参数item_id的值将作为参数itemid传递给函数。

运行这个示例并转到http://127.0.0.1:8000/items/foo，您将看到一个响应:

{"item_id" : "foo"}

声明带有类型的路径参数

注意：声明类型从参数为必填项

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

进入API交互式文档中校验

3.3 查询参数

当您声明不属于路径参数一部分的其他函数参数时，它们将被自动解释为“查询”参数。

from fastapi import FastAPI  

app = FastAPI()

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


@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):  # 例如skip就是查询参数
    return fake_items_db[skip : skip + limit]

查询是在?之后的键-值对集合。在URL中，用&字符分隔(查询字符串)

例如，在url中: 或者进入交互式文档中

http://127.0.0.1:8000/items/?skip=0&limit=10  

查询参数为： skip:值为0； limit:值为10

同样我们可以设置为可选参数 为None , 以及 参数类型装换

还有多路径参数 @app.get("/users/{user_id}/items/{item_id}")

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item
  

3.4 FastAPI请求体

当你需要从一个客户端(简单来说是浏览器)发送数据给你的API,你发送的内容叫做请求体。

一个请求体数据通过客户端发送到你的API。你的API返回一个响应体给客户端。

你的API总是发送一个响应体，但是客户端不必一直发送请求体。

为了声明一个请求体，使用Pydatic模型支持。

注意： 使用Pydatic 发送请求体不能是GET 请求，必须使用POST,PUT,DELETE等发送数据

from fastapi import FastAPI  
from pydantic import BaseModel  

# 这个是创建数据模型 继承与BaseModel
class Item(BaseModel):
    name: str
    description: str = None  # 不必要的默认值可以用Noen
    price: float
    tax: float = None  


app = FastAPI()
@app.post("/items/")
async def create_item(item_id:int, item: Item, q:str=None):  # 声明一个item参数指向Item数据模型
  	print(item.name)
    print(item.dict())
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

查看交互式文档

返回的为json数据

3.5 查询参数和字符串验证

FastAPI允许声明额外的信息并且验证你的参数。

从fastapi中导入Query

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: str = Query(None, min_length=3, max_length=50)):  # 限制长度
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

3.6 路径参数和数值验证

可以使用Query声明查询参数的更多验证和元数据的方式相同，也可以使用Path声明相同类型的验证和路径参数的元数据。

从fastapi中导入Path

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(..., title="The ID of the item to get"，ge=1),
    q: str = Query(None, alias="item-query"),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

始终需要一个路径参数，因为它必须是路径的一部分。

.. 作为改标记的必填项 gt:大于，le:小于等于，lt:小于

3.7 声明主体多参数

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):  # 项目数据模型
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel): # 用户数据模型
    username: str
    full_name: str = None


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

在这种情况下，FastAPI将注意到该函数中有多个主体参数（两个参数是Pydantic模型）。

因此，它将使用参数名称作为正文中的键（字段名称），并期望一个类似于以下内容的正文：

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

3.8 body 字段

您可以使用Query，Path和Body在路径操作函数参数中声明其他验证和元数据的方式相同，也可以使用Pydantic的Field在Pydantic模型内部声明验证和元数据。

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

# 请求体参数验证，使用 Field 字段
class Item(BaseModel):
    name: str
    description: str = Field(None, title="The description of the item", max_length=300)
    price: float = Field(..., gt=0, description="The price must be greater than zero")
    tax: float = None


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

demo2

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str = None
    price: float = Field(..., gt=0)
    tax: float = None


@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int,
    item: Item = Body(...,
        example={   # example是Body里没有的字段；不会添加任何验证，而只会添加注释；不是example也不行
            "name": "Foo",
            "description": "A very nice Item",  # 默认值
            "price": 0,
            "toooo": 3.2,
        },
    )
):
    results = {"item_id": item_id, "item": item}
    return results


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

Field的工作方式与Query，Path和Body相同，它具有所有相同的参数，等等。

3.9 body嵌套模型

使用FastAPI，可以定义，验证，记录和使用任意深度嵌套的模型（这要归功于Pydantic）

from typing import Optional, List

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义一个数据模型，子模型
class Image(BaseModel):
		url: str
		name: str
  

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []  # 列表类型
    tags2: Set[str] = set() # 集合类型
   	# 嵌套另一个子模型
    image: Optional[Image] = None
      


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

这意味着FastAPI将期望类似于以下内容的主体：

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": ["rock", "metal", "bar"],
  	"tags2": [
      "aaa",
      "aaaa"
    ],
    "image": {
        "url": "http://example.com/baz.jpg",
        "name": "The Foo live"
    }
}

3.10 额外的数据类型

目前为止我们看到的常用的数据类型有int，str，float，bool

除此之外还以更多的混合数据类型

UUID

* 一个标准的“通用唯一标识符”，在许多数据库和系统中通常作为ID使用。

*在请求和响应中将以str表示。

datetime.datetime 例如:2008-09-15T15：53：00 + 05：00

datetime.date: 例如：2008-09-15

datetime.time: 例如: 14：23：55.003

datetime.timedelta: 例如: Pydantic还允许将其表示为“ ISO 8601时间差异编码

# 额外的数据类型

from datetime import datetime, time, timedelta
from uuid import UUID
from uuid import uuid1
from fastapi import Body, FastAPI

app = FastAPI()


# 额外数据类型
# https://fastapi.tiangolo.com/tutorial/extra-data-types/
@app.put("/items/{item_id}")
async def read_items(
        item_id: UUID,
        start_datetime: datetime = Body(None),
        end_datetime: datetime = Body(None),
        repeat_at: time = Body(None),
        process_after: timedelta = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process

    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }


if __name__ == '__main__':
    import uvicorn

    print('uuid:', uuid1())
    uvicorn.run(app, host="127.0.0.1", port=8000)

本章具体解释请看官方文档 # https://fastapi.tiangolo.com/tutorial/extra-data-types/

3.11 cookie参数和header参数

cookie demo

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(*, ads_id: str = Cookie(None)):
    return {"ads_id": ads_id}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

header demo

from fastapi import FastAPI, Header
from typing import List

app = FastAPI()


# 要声明标头，您需要使用Header，否则参数将被解释为查询参数。

@app.get("/items/")
async def read_items(*, user_agent: str = Header(None), users_agent: str =Header(None)):
    
    return {"User-Agent": user_agent}, {"AAAAA": user_agent}, {'ABCD': users_agent}

if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

3.12 响应模型 Response Model

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

@app.get(), @app.post, @app.put @app.delete 等

from typing import List

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.post("/items/", response_model=Item)  # 响应输出pydantic模型
async def create_item(item: Item):
    return item

注意，response_model是“decorator”方法的参数（get，post等）。不像所有参数和主体一样，具有路径操作功能。

它接收的类型与您为Pydantic模型属性声明的类型相同，因此它可以是Pydantic模型，但也可以是例如一个list的Pydantic模型，例如List[Item]

FastAPI将使用此response_model进行以下操作：

• 把输出数据转化为类型声明的格式
• 验证数据
• 给响应添加一个json 模式, 在OpenAPI的路径操作下。
• 将由自动文档系统使用

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str = None


# Don't do this in production!
@app.post("/user/", response_model=UserIn)  # 声明响应模型
async def create_user(*, user: UserIn): 
    return user	# 同一模型声明输出

现在，每当浏览器使用密码创建用户时，API都会在响应中返回相同的密码。

在这种情况下，这可能不是问题，因为用户自己正在发送密码。

但是，如果我们对另一个路径操作使用相同的模型，则可能会将用户的密码发送给每个客户端。

应该添加一个输出模型

我们可以改用纯文本密码创建输入模型，而没有明文密码则创建输出模型：

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):  # 输入模型，在交互式文档中输入的
    username: str
    password: str
    email: EmailStr
    full_name: str = None
      
class UserOut(BaseModel):  # 输出模型 执行后所输出的模型数据
  	username:str
    email: EmailStr
    fill_name: str = None


# Don't do this in production!
@app.post("/user/", response_mode:UserOut)  # 声明响应模型
async def create_user(*, user: UserIn): 
    return user	# 同一模型声明输出
  

因此，FastAPI将负责过滤掉未在输出模型中声明的所有数据（使用Pydantic）。

在文档中查看

当您看到自动文档时，可以检查输入模型和输出模型是否都具有自己的JSON模式：

输入的模型数据

输出的数据

概括

使用路径操作修饰器的参数response_model定义响应模型，尤其是确保私有数据被过滤掉。

3.13 扩展模型

继续前面的示例，通常会有多个相关模型

用户模型尤其如此，因为：

input model 需要有一个密码

Output model 不需要密码

databases model 可能需要一个哈希密码

减少代码重复是FastAPI的核心思想之一。

随着代码重复的增加，错误，安全问题，代码不同步问题（当您在一个地方进行更新，而不是在另一个地方进行更新）等问题的可能性也增加了。

这些模型都共享大量数据，并且复制属性名称和类型。我们可以声明一个UserBase模型作为其他模型的基础。

然后，我们可以使该模型的子类继承其属性（类型声明，验证等）。

这样，我们可以只声明模型之间的差异（使用纯文本密码，使用hashed_password和不使用密码):

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str = None


# UserBase 作为模型的基础，其继承，相同的数据不要写 减少代码重复是FastAPI的核心思想之一
class UserOut(UserBase):
    pass


class UserInDB(UserBase):
    hashed_password: str


class UserIn(UserBase):
    password: str


# 进行哈希密码
def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


# 保存用户
def fake_save_user(user_in: UserIn):
    hash_password = fake_password_hasher(user_in.password)
    # 保存用户
    # **user_id.dict() 表示一个完整的模型数据dict **称之为一个关键字参数
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hash_password) # 伪哈希加密

    print("user saved! ... not really")

    print(user_in_db)

    return user_in_db


@app.post("/user/", response_model=UserOut)  # 响应输入模型
async def create_user(*, user_in: UserIn):  # 响应输出模型
    user_saved = fake_save_user(user_in)
    return user_saved


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

可以进行交互式文档查看

响应模型列表

同样，您可以声明对象列表的响应。

为此，请使用标准的Python typing.List

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

app = FastAPI()

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


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


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

得到的响应结果

200 ok
  {
    "name": "Foo",
    "description": "There comes my hero"
  },
  {
    "name": "Red",
    "description": "It's my aeroplane"
  }
]

同样你可以带dict中的响应

概括
使用多个Pydantic模型，并针对每种情况自由继承。

如果每个实体必须能够具有不同的“状态”，则无需为每个实体拥有单个数据模型。以用户“实体”为例，其状态包括password，password_hash和无密码。

3.14 响应状态码

响应状态码我们可以在任何路径操作中使用status_code 声明用于响应的HTTP状态代码

例：

from fastapi import FastAPI

app = FastAPI()


@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

每一个状态表示着不同操作

我们还可以用 fastapi.status的便捷变量

from fastapi import FastAPI, status

app = FastAPI()


@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

3.15 表单数据

安装 pip install python-multipart

导入

from fastapi import FastAPI, Form

使用正向或查询的方式创建表单参数

from fastapi import FastAPI, Form

app = FastAPI()


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

在登录注册项 或许需要发送username和password作为表单字段

规范要求这些字段必须准确命名为username和password，并作为表单字段（而不是JSON）发送。

3.16 请求文件

我们可以使用文件定义客户端要上传的文件

安装：pip install python-multipart

导入 File

from fastapi import FastAPI, File, UploadFile

定义File参数

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


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

这些文件将作为“表单数据”上传。

如果将路径操作函数参数的类型声明为字节(bytes)，FastAPI将为您读取文件，并且您将接收内容（字节）。

请记住，这意味着全部内容将存储在内存中。这对于小文件将很好地工作。

但是在某些情况下，您可以从使用UploadFile中受益。

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

该返回文件名

3.17 错误处理

在许多情况下，您需要将错误通知给正在使用API的客户端。

该客户端可以是带有前端的浏览器，其他人的代码，IoT设备等。

我们需要告知客户端: 客户端没有足够的权限进行该操作。客户端无权访问该资源。客户端尝试访问的项目不存在。

在这些情况下，您通常会返回400（从400到499）范围内的HTTP状态代码。

使用 HTTPException

返回有错误的http响应给客户端使用 HTTPException

导入 from fastapi import FastAPI, HTTPException

HTTPException是普通的Python异常，其中包含与API相关的其他数据。

因为它是Python异常，所以您不返回它，而是raise它。

在此示例中，当客户端通过不存在的ID请求商品时，引发状态代码为404的异常：

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, 
                            detail="Item not found"
                           headers={"X-Error": "There goes my error"})  # 自定义请求头
                    # X-Error 自定义。例如，对于某些类型的安全性。OAuth 2.0和某些安全实用程序在内部需要/使用此功能。

    return {"item": items[item_id]}  

结果响应

如果客户端请求http://example.com/items/foo（item_id为“bar”），则他将收到HTTP状态代码200和JSON响应：

{
  "item": "The Foo Wrestlers"
}

但是，如果客户端请求http://example.com/items/bar（不存在的item_id“ bar”），则他将收到HTTP状态码404（“找不到”错误）和JSON响应：

{
  "detail": "Item not found"
}

3.18 路径操作配置

您可以将几个参数传递给路径操作装饰器以对其进行配置。

注意，这些参数直接传递给路径操作装饰器，而不是传递给路径操作函数

官方文档: https://fastapi.tiangolo.com/tutorial/path-operation-configuration/

3.20 JSON兼容编码

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

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

为此，FastAPI提供了jsonable_encoder()函数。

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

例如，它不接收日期时间对象，因为它们与JSON不兼容。

因此，datetime对象必须转换为包含ISO格式数据的str。

同样，该数据库将不会接收Pydantic模型（具有属性的对象），而只会接收dict。

您可以为此使用jsonable_encoder。

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

from datetime import datetime

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

fake_db = {}


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str = None


app = FastAPI()


@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data
    # jsonable_encoder实际上由FastAPI在内部用于转换数据。但这在许多其他情况下很有用。
    print(json_compatible_item_data)
    print(type(json_compatible_item_data))
    print(fake_db)
    print(type(fake_db))
    
    return fake_db
    
# 在此示例中，它将Pydantic模型转换为dict，并将日期时间转换为str。
#调用它的结果是可以用Python标准json.dumps()进行编码的东西、。

3.21 body更新

官方文档：https://fastapi.tiangolo.com/tutorial/body-updates/

3.22 FastAPI 依赖部分

FastAPI有一个非常强大但直观的依赖注入系统。

依赖是在函数的执行前调用

当进行需要执行依赖操作时，这将非常有用，前提是：

• 具有共享逻辑（一次又一次地使用相同的代码逻辑）。
• 共享数据库连接
• 强制执行安全性，身份验证，角色要求等

所有这些，同时最大限度的减少了代码的重复

创建一个依赖项或“ dependable”

让我们首先关注依赖关系

它只是一个函数，可以采用路径操作函数可以采用的所有相同参数：

from fastapi import Depends, FastAPI

app = FastAPI()


# 被依赖的函数
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


# Depends 接收的参数必须是类似于函数
# 这里的 Depends 当每次请求前会调用 Depends()中的依赖项，
# 从这个依赖项函数中获取结果，最终将该结果分配给路径操作函数中的参数，也就是 common参数
# 这样，我们只需编写一次共享代码，FastAPI便会为路径操作调用共享代码。
@app.get("/items/")
async def read_items(common: dict = Depends(common_parameters)):
    common["test"] = "test"

    return common


@app.get("/users/")
async def read_users():
    return {"code": 200, "message": "ok"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

向路径操作装饰器添加依赖项

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


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

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

这些依赖将会在安全会中起到很大用处

3.23 FastAPI 中间件

“中间件”是一种功能，该功能可在任何请求通过任何特定路径操作处理之前与每个请求一起使用。以及返回之前的每个响应。

要创建中间件，请在函数顶部使用修饰符@app.middleware("http")。

import time

from fastapi import FastAPI, Request

app = FastAPI()

# 中间件，该功能在任何路径操作之前，先执行该中间件
# 计算http请求和生成响应所花费的时间
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response
  
  
@app.get("/")
async def main():
    return {"message": "Hello World"}

例如添加CORSMiddleware 中间件

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]
# app.add_middleware()`接收中间件类作为第一个参数，并接收任何其他要传递给中间件的参数。
app.add_middleware(         # 添加中间件
    CORSMiddleware,         # CORS中间件类
    allow_origins=origins,  # 允许起源
    allow_credentials=True,  # 允许凭据
    allow_methods=["*"],    # 允许方法
    allow_headers=["*"],    # 允许头部
)



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

有关CORS的更多信息，请查看Mozilla CORS文档。

3.24 FastAPI Cors 跨越请求

CORS或“跨源资源共享”是指浏览器中运行的前端具有与后端进行通信的JavaScript代码，并且后端与前端具有不同的“来源”的情况。

Origin是协议（http，https），域（myapp.com，localhost，localhost.tiangolo.com）和端口（80、443、8080）的组合。

使用 CORSMiddleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware( # 添加中间件
    CORSMiddleware,  # CORS中间件类
    allow_origins=origins,  # allow_origins=['*'], # 允许起源所有
    allow_credentials=True, # 允许凭据
    allow_methods=["*"],  # 允许方法
    allow_headers=["*"],  # 允许头部
)


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

3.25 FastAPI 大应用-多个文件

如果要构建应用程序或Web API，则很少将所有内容都放在一个文件中。 FastAPI提供了一种方便的工具，可在保持所有灵活性的同时构建应用程序。

相当与flask中的蓝图

文件结构

fastapi-nano
├── app                                 # primary application folder
│   ├── apis                            # this houses all the API packages
│   │   ├── api_a                       # api_a package
│   │   │   ├── __init__.py             # empty init file to make the api_a folder a package
│   │   │   ├── mainmod.py              # main module of api_a package
│   │   │   └── submod.py               # submodule of api_a package
│   │   └── api_b                       # api_b package
│   │       ├── __init__.py             # empty init file to make the api_b folder a package
│   │       ├── mainmod.py              # main module of api_b package
│   │       └── submod.py               # submodule of api_b package
│   ├── core                            # this is where the configs live
│   │   ├── auth.py                     # authentication with OAuth2
│   │   ├── config.py                   # sample config file
│   │   └── __init__.py                 # empty init file to make the config folder a package
│   ├── __init__.py                     # empty init file to make the app folder a package
│   ├── main.py                         # main file where the fastAPI() class is called
│   ├── routes                          # this is where all the routes live
│   │   └── views.py                    # file containing the endpoints of api_a and api_b
│   └── tests                           # test package
│       ├── __init__.py                 # empty init file to make the tests folder a package
│       └── test_api.py                 # test files
├── docker-compose.yml                  # docker-compose file
├── Dockerfile                          # dockerfile
├── LICENSE                             # MIT license
├── Makefile                            # Makefile to apply Python linters
├── mypy.ini                            # type checking configs
├── requirements.txt                    # app dependencies
└── requirements-dev.txt                # development dependencies

这样做的好处就是能够将独立功能的代码分开来进行管理编写

参照文档: https://fastapi.tiangolo.com/tutorial/bigger-applications/

4 FastAPI进阶内容

本节案例,请访问 test_fastapi/FastAPI进阶内容目录

4.1 自定义响应

默认情况下FastAPI中将使用JSONResponse 返回响应

但是除了JSONResponse 响应之外，还有 Response，HTML Response,RedirectResponse(重定向)响应等

HTML Response 响应

要直接从FastAPI返回带有HTML的响应，请使用HTMLResponse。

• 导入HTMLResponse。
• 将HTMLResponse作为路径操作的参数content_type传递。

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """

参数response_class也将用于定义响应的“媒体类型”。

HTTP标头Content-Type将设置为text/html。

RedirectResponse 重定向响应

# 重定向
# 返回HTTP重定向，默认情况下使用307状态代码（临时重定向）
@app.get("/typer")
async def read_typer():
    return RedirectResponse("https://typer.tiangolo.com")

当我们访问url时，默认情况会跳转到我们指定的url

Response响应

Response 返回的是文本或字节

它接受以下参数：

• content 一个str或bytes`。
• status_code - 一个int HTTP状态码
• headers - 一个字符串dict
• media_type - 一个给定媒体类型的str， 例如 "text/html"

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/legacy/")
def get_legacy_data():
    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")

4.2自定义响应cookie

当我们在响应头中响应cookie时，使用 response 参数

我们可以在路径操作函数中声明类型为Response的参数。然后，您可以在该时间响应对象中设置cookie

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return {"message": "Come to the dark side, we have cookies"}

或者直接返回一个Response 响应

当我们直接在代码中返回Response时，也可以创建cookie。

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


@app.post("/cookie/")
def create_cookie():
    content = {"message": "Come to the dark side, we have cookies"}
    response = JSONResponse(content=content)
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return response

4.3 自定义设置响应头

# 使用一个Response参数
# 您可以在路径操作函数中声明类型为Response的参数（就像对cookie一样）。然后，您可以在该时间响应对象中设置标题。
import uvicorn
from fastapi import FastAPI, Response
from starlette.responses import JSONResponse

app = FastAPI()


@app.get("/headers-and-object/")
def get_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"message": "Hello World"}


# 或者直接返回一个响应头  -- 通常情况下会这样使用
@app.get("/headers/")
def get_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
    return JSONResponse(content=content, headers=headers)

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


# 自定义 Headers
# 请记住，可以使用“X-”前缀添加自定义专有标题。
#
# 但是，如果您希望浏览器中的客户端看到自定义标头，则需要使用Starlette's中记录的参数Exposure_headers将它们添加到CORS配置中（在CORS（跨源资源共享）中了解更多信息）。 CORS文件。

4.4 HTTP的基本验证

为了安全起见，我们可以使用最简单的情况，可以使用HTTP基本身份验证

使用简单的 Http Basic Auth 来进行验证，应该程序需要包含用户名和密码的标记来进行验证，如果未验证成功，则会返回HTTP 401 未经授权的错误。

• 导入HTTPBasic和HTTPBasicCredentials.
• 使用HTTPBasic创建一个"security 模式`
• 在路径操作中将此安全性与依赖项结合使用。
• 它返回HTTPBasicCredentials类型的对象：
  • 它包含发送的用户名和密码。

import secrets

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

app = FastAPI()

# http验证
security = HTTPBasic()


# 被依赖项函数，用于检查用户名和密码是否正确
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
    # 验证用户名和密码
    correct_username = secrets.compare_digest(credentials.username, "admin")
    correct_password = secrets.compare_digest(credentials.password, "admin")
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Basic"},
        )

    return credentials.username


@app.get("/users/me")
def read_current_user(username: str = Depends(get_current_username)):
    return {"username": username}


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

4.5 Request 对象

在FastAPI中 提供了Request 对象，在某些情况下，我们可能需要直接访问Request请求对象做一些额外的事情，例如 获取前端传递过来的参数。

假设您要在路径操作功能中获取客户端的IP地址/主机。

为此，您需要直接访问请求。

import uvicorn
from fastapi import FastAPI, Request, Form

app = FastAPI()


@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    client_host = request.client.host  # 获取客户端的主机ip
    client_path = request.url.path  # 获取当前url路径
    return {"client_host": client_host, "item_id": item_id, "path": client_path, "form": client_username}

# 通过声明类型为Request的路径操作函数参数，FastAPI将知道在该参数中传递Request。


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

具体详情功能请访问Request文档

https://www.starlette.io/requests/

4.6 高级中间件

由于FastAPI基于Starlette并实现了ASGI规范，因此可以使用任何ASGI中间件。

Fastapi是基于Starlette的一个ASGI框架

ASGI，为了规范支持异步的Python Web服务器、框架和应用之间的通信而定制的一个协议

ASGI中间件定义

from unicorn import UnicornMiddleware

app = SomeASGIApp()
new_app = UnicornMiddleware(app, some_config="rainbow")

FastAPI 集成了几种常用的中间件

1 HTTPSRedirectMiddleware

强制所有传入请求必须为https或wss。

app.add_middleware()接收中间件类作为第一个参数，并接收任何其他要传递给中间件的参数。

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# 强制所有传入请求必须为https或wss。
app.add_middleware(HTTPSRedirectMiddleware)


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


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

FastAPI在fastapi.middleware中提供了几种中间件，以方便开发人员。但是大多数可用的中间件直接来自Starlette。

为了查看其他的中间件，查看Starlette's Middleware docs和ASGI Awesome List

4.7 事件启动与结束

您可以定义在应用程序启动之前或在应用程序关闭时需要执行的事件处理程序（函数）。

startup事件 与 shutdown结束事件

from fastapi import FastAPI

app = FastAPI()

items = {}

# 程序启动前要做的事情
@app.on_event("startup")
async def startup_event():
    print("程序启动之前执行")
    items["foo"] = {"name": "Fighers"}
    items["bar"] = {"name": "Tenders"}


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

    return items[item_id]

# 程序结束后要做的事情
# shutdown
@app.on_event("shutdown")
def shutdown_event():
    # 程序结束之后执行
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")

4.8 WebSocket

HTML5定义了WebSocket协议，能更好的节省服务器资源和带宽，并且能够更实时地进行及时通讯。

简单示例 单个客户端连接

查看文档 ：https://www.starlette.io/websockets/

main.py

from fastapi import FastAPI
from starlette.responses import HTMLResponse
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:5000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


# @app.get("/")告诉FastAPI如何去处理请求
# 路径 /
# 使用get操作
@app.get("/")
async def get():
    # 返回表单信息
    return HTMLResponse(html)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    # 接收连接数
    await websocket.accept()

    try:
        while True:
            # 接收文本
            data = await websocket.receive_text()
            # 发送文本数据
            await websocket.send_text(f"Message text was: {data}")
    # 当WebSocket连接关闭时，await websocket.receive_text()会引发WebSocketDisconnect异常，
    except WebSocketDisconnect:
        # 断开连接
        await websocket.close()
        print("客户端关闭成功")




if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5000)

多个客户端连接收发消息

from fastapi import FastAPI
from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute
from starlette.websockets import WebSocketDisconnect

info = {}

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off" placeholder="" />
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            document.getElementById("messageText").placeholder="第一次输入内容为昵称";

            var ws = new WebSocket("ws://localhost:5000/ws");

            // 接收
            ws.onmessage = function(event) {
                // 获取id为messages的ul标签内
                var messages = document.getElementById('messages')
                // 创建li标签
                var message = document.createElement('li')
                // 创建内容
                var content = document.createTextNode(event.data)
                // 内容添加到li标签内
                message.appendChild(content)
                // li标签添加到ul标签内
                messages.appendChild(message)
            };

            var name = 0;
            // 发送
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()

                if (name == 0){
                    document.getElementById("messageText").placeholder="";
                    name = 1;
                }
            }
        </script>
    </body>
</html>
"""


# HTTPEndpoint HTTP端点 提供了基于类的视图模式来处理HTTP方法
class Homepage(HTTPEndpoint):
    async def get(self, request):
        # 接受一些文本或字节并返回HTML响应。
        return HTMLResponse(html)


# WebSocketEndpoint WebSocketEndpoint
# 该WebSocketEndpoint 是一个ASGI应用呈现，封装了WebSocket实例。
class Echo(WebSocketEndpoint):
    encoding = "text"

    # 修改socket
    async def alter_socket(self, websocket):
        socket_str = str(websocket)[1:-1]
        socket_list = socket_str.split(' ')
        socket_only = socket_list[3]
        return socket_only

    # 连接 存储
    async def on_connect(self, websocket):
        # 接受连接
        await websocket.accept()

        # 用户输入名称  receive_text() 接收资料
        name = await websocket.receive_text()

        socket_only = await self.alter_socket(websocket)
        # 添加连接池 保存用户名
        info[socket_only] = [f'{name}', websocket]  # 存储到字典中

        # 先循环 告诉之前的用户有新用户加入了
        for wbs in info:
            await info[wbs][1].send_text(f"{info[socket_only][0]}-加入了聊天室")

    # 收发
    async def on_receive(self, websocket, data):

        socket_only = await self.alter_socket(websocket)

        for wbs in info:
            await info[wbs][1].send_text(f"{info[socket_only][0]}：{data}")

            print(wbs)

    # 断开 删除
    async def on_disconnect(self, websocket, close_code):
        try:
            socket_only = await self.alter_socket(websocket)
            # 删除连接池
            info.pop(socket_only)
            print(info)
        except WebSocketDisconnect:
            await websocket.close()


routes = [
    Route("/", Homepage),
    WebSocketRoute("/ws", Echo)
]

app = FastAPI(routes=routes)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5000)

5 FastAPI 认证

5.1 安全介绍

官方文档：https://fastapi.tiangolo.com/tutorial/security/

5.2 Jwt 介绍

JSON Web Token（JWT) 是目前前后端分离开发中用户身份认证的一种解决方案

这是将JSON对象编码为长而密集的字符串且没有空格的标准。看起来像这样：

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZ

如果要使用JWT令牌并查看其工作原理，请查看https://jwt.io。

JWT 认证流程：

1. 用户使用用户名和密码来请求服务器进行登录
2. 服务器验证用户的登录信息
3. 服务器通过验证，生成一个token并返回给用户
4. 客户端存储token，并在每次验证请求携带上这个token值
5. 服务端验证token值，并返回数据

5.3 使用密码和JTW凭证进行认证

1 需要安装依赖的包

​ 安装pyjwt PyJWT来生成和验证Python中的JWT令牌

​ pip install pyjwt

​ 密码 passlib

​ PassLib是一个很棒的Python程序包，用于处理密码哈希。

​ 它支持许多安全的哈希算法和实用程序来使用它们。推荐的算法是“ Bcrypt”。

	pip install passlib[bcrypt]

demo

from fastapi import Depends, FastAPI, HTTPException  # , status
from starlette import status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

# 用户数据（模拟）
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[e-mail]",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[e-mail]",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,  # 关闭此用户
    },
}

app = FastAPI()


# 哈希密码（模拟）
def fake_hash_password(password: str):
    return "fakehashed" + password


# oauth2_scheme是令牌对象，token: str = Depends(oauth2_scheme)后就是之前加密的令牌
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")  # Bearer 持票人（承载、数据载体）


# 用户信息模型
class User(BaseModel):
    username: str
    email: str = None
    full_name: str = None
    disabled: bool = None


# 用户输入模型
class UserInDB(User):
    hashed_password: str  # 哈希密码
    username: str  # 此行多余，为了vscode不报错而已。


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


# 解码令牌（模拟）
def fake_decode_token(token):
    # This doesn't provide any security at all
    # Check the next version
    user = get_user(fake_users_db, token)
    return user


# 获取当前用户
# 如果用户不存在或处于非活动状态，则这两个依赖项都将仅返回HTTP错误。
async def get_current_user(token: str = Depends(oauth2_scheme)):
    print('oauth2_scheme', oauth2_scheme)
    print('token', token)
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",  # 无效的身份验证凭据
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


# 获取当前活跃用户，get（read_users_me）专属
async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token")  # name = johndoe,alice     password = secret,secret2
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 使用username来自表单字段的从（假）数据库获取用户数据。
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    # 先将这些数据放到Pydantic UserInDB模型中
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    # 如果正确返回
    return {"access_token": user.username, "token_type": "bearer"}


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


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

具体认证篇请看

6 数据库

6.1 SQLAlchemy 数据库

**1 ORM **

ORM 称之为对象关系映射。

ORM 将数据库中的表与面向对象中的类建立了一种对应关系。

当我们去操作数据库，数据库中的表或者表中的一条记录就可以直接通过操作类或者类实例来完成

SQLAlchemy 对象

SQLAlchemy 是ORM 知名的工具之一，为高效和高性能的数据库访问设计。

SQLAlchemy 实际上就是利用Python来构建关系型数据库结构与表达式的系统

SQLAlchemy 是对数据库操作的封装，让开发者不用直接和sql打交道，而是通过python对象

详情请看官方文档 https://www.sqlalchemy.org/

6.2 alembic 数据库迁移篇

alembic是用来做ORM模型与数据库的迁移与映射。alembic使用方式跟git有点类似，表现在两个方面，第一个，alemibi的所有命令都是以alembic开头；

整体目录

.
├── alembic.ini  # alembic 配置文件
├── db # 数据库链接配置
│   ├── __init__.py
│   └── base.py  
├── alembic  # 初始化仓库
│   ├── README
│   ├── __pycache__
│   │   └── env.cpython-37.pyc
│   ├── env.py
│   ├── script.py.mako
│   └── versions
│       └── __pycache__
└── models # 数据库模型
    ├── __init__.py
    └── user_models.py  

base.py

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


DB_URI = "mysql+pymysql://root:python12@127.0.0.1:3306/test6?charset=utf8"

engine = create_engine(DB_URI)
# 创建基类
Base = declarative_base(engine)

1 定义数据库模型 以SQLAlchemy 为例子 假设为user用户模型

user_models

# 数据库模型所对应的SQLAlchemy
from sqlalchemy import Column,Integer,String,create_engine
from db.base import Base

class User(Base):
    # 对应MySQL中数据表的名字
    __tablename__ = 'users'

    # 创建字段
    id = Column(Integer, primary_key=True)  # users表中的id字段(主键)
    username = Column(String(64), nullable=False, index=True)  # users表中的username字段
  

2 在终端中初始化，创建一个仓库

alembic init alembic

初始化完成后，会生成一个alembic.ini 配置文件以及一个alembic目录

3 修改配置文件

3.1 修改alembic.ini 配置文件，只修改数据库连接部分即可

将

sqlalchemy.url = driver://user:pass@localhost:port/dbname

修改为

sqlalchemy.url = mysql+pymysql://root:python12@127.0.0.1:3306/test6

3.2 修改alembic/env.py

将

target_metadata = None

修改为

import sys
import os

# 1.__file__：当前文件（env.py）
# 2.os.path.dirname(__file__)：获取当前文件的目录
# 3.os.path.dirname(os.path.dirname(__file__))：获取当前文件目录的上一级目录
# 4.sys.path: python寻找导入的包的所有路径
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from db.base import Base
from models import user_models  # 这里要引入orm模型,不然迁移文件里面没有表字段
target_metadata = Base.metadata  # 添加模型的元数据对象

4 生成迁移脚本

# 由于提供了模型类，所以可以用–autogenerate参数自动生成迁移脚本。运行
alembic revision --autogenerate -m "alembic table init"

生成成功后会得到一个迁移文件 在 alembic/versions 目录下

5 将生成的迁移脚本映射到数据库中

alembic upgrade head

如果后续要添加模型，修改模型字段，重复4、5步骤即可，

降级数据库的话

alembic downgrade 版本号

具体详情请查看附件

https://alembic.sqlalchemy.org/en/latest/

6.3 SQLAlchemy 案例

具体详情案例请访问 test_fastapi/ databases 目录下文件运行

from fastapi import FastAPI, Depends, HTTPException
from passlib.context import CryptContext
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from starlette import status

# ===================== 数据库操作初始化 ====================

# 链接是需要指定要用到的MySQL数据库
engine = create_engine('mysql+pymysql://root:python12@localhost:3306/test6?charset=utf8', echo=True)
Base = declarative_base()  # 生成SQLORM基类

# SessionLocal类的每个实例将是一个数据库会话。该类本身还不是数据库会话。
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建session对象
session = SessionLocal()  # 生成链接数据库的实例


class User_My(Base):  # 声明数据库某表的属性与结构
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)  # 主键
    email = Column(String(64), unique=True, index=True)
    hashed_password = Column(String(128))
    is_active = Column(Boolean, default=True)


# 执行下面的代码，会创建userinfo表（所有表结构）
# Base.metadata.create_all(engine)

# 创建 pydantic 数据模型
class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):  # 继承子模型 UserBase
    password: str


class User(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True  # Pydantic的orm_mode将告诉Pydantic模型读取数据，即使它不是dict而是ORM模型


# 创建app应用
app = FastAPI()


# ==================== 数据库操作 使用依赖项 ======================

def get_db():
    try:
        db = SessionLocal()
        yield db

    finally:
        db.close()
        print("数据库关闭")


# Context是上下文  CryptContext是密码上下文 schemes是计划 deprecated是不赞成（强烈反对）
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


# 验证哈希密码
# verify_password验证密码   plain_password普通密码      hashed_password哈希密码
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


# 获取密码哈希
def get_password_hash(password):
    return pwd_context.hash(password)


# =================== 数据库操作方法 =======================
# 通过id查询用户信息  查询前先添加一个数据
def get_user(db: session, user_id: int):
    users = db.query(User_My).filter(User_My.id == user_id).first()
    print("users_id", users)
    return users


# 新建用户到数据库中
def db_create_user(db: session, user: UserCreate):
    # if user.email:
    #     raise HTTPException(status_code=400,
    #                         detail="user exist")
    # 1 对密码进行加密
    hashed_password = get_password_hash(user.password)
    # 添加到数据库中
    db_user = User_My(email=user.email, hashed_password=hashed_password)
    # 添加到session中
    db.add(db_user)

    # 添加到数据库中
    db.commit()
    db.refresh(db_user)  # 刷新

    return db_user


# ============== post和get请求 ===============
# 新增用户请求
@app.post("/users/", response_model=User)
def db_create_users(user: UserCreate, db: session = Depends(get_db)):
    # Depends(get_db)使用依赖关系可防止不同请求意外共享同一链接
    return db_create_user(db=db, user=user)


# 查询用户请求
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: session = Depends(get_db)):
    db_user = get_user(db=db, user_id=user_id)

    print(db_user)
    if db_user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail="User not found")

    return db_user


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

6.4 tortoise-orm (数据库)

1 tortoise-orm 介绍

Tortoise ORM是受Django启发的易于使用的 异步 ORM （对象关系映射器）

**2 安装 tortoise-orm **

pip install tortoise-orm

以及MySQL 或其他db驱动程序

pip install tortoise-orm[aiomysql]

pip install tortoise-orm[asyncpg]

3 tortoise-orm 案例

运行此案例在 test_fastapi/databases 目录下运行

具体数据库使用及功能请查看下方

tortoise-orm迁移: https://tortoise-orm.readthedocs.io/en/latest/migration.html

tortoise-orm文档：https://tortoise-orm.readthedocs.io/en/latest/

7 FastAPI 日志管理

Loguru是一个旨在以Python带来令人愉悦的日志记录的库。

Loguru的主要概念是只有一个日志记录器。

pip install loguru

基本输出日志

from loguru import logger
logger.debug("这是一条debug日志")

集成到FastAPI 新建了一个文件包extensions/专门存放扩展文件
然后在文件目录下创建了extensions/logger.py文件， 简单配置

logger.py

import os
import sys
import time
import traceback
from loguru import logger

# 项目所在目录
# abspath 返回绝对路径
# dirname 返回文件路径
basedir = os.path.abspath(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


# 定位到log日志文件中
log_path = os.path.join(basedir, 'logs')

# print(log_path)
# 如果文件夹不存在就新建
if not os.path.exists(log_path):
    os.mkdir(log_path)
# 定义log文件名
log_path_error = os.path.join(log_path, f'{time.strftime("%Y-%m-%d")}_error.log')

# 日志简单配置
# 具体其他配置 可自行参考 https://github.com/Delgan/loguru
# retention  Cleanup after some time
# rotation New file is created each day at noon
# enqueue 确保日志完整的输出保存 异步写入
logger.add(log_path_error, rotation="12:00", retention="5 days", enqueue=True)

_ init _.py

from extensions import logger

使用

# 得先在 extensions/__init__.py导入logger 才可以这样导入logger
from extensions.logger import logger

logger.debug(f"日志记录")
logger.info(f"日志记录")
logger.error(f"xxx")

最后输出在 logs文件中

具体详情请看附件（文档）

https://pypi.org/project/loguru/

8 FastAPI 缓存

本案例在 test_fastapi/fastapi-cache目录下运行

1 安装第三方缓存

pip install fastapi-cache

2 demo

from fastapi import FastAPI, Depends, HTTPException
import json
from fastapi_cache import caches, close_caches
from fastapi_cache.backends.redis import CACHE_KEY, RedisCacheBackend

app = FastAPI()


def redis_cache():
    return caches.get(CACHE_KEY)


@app.get('/')
async def hello(
        cache: RedisCacheBackend = Depends(redis_cache)
):
    print(cache)
    # 获取redis对象
    in_cache = await cache.get('some_cached_key')
    if not in_cache:
        cache_data = {
            "name": "xiaoming",
            "age": 20,
            "address": "Shanghai"
        }
        # 设置redis对象
        await cache.set('some_cached_key', json.dumps(cache_data))

    return {'response': in_cache or 'default'}

# 清除缓存
@app.delete("/delete_redis_cache")
async def test_del_redis(cache: RedisCacheBackend = Depends(redis_cache)):
    # 1 获取redis对象
    in_cache_redis = await cache.get("some_cached_key")

    if in_cache_redis:
        print(in_cache_redis)
        # await cache.delete("some_cached_key")
    else:
        raise HTTPException(status_code=200, detail="未找到reids中的key")
    return {"resposne": "ok"}

# 事件处理函数

# 程序启动前要做的事情
@app.on_event('startup')
async def on_startup() -> None:
    print("启动")
    rc = RedisCacheBackend('redis://127.0.0.1:6379')
    caches.set(CACHE_KEY, rc)


# 程序结束后要执行的事情
@app.on_event('shutdown')
async def on_shutdown() -> None:
    print("退出")
    await close_caches()


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5000)

9 FastAPI Users 用户管理

本章案例在test_fastapi/FastAPI_User目录下运行

fastapi-user 能够快速向我们的FastAPI 项目添加注册和身份验证系统，FastAPI用户旨在尽可能地自定义和适应。

2.1 特征

• 可扩展的基本用户模型

• 即用型注册，登录，忘记和重置密码路径

• 即用型OAuth2流程

• 依赖可调用以将当前用户注入路由

• 可定制的数据库后端

  • 得益于databases 数据库包括 SQLALchemy异步后端
  • 支持mongodb 异步后端
  • Tortoise ORM backend included

• 多个可定制的身份验证码后端

  • 包括JWT身份验证后端
  • 包括Cookie身份验证后端

• 全面的OpenAPI架构支持，即使有多个身份验证后端

2.2 安装

具有SQLAlchemy支持

pip install fastapi-users[sqlalchemy]

包括Tortoise-orm支持

pip install fastapi-users[mongodb]

这里我们使用的tortoise-orm 数据库来作为此案例

本章案例在 test_fastapi/fastapi-users 运行，结合文档看详情说明。

tortoise-orm迁移: https://tortoise-orm.readthedocs.io/en/latest/migration.html

fastapi-users文档 https://frankie567.github.io/fastapi-users/

2.3 说明

混合提供了四种Pydantic模型变体

• BaseUser提供基础领域和验证;
• BaseCreateUser，专门用于用户注册，由必填字段email和password必填字段组成；
• BaseUpdateUser，专门用于用户个人资料更新，其中添加了一个可选password字段；
• BaseUserDB，这是数据库中用户的表示形式，并添加了一个hashed_password字段。

10 FastAPI 定期任务

本案例在 test_fastapi/fastapi-utils-tasks 目录下运行

1 安装 第三方模块

pip install fastapi-utils

2 定时任务

启动和关闭事件是触发与服务器生命周期相关的操作的好方法。

但是，有时您希望任务不仅在服务器启动时触发，而且要定期触发。例如，可能想定期重置内部缓存，或从数据库中删除过期的令牌。

可以通过 该fastapi_utils.tasks.repeat_every装饰实现定时任务

调用用@repeat_every(...)修饰器修饰的函数时，将启动循环，并以seconds提供给修饰器的参数确定的延迟定期调用该函数。

例子：

from fastapi import FastAPI
from sqlalchemy.orm import Session

from fastapi_utils.session import FastAPISessionMaker
from fastapi_utils.tasks import repeat_every

# 创建app应用
app = FastAPI()

ret = 0


async def tasks_message():
    global ret

    ret = ret + 1

    print(ret)

# 程序启动后调用该函数
@app.on_event("startup")
@repeat_every(seconds=60 * 1)  # 1 min 每一分钟调用一次
async def remove_expired_tokens_task() -> None:
    print("启动")
    # 调用函数
    await tasks_message()


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

关键字参数

以下是针对以下各项的各种关键字参数的详细说明repeat_every：

• seconds: float ：连续两次呼叫之间等待的秒数
• wait_first: bool = False：如果False（默认值），则在第一次调用修饰函数时立即调用包装函数。如果为True，则装饰函数将等待一个时间段，然后再调用包装函数
• logger: Optional[logging.Logger] = None ：如果您通过记录器，则在重复执行循环中引发的任何异常都将记录（带有回溯）到提供的记录器中。

具体详情文档请点击：https://fastapi-utils.davidmontague.xyz/user-guide/repeated-tasks/

11 FastAPI后台接口文档

博客类型项目实现了基本的增删改查

数据库为SQLAlchemy ORM

前台目录:

1. 在blog-crud/frontend目录下
2. 安装依赖: npm install
3. 启动开发: npm run dev
4. 打包命令: npm run build

后台项目:

1. 在blog-crud/fastapi-blog目录下
2. 数据库为 SQLAlchemy ORM

1 接口文档说明：

前端：http://127.0.0.1:8000/

后端API交互文档地址：http://127.0.0.1:5000/redoc

2 登录

2.1 登录验证接口

请求路径： /api/v1/login/access-token

请求方法: post

请求参数

username 用户名

password 密码

响应参数

access_token 基于jwt的令牌
token_type 令牌类型

3 用户管理

3.1 添加用户

• 请求路径: /api/v1/users
• 请求方法: POST
• 请求参数
  • email 邮箱
  • Username 用户姓名
  • Password 哈希密码
• 响应参数
  • email: 邮箱
  • Username: 用户姓名
  • id：用户id
  • Is_active 激活状态
  • is_admin 管理员

4 首页

4.1 首页文章

请求路径：GET api/v1/posts?page=1&limit=10 显示所有文章

4.2 首页分类

请求路径：GET api/v1/categories 显示所有分类

4.3 管理员

管理员(前提登录并且是管理员) 在User模型上中is_admin字段改为True

请求路径: /manage/posts 文章管理
请求路径: /manage/post? 文章编辑/创建
请求路径: /manage/categories 分类管理
请求路径: /manage/category? 分类编辑/创建

4 后台管理

4.1 文章管理

• 请求路径: GET /api/v1/posts/admin 后台显示所有文章

• 请求路径: POST /api/v1/posts/ 后台创建文章

• 请求参数:

• title: 标题 content: 内容 cover_image 图片 is_publish 是否发布 can_comment 是否评论 category_id 分类id（外键）

• 响应参数

  content: 创建文章成功 201

4.2 分类管理

• 请求路径: GET /api/v1/categories/ 显示所有分类

• 请求路径: POST /api/v1/categories/ 添加分类

• 请求参数: name: 分类姓名 img: 图片

• 请求路径: /api/v1/categories/<int: categories>

• 请求方法：DELETE

• 请求路径参数： category_id 分类id

9 docker + Nginx 部署FastAPI

1 使用Pycharm 将本地项目上传到远程服务器

注意只有Pycharm专业版才具有部署的功能，社区版（无需破解）是没有此功能

添加配置，起个名字

以上都配置好了，然后我们将我们本地的项目上传到远程服务器中

最后进入到我们的服务器查看上传的项目。

2 docker部署项目

• 转到您的项目目录。

  都需要在同一路径中

  

• 创建一个Dockerfile：

  FROM python:3.7
  # 复制文件到容器
  
  # 将本地项目和依赖包添加到容器 /home 路径
  ADD requirements.txt /home
  ADD ./app /home
  
  # 跳转到指定目录
  WORKDIR /home
  
  # 运行安装 requirements.txt
  RUN pip install -r requirements.txt -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com
  
  # 暴露端口
  EXPOSE 5000
  
  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "5000"]
  

• 创建一个docker-compose

  Compose项目是 Docker 官方的开源项目，负责实现对 Docker 容器集群的快速编排,同时也是对dockerfile进行管理

  创建 docker-compose.yml 配置文件

  version: '3'
  services:
    fastapi2:
      build: .  # 这个是查找当前dockerfile文件进行构建
      ports:
       - "5000:5000"
  
    mysql:
          image: mysql:5.7
          expose:
              - 3306
          ports:
              - "3307:3306"
          environment:
              MYSQL_ROOT_PASSWORD: python
              MYSQL_USER: root
  

• 运行构建镜像命令

  • docker-compose up -d --build

    用于部署一个 Compose 应用

    默认情况下该命令会读取名为 docker-compose.yml 或 docker-compose.yaml 的文件当然用户也可以使用 -f 指定其他文件名。通常情况下，会使用 -d 参数令应用在后台启动

  • docker-compose stop

    停止 Compose 应用相关的所有容器，但不会删除它们

    被停止的应用可以很容易地通过 docker-compose restart 命令重新启动

  • docker-compose rm

    用于删除已停止的 Compose 应用

    它会删除容器和网络，但是不会删除卷和镜像

  • docker-compose restart

    重启已停止的 Compose 应用

    如果用户在停止该应用后对其进行了变更，那么变更的内容不会反映在重启后的应用中，这时需要重新部署应用使变更生效

  • docker-compose ps

    用于列出 Compose 应用中的各个容器

    输出内容包括当前状态、容器运行的命令以及网络端口

  • docker-compose down
    停止并删除运行中的 Compose 应用

    它会删除容器和网络，但是不会删除卷和镜像

  最后当我们构建成功之后，会看到我们运行的程序，同时也可以访问服务器ip和端口

  

4 nginx 转发

nginx 是一个高性能HTTP服务器和反向代理服务器，最终的目的是实现网站的负载均衡；

nginx 作为web服务器，对处理索引文件和静态文件效率非常高，例如我们项目中的静态文件（HTML、CSS、JS）就交给nginx进行处理。

强大的反向代理和负载均衡功能，平衡集群中各个服务器的负载压力

安装nginx

sudo apt-get install nginx

2 运行(启动)nginx

nginx

3 服务器查看运行状态 或浏览器打开

3 停止/重启 nginx

nginx -s stop/reload

4 查询nginx路径

nginx -t   # 腾讯云的同学要加sudo
# 第一个文件路径是等会我们要去添加配置的文件
# nginx: the configuration file /etc/nginx/nginx.conf syntax is ok 
# nginx: configuration file /etc/nginx/nginx.conf test is successful

修改nginx.conf 文件 /etc/nginx/nginx.conf 使用sudo vim

注意重要的一点，这里部分截图的最后有两行文件

include /etc/nginx/conf.d/*.conf;

include /etc/nginx/sites-enabled/*; 

先说下原因： 因为nginx中有个默认配置文件default, 里面的默认监听端口是80端口，那么我们需要在阿里云配置的端口也是80端口，此时就会出现上我们自身配置的80端口是起冲突的，也就是说最终我们配置好了nginx信息，访问的时候还是nginx最初的页面 welcome to nginx 。 所以我们需要删除掉nginx默认的配置文件，让nginx访问我们配置的端口信息。

步骤：

cd 进入 /etc/nginx/sites-enabled/ 路径下能够看到有个default文件, 通过cat default 能够看到默认配置信息, 我们需要做的就是 删除 或者 对文件进行移动到上一层, mv deault ../ 此时nginx就不会加载默认的配置了,通过curl 122.51.67.247:80 是不会访问最初的配置了，而是显示的拒绝连接的了。

添加需要配置的文件到 vim /etc/nginx/nginx.conf

配置信息

 upstream app {
            server 139.129.99.129:5000;  # docker 配置
        }

        server {

             listen 80;
             location / {
                 proxy_pass http://app;  # 添加到这里
                 }

配置好nginx信息后 通过 nginx -t 检测配置是否正确

# 测试nginx配置文件是否正确
sudo nginx -t

# 如打印以下内容，表示配置没有问题
nginx: the configuration file /etc/nginx/nginx.conf syntax is ok
nginx: configuration file /etc/nginx/nginx.conf test is successful
  
# 这时我们需要重新加载下nginx配置
sudo nginx -s reload 

最后访问我们服务器ip就可以了