章工运维 章工运维
首页
  • linux
  • windows
  • 中间件
  • 监控
  • 网络
  • 存储
  • 安全
  • 防火墙
  • 数据库
  • 系统
  • docker
  • 运维工具
  • other
  • elk
  • K8S
  • ansible
  • Jenkins
  • GitLabCI_CD
  • 随笔
  • 面试
  • 工具
  • 收藏夹
  • Shell
  • python
  • golang
友链
  • 索引

    • 分类
    • 标签
    • 归档
    • 首页 (opens new window)
    • 关于我 (opens new window)
    • 图床 (opens new window)
    • 评论 (opens new window)
    • 导航栏 (opens new window)
周刊
GitHub (opens new window)

章工运维

业精于勤,荒于嬉
首页
  • linux
  • windows
  • 中间件
  • 监控
  • 网络
  • 存储
  • 安全
  • 防火墙
  • 数据库
  • 系统
  • docker
  • 运维工具
  • other
  • elk
  • K8S
  • ansible
  • Jenkins
  • GitLabCI_CD
  • 随笔
  • 面试
  • 工具
  • 收藏夹
  • Shell
  • python
  • golang
友链
  • 索引

    • 分类
    • 标签
    • 归档
    • 首页 (opens new window)
    • 关于我 (opens new window)
    • 图床 (opens new window)
    • 评论 (opens new window)
    • 导航栏 (opens new window)
周刊
GitHub (opens new window)
  • python

    • python基础

    • FastAPI

      • FastAPI-快速入门1
      • FastAPI-路由2
      • FastApi-参数提交3
      • FastApi-响应报文
      • FastApi-错误处理
      • FastApi-中间件
      • FastApi-跨域处理
      • 依赖注入之Depends
        • 简单的依赖说明
        • 把类当作被依赖对象
        • 多层嵌套依赖
        • list列表依赖
        • 多依赖对象注入和列表其实是一样的:
    • python每日练习脚本

    • python3给防火墙添加放行
    • python生成部署脚本
    • python将多个文件内容输出到一个文件中
    • 使用 Aligo 定时备份服务器文件
    • python监控日志文件并发送钉钉告警
    • python监控数据库脚本并发送钉钉告警
    • 使用python编写自动化发布脚本
    • 查询redis列表某个元素
    • centos7安装python3
    • python环境管理工具介绍
    • conda安装和镜像源配置
    • pip更换国内源
    • python爬虫
    • python环境启动服务报错缺少glibc库版本
    • 监控目录或文件变化
    • 批量更改文件
    • python引用数据库
  • shell

  • go

  • 编程
  • python
  • FastAPI
章工运维
2024-12-04
目录

依赖注入之Depends

# 一、概述

看官网的描述Depends的使用,似乎比较懵懵懂懂的,于是乎还是需要花些时间再次学习一下关于依赖注入。

首先依赖注入它可以是函数也可以是类,如下面的函数形式的依赖注入:

# 简单的依赖说明

import uvicorn
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}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    commons.update({'小钟': '同学'})
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

if __name__ == '__main__':
    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

梳理一下接口请求的流程:

  • 1: 上面的commons: dict = Depends(common_parameters)它声明了一个依赖关系: Depends(common_parameters): 这对接口的依赖进行了一个声明,表示的是接口参数请求依赖于common_parameters的函数。

    当接口被调用的时候,回调用common_parameters函数进行请求处理。

  • 2: common_parameters函数主要是负责接收函数,处理后返回一个字典,

  • 3:然后把Depends(common_parameters)返回的结果 传递给commons: dict,这个就是一个依赖注入的过程。

所以在上面的示例中common_parameters是我们被依赖对象

这个被依赖的对象,对接口请求的要求就是:

  • 可选查询参数q那是一个str.

  • 可选查询参数skip那是int,默认情况下是0.

  • 可选查询参数limit那是int,默认情况下是100.

  • 返回一个字典

请求示例:

http://127.0.0.1:8000/items/

e1e9889b6a05ddca.png

这依赖注入的方式其实也挺方便,类似于接口装饰器的方式,比如common_parameters中我们可以先对相关的参数进行校验拦截,还可以再传递。 场景可以和我们之前的bottle的装饰器差不多类似:

  • 相同的逻辑判断处理

  • 用户身份鉴权

# 把类当作被依赖对象

上面我们的被依赖的对象是以函数的形式出现,那FastAPI它也支持以类的形式来表达。按官网说法被依赖对象它应该是必须一个可以调用的对象比如:类,函数之类的···

这里看一下以类的形式:

import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()

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

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    # 如果q存在
    if commons.q:
        # 我们就把q加到一个新字典
        response.update({"q": commons.q})
        response.update({"小钟": '同学'})
    # 然后在我们的fake_items_db进行截取
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response

if __name__ == '__main__':
    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

上面我们的CommonQueryParams是一个类,它和我的函数其实差不多,当我们的接口被调用的时候,类对象就回被初始化, 按官网的说法: commons: CommonQueryParams = Depends(CommonQueryParams) 和 commons = Depends(CommonQueryParams) 是等价的。 还有一种是: commons: CommonQueryParams = Depends()

示例运行演示

有Q参数:

http://127.0.0.1:8000/items/?q=2323

4f6585a26fdd5d7d.png

没有Q参数:

http://127.0.0.1:8000/items/

8a65394ce7390cb5.png

# 多层嵌套依赖

多层嵌套的意思就是可以类可以类的意思。函数可以依赖函数。其实和我们的之前的参数校验一样。

比如下面的代码:

import uvicorn
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: str = None):
    return q

def query_or_cookie_extractor(
        q: str = Depends(query_extractor), last_query: str = Cookie(None)
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

if __name__ == '__main__':
    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

其实意思就是query_or_cookie_extractor 依赖于query_extractor,然后query_or_cookie_extractor被注入到接口上也被依赖的对象。

官网的截图上面接口执行流程如下:

c667e42be47e0c93.png

对于同一个依赖,如果处理的结果是一样的,就是返回值是一样的话,我们可以进行多次调用依赖,这时候可以对被依赖的对象设置是否使用缓存机制:

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):  return {"fresh_value": fresh_value}
1

请求一下页面

http://127.0.0.1:8000/items/

2a3f0801b78ec393.png

由于没有cookie,返回空

# list列表依赖

我们先看官方提供的示例代码:

import uvicorn
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"}]

if __name__ == '__main__':
    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

上述的代码意思是对我们的请求头部信息Header进行验证,因为示例是...三个点,说明是必选的字段:

分析上述的代码之后,运行一下试一试看看结果:

1:什么头部参数都不传递的情况提示,我们的头部参数异常

http://127.0.0.1:8000/items/

c6b41c0528b7a27a.png

2:头部参数填写的情况:

注意点:参数提交的格式,因为是头部的参数,所以我们的代码上的x_token 会应该要写:x-token才对

dc88af763463abee.png

错误的示例:

da9c9672b91be58a.png

所以上面列表的依赖的意思就是必须两天条件都成立才通过。这个感觉后期还是用到的比较多的哟!

# 多依赖对象注入和列表其实是一样的:

import uvicorn

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi import Depends, FastAPI

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")
    return x_token


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"}]

@app.get("/items2/")
async def items2(xt: str = Depends(verify_token), xk: str = Depends(verify_key)):
    return {"xt": xt, 'xk': xk}

if __name__ == '__main__':
    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

如上面的xt: str = Depends(verify_token),xk: str = Depends(verify_key),也是需要必须两个条件成立才可以。

正常情况:

x-token=fake-super-secret-token x-key=fake-super-secret-key

80c933121979702f.png

非正常情况:

x-token=fake-super-secret-token x-key=fake-super-secret-key--fei

90ef15c66a32ee56.png


原文链接 (opens new window)

微信 支付宝
上次更新: 2024/12/13, 12:36:19

← FastApi-跨域处理 监控系统资源情况并发送邮件告警→

最近更新
01
shell脚本模块集合
05-13
02
生活小技巧(认知版)
04-29
03
生活小技巧(防骗版)
04-29
更多文章>
Theme by Vdoing | Copyright © 2019-2025 | 点击查看十年之约 | 鄂ICP备2024072800号
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式