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 会做哪些事？

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

 

解析 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)

 

查看 Swagger API 文档

 

正确传参的请求结果