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)
查看 Swagger API 文档
正确传参的请求结果
