上一篇中说过，希望自己能坚持继续整理自己的在学习python web开发框架的Fastapi的相关的笔记。既然有了开头，所以也希望自己有个结尾，即使没结尾~也至少还有一个过程吧！所以继续整理之前我再使用过程遇到一些问题！

Fastapi 基础补充扩展

上一篇中相关的知识的介绍主要是官网的提供的示例多，仅仅是自己基于上面进行操作实际演练，而实际应用上，涉及的问题还是比较多，对于基础的上，其实还有很多需要注意的点。

1 Fastapi app对象初始化的配置项说明

app = FastAPI()
复制代码

从实例化一个应用开始，我们的FastAPI()就只是使用默认的参数，但是里面涉及的配置项还是比较多。这里我补充一下。

从配置项里我们可以学到主要有：

• 1：如何对app对象进行相关的描述

• 2：如何的配置我们的api文档的地址和关闭这个文档地址

• 3：如何添加APP启动和关闭的事件回调的通知

首先我们的进入内部源码看到的是:

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

里面涉及的参数还真不是一一般多的吧，但是所有的项都是非必传的项，都有相关的默认值。

这里我就简单给一些参数做简单的注释：

• debug: bool = False 是否开启debug

• routes: Optional[List[BaseRoute]] = None app 对象下挂在的所有的路由列表

• title: str = "FastAPI", 给我们app应用的文档定义一个标题文，会对应到我们的文档上的标题

• description: str = "" app应用的文档具体描述

• version: str = "0.1.0", app应用的文档版本号信息

• openapi_prefix: str = "" 访问的api地址的访问的前缀

• openapi_url: Optional[str] = "/openapi.json" api文档描述的整个JSON描述访问的URL地址，

PS：后续可以直接导入这个到 swagger，就可以看到我们的应用的所有的接口。【Swagger对于的openapi规范的文档格式】

如使如下的示例展示我们的API文档:

 return get_swagger_ui_html(
        openapi_url=openapi_url,
        title="Swagger UI",
        oauth2_redirect_url=oauth2_redirect_url,
        init_oauth=app.swagger_ui_init_oauth,
        swagger_js_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@3.48.0/swagger-ui-bundle.js",
        swagger_css_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@3.48.0/swagger-ui.css",
    )
    )
复制代码

• openapi_tags: Optional[List[Dict[str, Any]]] = None 每个API文档接口的标签

• servers: Optional[List[Dict[str, Union[str, Any]]]] = None 暂时不知道可以用来做什么

• dependencies: Optional[Sequence[Depends]] = None 全局依赖的对象（一般起全局作用）

• default_response_class: Type[Response] = Default(JSONResponse) 默认响应报文返回对象，它返回application/json格式的response。这个是FastAPI默认返回的response。

• docs_url: Optional[str] = "/docs" swagger api文档的访问的地址，可以自定义，

PS: 也可以去除设置为None则就是不存在这个API文档

• redoc_url: Optional[str] = "/redoc", redocapi文档的访问的地址，可以自定义，

• swagger_ui_init_oauth: Optional[Dict[str, Any]] = None, swagger ui api文档 授权初始化

• swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect" swagger api文档的访问的授权的回调地址

• middleware: Optional[Sequence[Middleware]] = None 应用的中间件

• exception_handlers: Optional[ Dict[ Union[int, Type[Exception]], Callable[[Request, Any], Coroutine[Any, Any, Response]], ] ] = None 应用的异常错误处理器对象，可以传入对于的函数进行的对应异常的处理

示例如:

from starlette.responses import Response


async def Not404Fount(request: Request, exc):
   response = Response('{"Error": "500"}')
   return response


async def Not500Fount(request: Request, exc):
   msg = ''
   if isinstance(exc, ZeroDivisionError):
       msg = 'ZeroDivisionError'
   response = Response(msg)
   return response


# 添加404
app.add_exception_handler(404, Not404Fount)
# 添加404
app.add_exception_handler(500, Not500Fount)

复制代码

• on_startup: Optional[Sequence[Callable[[], Any]]] = None 应用启动的时候的回调函数

• on_shutdown: Optional[Sequence[Callable[[], Any]]] = None 应用退出的时候的回调函数

启动和关闭的回调的示例：

def f_startup():
    print("启动起来")
    pass


def f_shutdown():
    print("关闭")
    pass


# 启动回调
app.add_event_handler('startup', f_startup)
# 关闭回调
app.add_event_handler('shutdown', f_shutdown)
复制代码

• root_path: str = "", 当前应用所处的根目录

• root_path_in_servers: bool = True,

• deprecated: Optional[bool] = None 标记当前的接口的文档当前版本是否可用，设置为True的情况下:

• include_in_schema: bool = True, 是否开启API文档的查看，False的话所有的从生成的OpenAPI架构（以及自动文档系统）会自动清除

2 配置项参数deprecated和include_in_schema其他说明

在app对象里面我们的deprecated和include_in_schema是起到全局的作用的，但是如果仅仅只是针对某些接口的进行相关接口的排除在文档中，或是进行标记当前版本的可用性的话，如果操作呢？

示例代码:

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading

@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync",include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"

# 这个接口在Api文档中 标记为已经无法使用，相当于过期，不建议再使用
@app.get("/async",deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成！"


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

文档展示结果:

3 写好API接口文档描述

PS:要想看到文档，第一次肯定是需要网络滴哟！当然你可以把这个/docs访问的静态页面所需的资源，对应的文档的访问使用自定义访问的方式，把相关涉及的资源地址请求转到本地:

具体方案就是：替换我们的：get_swagger_ui_html中包含的具体参数值信息：

• swagger_js_url

• swagger_favicon_url

• oauth2_redirect_url

get_swagger_ui_html源码如下：

def get_swagger_ui_html(
    *,
    openapi_url: str,
    title: str,
    swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui-bundle.js",
    swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui.css",
    swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
    oauth2_redirect_url: Optional[str] = None,
    init_oauth: Optional[Dict[str, Any]] = None,
) -> HTMLResponse:
复制代码

3.1 接口基本的文档描述

我们知道这个swagger API 文档是挺好用，但是一些接口的API的描述，我们需要配置上，不然我们的也不知道这个接口干嘛！其实说白了，就是我是英文的菜鸟，需要中文来解读！哈哈

比如这个接口，完全不知道用来干嘛！

一个好的API接口文档还是有需要滴！

示例如:

#!/usr/bin/evn python
# coding=utf-8


import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading


@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync", include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"


# 这个接口在Api文档中 标记为已经无法使用，相当于过期，不建议再使用
@app.get("/async", deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成！"


@app.get(path='/api/v1/get/user', summary='获取用户', description='这个接口是用来添加用户的', tags=['用户模块'])
def getuser():
    return {"code": 0, "msg": "获取成功", "data": None}


@app.post(path='/api/v1/add/user', summary='添加用户', description='这个接口是用来获取用户信息的', tags=['用户模块'])
def adduser():
    return {"code": 0, "msg": "添加成功", "data": None}


@app.put(path='/api/v1/updata/user', summary='更新用户', description='这个接口是用来更新用户信息的', tags=['用户模块'])
def updatauser():
    return {"code": 0, "msg": "修改成功", "data": None}


@app.put(path='/api/v1/delete/user', summary='删除用户', description='这个接口是用来删除用户信息的', tags=['用户模块'])
def deleteuser():
    return {"code": 0, "msg": "删除成功", "data": None}


@app.put(path='/api/v1/add/data', summary='新增数据', description='这个接口是用来新增数据', tags=['数据模块'])
def adddatas():
    return {"code": 0, "msg": "删除成功", "data": None}


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

文档效果示例:

3.2 API接口内部注释描述的作用

函数内部注释示例：

# 用户请求
@app.get("/asyncio/task")
async def do_task(username:str):
    '''
    后台异步人的接口
    :param username:  用户的姓名
    :return: 返回信息
    '''
    asyncio.create_task(write_log_2())
    return "你好，协程下的异步阻塞！"
复制代码

对应的文档显示注释的地方：

3.3 使用APIRouter的方式描述接口文档

代码示例如：

from fastapi import APIRouter
rstest = APIRouter(tags=["APIRouter相关"])

@rstest.post("/rstest", name="新增s示例",description="新增的示例的详细说明")
async def create():
    return "你好，协程下的异步阻塞！"

app.include_router(prefix="/v1", router=rstest)

if __name__ == "__main__":

    uvicorn.run('main:app', host="127.0.0.1", port=8000,reload=True)

复制代码

文档结果示例如：

3.4 接口相关参数文档描述

查看我们的源码

Query：源码

def Query(  # noqa: N802
    default: Any,
    *,
    alias: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    gt: Optional[float] = None,
    ge: Optional[float] = None,
    lt: Optional[float] = None,
    le: Optional[float] = None,
    min_length: Optional[int] = None,
    max_length: Optional[int] = None,
    regex: Optional[str] = None,
    example: Any = Undefined,
    examples: Optional[Dict[str, Any]] = None,
    deprecated: Optional[bool] = None,
    **extra: Any,
    
    
基于Query模块声明缺省值

- 可选参数声明 
- 形式：q: str = Query(None)  # 等同于  q: str = None


- 默认值参数声明
- 形式: q: str = Query("query")  # 等同于  q: str = "query"

- 必选参数声明
- 形式: q: str = Query(...)  # 等同于 q: str

自动化文档的附加信息主要参数：

- title：参数标题
- description：参数描述信息
- deprecated：表示参数即将过期，或不可用
特殊附加信息：
- alias：参数别名
    
复制代码

Body 源码：

def Body(  # noqa: N802
    default: Any,
    *,
    embed: bool = False,
    media_type: str = "application/json",
    alias: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    gt: Optional[float] = None,
    ge: Optional[float] = None,
    lt: Optional[float] = None,
    le: Optional[float] = None,
    min_length: Optional[int] = None,
    max_length: Optional[int] = None,
    regex: Optional[str] = None,
    example: Any = Undefined,
    examples: Optional[Dict[str, Any]] = None,
    **extra: Any,
) -> Any:


自动化文档的附加信息主要参数：

embed=True 加上这个会我们的示例example就不会显示出来
- title：参数标题
- description：参数描述信息
- deprecated：表示参数即将过期，或不可用
特殊附加信息：
- alias：参数别名
example ：示例
examples：示例列表
复制代码

他们的参数上都代理对于的相关的参数作用描述还有相关的示例传入。

所以我们的可以用上面的展示我们的参数的说明和解释.

示例如:

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading


@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync", include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成！"


# 这个接口在Api文档中 标记为已经无法使用，相当于过期，不建议再使用
@app.get("/async", deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成！"


@app.get(path='/api/v1/get/user', summary='获取用户', description='这个接口是用来添加用户的', tags=['用户模块'])
def getuser():
    return {"code": 0, "msg": "获取成功", "data": None}


@app.post(path='/api/v1/add/user', summary='添加用户', description='这个接口是用来获取用户信息的', tags=['用户模块'])
def adduser():
    return {"code": 0, "msg": "添加成功", "data": None}


@app.put(path='/api/v1/updata/user', summary='更新用户', description='这个接口是用来更新用户信息的', tags=['用户模块'])
def updatauser():
    return {"code": 0, "msg": "修改成功", "data": None}


@app.delete(path='/api/v1/delete/user', summary='删除用户', description='这个接口是用来删除用户信息的', tags=['用户模块'])
def deleteuser():
    return {"code": 0, "msg": "删除成功", "data": None}


@app.put(path='/api/v1/add/data', summary='新增数据', description='这个接口是用来新增数据', tags=['数据模块'])
def adddatas():
    return {"code": 0, "msg": "新增数据", "data": None}


@app.get("/api/v1/ceshi/data", summary='查看数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def read_items(
        uerid: str = Query(..., min_length=3, max_length=50,
                           title='用户uerid',
                           description='这个必须，不然找不到用户滴',
                           alias='UID',
                           example="UUU323423385783785"),
        uername: str = Query(..., min_length=3, max_length=50,
                             title='用户uername',
                             description='用户昵称',
                             alias='UNAME',
                             example="小钟同学"),

):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    print(uername)
    if uerid:
        results.update({"uerid": uerid})
    return results


class Item(BaseModel):
    name: str
    description: str = Field(None, title="描述", description="描写的详细内容", max_length=300)
    price: float = Field(..., gt=0, title="价格", description="价格具体说明")
    tax: float = None

# class Item(BaseModel):
#     name: str
#     description: str = None
#     price: float
#     tax: float = None

@app.post("/api/v1/udaoay/data", summary='修改数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def uddata(item: Item = Body(..., embed=False, example={'para1': "必填，格式要求是字符串", 'name': "必填,格式要求是字符串",
                                                             'description': "选填，格式要求是字符串,默认值是None",
                                                             'price': "必填，格式要求是float",
                                                             'tax': "选填，格式要求是浮点型"})):
    return {"code": 0, "msg": "修改成功", "data": item}



if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

那关于查询参数的结果示例描述如:

关于请求体的参数的结果示例描述如:

3.5 使用模型的Config来描述接口参数的示例

class Item(BaseModel):
    name: str
    description: str = Field(None, title="描述", description="描写的详细内容", max_length=300)
    price: float = Field(..., gt=0, title="价格", description="价格具体说明")
    tax: float = None

    class Config:

        schema_extra={
            "example":
                {
                    "name":"描写的详细内容",
                    "price": "价格具体说明",
                }
        }


@app.post("/api/v1/udaoay/data", summary='修改数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def uddata(item: Item = Body(..., embed=False, example={'para1': "必填，格式要求是字符串", 'name': "必填,格式要求是字符串",
                                                             'description': "选填，格式要求是字符串,默认值是None",
                                                             'price': "必填，格式要求是float",
                                                             'tax': "选填，格式要求是浮点型"})):
    return {"code": 0, "msg": "修改成功", "data": item}
复制代码

4 API安全性问题

API文档的对于我们的项目的而言就是所有的入口，安全性势必也是不可忽视的。对于安全性的方案大致有：

• 1：线上环境关闭API接口的文档，取消API接口的文档的地址

• 2：全局性的关闭API文档的所有接口，通过全局的include_in_schema来关闭

• 3：仅仅支持内网可见的API部署一个实例出来

• 4: 通过Nginx 启用 Base Auth 实现用户认证或对应的IP访问限制某API地址

• 5：APP应用上对请求来源IP做限制

• 6：App应用上也启用用户认知机制

补充App内自己用户和密码认知的一种简单的方式:

4.1 全局访问认证限制

import uvicorn

from fastapi import Depends
from fastapi import FastAPI, File, UploadFile
from config import settings

from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.exceptions import HTTPException
from fastapi.openapi.docs import get_swagger_ui_html
from starlette.status import HTTP_401_UNAUTHORIZED

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是于XXXXXX',
    version=settings.API_V1_STR,
    # 自定义开启的API文档的接口的时候，需要设置默认的关闭
    docs_url=None
)


security = HTTPBasic()


@app.get("/docs",include_in_schema=False)
async def get_documentation(credentials: HTTPBasicCredentials = Depends(security)):
    print("sssssssssssssssssss")
    if credentials.username != "xiaozhongtongxue" or credentials.password != "xiaozhongtongxue":
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail="y用户和密码不对！！！！！",
            headers={"WWW-Authenticate": "Basic"},
        )
    else:
        return get_swagger_ui_html(openapi_url="/openapi.json", title="doc")


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

PS:全局的模式，需要开启APP内的认证的话，需要设置默认的关闭docs_url=None

4.2 针对某接口进行认证限制

上面的方式是开启后，只要认证通过一次后，我就可以查看所有的，当然，如果再那个的基础再针对某个接口的限制的话，也是可以的，只需要我们对我们的需要限制的访问的接口做相关的依赖注入的限制就可以了！，但是这样感觉后续对API接的访问不合理。比较我们的接口是用于访问的！

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="y用户和密码不对！！！！！",
            headers={"WWW-Authenticate": "Basic"},
        )

    return credentials.username


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


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

复制代码

当需要查看某和接口的API文档的时候：

5 API响应报文类型

对于响应报文，通常我们的写API接口的肯定是一般返回json格式的报文体，但是也不排除我单纯只是需要返回字符串，或者返回HTML模板，返回的是数据流，返回XML格式等。

所以这里也需要有所针对性的进行定制，但是我们的Fastapi其实基本上都封装好了，不需要我们再自己去处理了.

PS如果存在自定义的错误参数的校验的时候，直接的返回字符串类似的是会报错的，必须使用PlainTextResponse才能正确返回。

如下示例：

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    # return JSONResponse({'mes':'触发了RequestValidationError错误，，错误信息:%s 你妹的错了！'%(str(exc))})
   # TypeError: 'str' object is not callable 
    return "小姐姐在哪里！"

需要返回使用 return PlainTextResponse（"小姐姐在哪里！"）才正确
复制代码

5.1 responses 类型

from fastapi.responses import (HTMLResponse, JSONResponse, ORJSONResponse, Response, PlainTextResponse, UJSONResponse,StreamingResponse,FileResponse)
复制代码

• Response 需要自己的再扩展的基础响应体

• HTMLResponse 返回HTML模板

• JSONResponse jison格式

• ORJSONResponse 使用了另一个库的JSON格式

• UJSONResponse 使用了另一个库的JSON格式

• StreamingResponse 输出流的格式

• FileResponse 文件响应体的方式

5.2 自定义Response

示例如返回XML格式响应:

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")
复制代码

5.3 返回字符串PlainTextResponse

示例如:

@app.get("/PlainTextResponse1/")
async def PlainTextResponse1():
    return PlainTextResponse('你好啊')


@app.get("/PlainTextResponse2/", response_class=PlainTextResponse)
async def PlainTextResponse2():
    return "你好！"

复制代码

5.4 返回JSON数据格式

示例如:

@app.post("/api/v1/udaoay/data")
async def uddata():
    return {"code": 0, "msg": "修改成功", "data": None}
    或
    return JSONResponse(status_code=404, content={"message": "Item not found"})

复制代码

5.5 返回HTMLResponse

示例如：

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()


def generate_html_response():
    html_content = """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)


@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return generate_html_response()
复制代码

5.6 返回RedirectResponse临时重定向

示例如：

@app.get("/RedirectResponse/", response_class=HTMLResponse)
async def RedirectResponse():
   return RedirectResponse("https://wwww.baifuw.com")
复制代码

5.7 返回流信息StreamingResponse

示例1如：接受异步生成器或普通生成器/迭代器，并对响应体进行流。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer()
复制代码

示例2如：如果您有一个类似文件的对象(例如，由open())，您可以在StreamingResponse.

这包括许多与云存储、视频处理等交互的库。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")
复制代码

5.8 返回FileResponse

异步地将文件作为响应流。

使用与其他响应类型不同的参数集实例化：

• path-要流的文件的文件路径。

• headers-任何要包括的自定义标头，作为字典。

• media_type-显示媒体类型的字符串。如果未设置，文件名或路径将用于推断媒体类型。

• filename-如果已设定，这将包括在答复中Content-Disposition.

文件响应将包括适当的Content-Length, Last-Modified和ETag标题。

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
async def main():
    return FileResponse(some_file_path)
复制代码

6 BaseSettings 配置文件读取

以往写Flask的时候，通常我们的会把一些配置信息，写到一个配置的文件里面，然后是根据测试环境还是正式环境的的方式来区分读取相关的配置值。

定义位置文件Congfig.py

from functools import lru_cache
import secrets
from pydantic import AnyHttpUrl, BaseSettings

import pprint

pp = pprint.PrettyPrinter(indent=4)


class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str = secrets.token_urlsafe(32)

    JWT_SECRET_KEY: str = ''
    JWT_ALGORITHM: str = ''
    JWT_EXPIRES: int = 60
    JWT_REFRESH_EXPIRES_DAYS: int = 1

    class Config:
        env_file = ".env"
        case_sensitive = True
        env_file_encoding = 'utf-8'


@lru_cache()
def get_settings():
    return Settings()


print("config.py > settings : ...")
settings = Settings()
pp.pprint(settings.dict())

复制代码

运行输出：

config.py > settings : ...
{   'API_V1_STR': '/api/v1',
    'JWT_ALGORITHM': '',
    'JWT_EXPIRES': 60,
    'JWT_REFRESH_EXPIRES_DAYS': 1,
    'JWT_SECRET_KEY': '',
    'SECRET_KEY': 'pGR3SgBuuMq-nbmjRTJHh-px4huaPpTpgQwnM5MDcik'}
复制代码

然后在我们App初始化的时候引入我们上面的单列对象settings

import uvicorn

from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query
from fastapi.responses import JSONResponse,ORJSONResponse,UJSONResponse,PlainTextResponse,HTMLResponse,StreamingResponse,RedirectResponse,FileResponse,Response
from fastapi.exceptions import RequestValidationError
from fastapi import FastAPI, File, UploadFile
from config import settings

app = FastAPI()


app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是于XXXXXX',
    version=settings.API_V1_STR
)

复制代码

7 Fastapi后台任务的补充（协程模式）

本来想补充到上一篇去，但是字数超了！这里再补充一下，后台的任务另一种方式（非线程的方式）：

7.1 线程模式的-后台任务

import uvicorn

from fastapi import FastAPI, File, UploadFile


app = FastAPI()

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def tasks(msg):
    print('耗时任务开始执行',msg)
    import time
    time.sleep(5)
    print('耗时任务开始结束', msg)


@app.post("/backgroundTasks")
async def runBackgroundTasks(msg: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(tasks,msg)
    return "任务已经再处理中！"

if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)


复制代码

7.2 协程模式的-后台任务

PS:协程的方式的后台，如果你的异步任务超级的需要耗时的，则不建议使用此方式了！比较是单线程的！

import uvicorn
from fastapi import  FastAPI, Request
import asyncio
import threading

app = FastAPI()


from fastapi.openapi.docs import get_swagger_ui_html
import time
# 后台任务
async def write_log():
    print(threading.current_thread().getName())
    # 模拟再协程使用同步的阻塞的方式，会发现有的时候回阻塞整个线程的执行！！！！
    time.sleep(5)

# 后台任务
async def write_log_2():
    print(threading.current_thread().getName())
    # 这里需要使用的异步的模拟阻塞的方式，所以    time.sleep(5)的区别很明显
    await asyncio.sleep(5)

# 用户请求
@app.get("/syncio/task")
async def do_task():
    asyncio.create_task(write_log())
    return "你好，协程下的同步阻塞！"

# 用户请求
@app.get("/asyncio/task")
async def do_task():
    asyncio.create_task(write_log_2())
    return "你好，协程下的异步阻塞！"

#定义路由，并且在方法中带上request: Request
@app.get("/request/msg")
def read_request(item_id: str, request: Request):
    scope = request.scope
    import json
    print(scope)
    client = request.client
    print(client)
    return get_swagger_ui_html()

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

复制代码

8 从Fastapi的request里提取客户端请求信息

首先之前我使用flask的时候，flask对应的request请求提里面是包含了相关多的信息，并且，我们所有请求的参数来源，也都是直接通过request来提取，其实这里我们的fastapi也同样可以提取相关的参数信息，甚至如果你想 在fastapi里引入其他校验库的话，通常我们就需要自己去request里拿数据。

那我们可以看一下我们的fastapi的Request可以拿到什么数据：

PS 区别于flask对应的request，fastapi的request需要再我们的路由上进程传递才可以！无法脱离路由的函数而独立存在。

Request的源码：

HTTPConnection的源码（省略一些初始化函数和其他魔法函数）：

从上面的源码可以看得到我们的Fastapi的request可以提取到的信息也是非常多滴：

• request.scope里可以提取的信息有：

  - 'type': 'http',
  -  'asgi': {'version': '3.0', 'spec_version': '2.1'}, 
  - 'http_version': '1.1', 
  -  'server': ('127.0.0.1', 8000),
  -  'client': ('127.0.0.1', 50863), 
  -  'scheme': 'http', 
  - 'method': 'GET', 
  -  'root_path': '', 
  -  'path': '/request/msg', 
  - 'raw_path': b'/request/msg', 
  -  'query_string': b'item_id=55555555', 
  -  'headers': [
     - (b'host', b'127.0.0.1:8000'),
     - (b'connection', b'keep-alive'), 
     - (b'accept', b'application/json'), 
     - (b'user-agent', b'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.25 Safari/537.36 Core/1.70.3870.400 QQBrowser/10.8.4405.400'),
      - (b'referer', b'http://127.0.0.1:8000/docs'), 
      - (b'accept-encoding', b'gzip, deflate, br'),
      - (b'accept-language', b'zh-CN,zh;q=0.9')]
  
    - 'fastapi_astack': <contextlib.AsyncExitStack object at 0x000001BE71FA19B0>, 'app': <fastapi.applications.FastAPI object at 0x000001BE71F91D68>,
    - 'router': <fastapi.routing.APIRouter object at 0x000001BE71FA1320>, 
    - 'endpoint': <function read_request at 0x000001BE71F76A60>, 
  
    - 'path_params': {}
复制代码

• type: 请求的协议是http还是https

• asgi: 使用的asgiweb服务的版本信息

• http_version 使用的的http协议版本

• server：请求服务端对应的IP和端口（TCP的四元组对象的必备）

• client：客户端来源的IP和端口（TCP的四元组对象的必备）

• scheme：客户端请的协议方案

• method：客户端请求提交的方式

• root_path：跟目录

• path：客户端请的路径地址

• raw_path：客户端请的路径地址（bytes）

• headers:客户端请求来源请求头的信息列表

• request.headers.get("Authorization")

• query_string：客户端请求提交的查询参数

• path_params：客户端请求提交的路径参数

• url :客户端的请的URL信息

• request.url.path

• request.url.port

• request.url.scheme

• base_url

• body 客户端如果叫的是字节数据，可以使用request.body() 获取

• json 客户端提交的如果是JSON格式的数据，可以直接的转为字段数据

• form 客户端提交的表单的数据

• cookies 客户端提交的cookies信息

• auth

• session

• client

总结

这是基础的只是第二篇，纯属个人的学习笔记，如有纰漏还希望多多指点。

后续打算针对安全和数据库这些都做专题的梳理。下一期再见！

END

小钟同学 | 文 【原创】【欢迎一起学习交流】| QQ：308711822