FastAPI（29）- Dependencies 依赖注入的初步使用

小菠萝测试笔记

发布于 2021-09-27 16:07:29

2K00

代码可运行

文章被收录于专栏：自动化、性能测试自动化、性能测试

运行总次数：0

代码可运行

FastAPI 的依赖注入

FastAPI 有一个非常强大但直观的依赖注入系统
它被设计为非常易于使用，并且使任何开发人员都可以非常轻松地将其他组件与 FastAPI 集成

什么是依赖注入

在编程中，为保证代码成功运行，先导入或声明其所需要的【依赖】，如子函数、数据库连接等等
它和钩子函数非常相似

依赖注入有什么作用

业务逻辑复用的场景使用，可以减少重复代码
共享数据库连接
强制执行安全性、身份验证、角色管理等
其他使用场景

FastAPI 的兼容性

依赖注入系统的简单性使得 FastAPI 兼容：

所有的关系型数据库
NoSQL 数据库
第三方的包和 API
认证、授权系统
响应数据注入系统

依赖注入的简单栗子

第一步：创建依赖项

from typing import Optional

# 1、编写依赖项
async def common_parameters(q: Optional[str] = None,
                            skip: int = 0,
                            limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

它就是一个普通函数，只不过加了 async 异步关键字
函数参数有三个，都是可选的
依赖项可以返回任意内容，这里的返回值是一个 dict，把传进来的值塞进 dict，再进行返回

第二步：导入 Depends

from typing import Optional
# 2、导入 Depends
from fastapi import Depends


# 1、编写依赖项函数
async def common_parameters(q: Optional[str] = None,
                            skip: int = 0,
                            limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

第三步：声明 Depends，完成依赖注入

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠萝测试笔记
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/24 1:08 下午
# file: 25_dependency.py
"""
from typing import Optional, Dict, Any
# 2、导入 Depends
from fastapi import Depends, FastAPI

import uvicorn

app = FastAPI()


# 1、编写依赖项函数
async def common_parameters(q: Optional[str] = None,
                            skip: int = 0,
                            limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


# 3、编写路径操作函数，参数声明为 Depends
@app.get("/items")
async def read_items(commons: Dict[str, Any] = Depends(common_parameters)):
    return commons


if __name__ == "__main__":
    uvicorn.run(app="25_dependency:app", host="127.0.0.1", port=8080, reload=True, debug=True)

commons 参数声明了 Depends() ，和 Body()、Query() 的使用方式一样
注意：填写依赖项的时候不需要加 ()，只写函数名就行，且填写的依赖项必须是一个函数

Depends(common_parameters)

当请求 /items 时，FastAPI 会做哪些事？

通过正确的参数调用依赖项函数 common_parameters
从依赖项函数中获取 return 值
将返回值赋值给路径操作函数中的参数 commons
执行完依赖项函数后，才会执行路径操作函数

解析 commons: Dict[str, Any] = Depends(common_parameters)

虽然 commons 参数类型声明为 Dict，但请求 /items 的时候并不是通过 Request Body 来传参的
因为 commons 参数接收的是依赖项函数 common_parameters 的返回值，所以这个 Dict 是限制了依赖项函数的返回值类型，并不是请求数据类型
/items 的请求传参方式是查询参数

正确传参的请求结果

直接在 Swagger API 文档上测试了，顺便看看长啥样

从文档中也可以看到，/items 要传三个查询参数，其实就是依赖项函数的参数
FastAPI 会将所有依赖项信息添加到 OpenAPI Schema 中，以便在 Swagger API 中显示（如上图）

请求数据验证失败的请求结果

即使是依赖项，FastAPI 也会对它做数据验证，不符合类型则报错

async 或 not async

可以在非 async 路径操作函数中使用 async 的依赖项
也可以在 async 路径操作函数中使用非 async 的依赖项
FastAPI 知道要怎么处理

# 非 async 依赖项
def test_dep(name: str):
    return name


# async 路径操作函数
@app.get("/name")
async def test_name(name: str = Depends(test_dep)):
    return name


# async 依赖项
async def test_user_info(info: dict):
    return info


# 非 async 路径操作函数
@app.get("/info")
def test_info(info: dict = Depends(test_user_info)):
    return info

依赖项函数参数类型是一个字典

from typing import Optional, Dict, Any
# 2、导入 Depends
from fastapi import Depends, FastAPI

import uvicorn

app = FastAPI()


# 1、编写依赖项
async def common_parameters(*,
                            q: Optional[str] = None,
                            skip: int = 0,
                            limit: int = 100,
                            # 新增一个 Dict 类型的参数
                            info: Dict[str, Any]):
    return {"q": q, "skip": skip, "limit": limit, "info": info}


# 3、编写路径操作函数，参数声明为 Depends
@app.get("/items")
async def read_items(commons: Dict[str, Any] = Depends(common_parameters)):
    return commons


if __name__ == "__main__":
    uvicorn.run(app="25_dependency:app", host="127.0.0.1", port=8080, reload=True, debug=True)