今天就跟大家聊聊有关python fastapi中依赖注入系统的工作原理以及使用方法，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。
创新互联公司专注于工农企业网站建设,响应式网站设计,电子商务商城网站建设。工农网站建设公司,为工农等地区提供建站服务。全流程按需定制制作，专业设计，全程项目跟踪，创新互联公司专业和态度为您提供的服务
在 fastapi 中有一个强大并且简单易懂的系统：依赖注入系统
1
什么是依赖注入
依赖注入首先意味着在程序中我们的代码可以声明一些它必须依赖的项：我们称之为 dependencies，也就是依赖项。然后，在实际运行中，fastapi 会把所有需要的依赖项提供给你的代码，称之为注入依赖项。
一句话简单解释:原本接受各种参数来构造一个对象，现在只接受一个参数这个参数是已经实例化的对象，对象的『依赖』是注入进来的，而和它的构造方式解耦了。构造它这个『控制』操作也交给了第三方。
1.1
依赖注入适用场景列举：
1.业务逻辑复用
2.共享数据库连接
3.安全机制、权限校验、角色管理等
所有上述使用场景，借助于依赖注入可提高代码复用率，减少代码重复。
2
依赖注入实现方案
在看具体的实现依赖注入之前我们先从流程上来理解一下整个数据流转。当一个新的请求到来的时候，实际的调用流程如下:
1.调用依赖项函数(传递合适的参数)
2.得到依赖项目函数的返回结果
3.把返回结果传递给路由函数中对应的参数
4.路由函数中业务流数据处理
5.获取的数据返回给客户端/前端
2.1
函数级依赖项
# -*- encoding: utf-8 -*-from fastapi import Depends, FastAPIapp = FastAPI()async def common_paras(request_source: str = None, page: int = 0, limit: int = 100): return {"request_source": request_source, "page": page, "limit": limit}@app.get("/items/info")async def read_items_info(commons: dict = Depends(common_paras)): return commons@app.get("/users/info")async def read_users_info(commons: dict = Depends(common_paras)): return commons2.2
类级依赖项
# -*- encoding: utf-8 -*-from fastapi import Depends, FastAPIapp = FastAPI()fake_items_db = [{"user_name": "HaiShiNiu"}, {"user_name": "WangXiaoXiao"}, {"user_name": "ZhongCen"}]class CommonQueryParams: def __init__(self, request_source: str = None, page: int = 0, limit: int = 100): self.request_source = request_source self.page = page self.limit = limit@app.get("/items/info")async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): response = {} if commons.request_source: response.update({"q": commons.q}) items = fake_items_db[commons.page : commons.page + commons.limit] response.update({"items": items}) return response2.3
递归依赖项
# -*- encoding: utf-8 -*-from fastapi import Cookie, Depends, FastAPIapp = FastAPI()def query_extractor(q11: str, q12: str): return q11 + q12def query_or_cookie_extractor(q2: str = Depends(query_extractor), last_query: str = Cookie(None)): if not q2: return last_query return q2@app.get("/items/")async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}注意:有些场景是需要对依赖项的多次调用的情况。
如果某个依赖项在同一个路径操作中被声明了多次，例如，多个依赖项都有一个共同的子依赖项，那么 fastapi 默认在每一次请求中只会调用这个依赖项一次。fastapi 会把这个依赖项的返回值缓存起来，然后把这个值传递给需要的依赖项，而不是在同一个请求中多次调用这个依赖项。在有些场景下，我们并不需要缓存这个依赖项的返回值，而是需要多次调用，那么我们可以使用参数 use_cache=False 来禁止依赖项的缓存。
async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)): return {"fresh_value": fresh_value}2.4
基于路径操作装饰器依赖项
在有些情况下，我们并不需要依赖项的返回值，但仍然需要依赖项被执行。在这种情况下，我们可以通过路径操作装饰器来操作依赖项的一个列表。路径操作装饰器接收一个可选的参数 dependencies，参数内容是 Depends()列表。
# -*- encoding: utf-8 -*-from fastapi import Depends, FastAPI, Header, HTTPExceptionapp = FastAPI()async def verify_token(x_token: str = Header(...)): if x_token != "token": raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header(...)): if x_key != "key": raise HTTPException(status_code=400, detail="X-Key header invalid") return x_key@app.get("/items/info", dependencies=[Depends(verify_token), Depends(verify_key)])async def read_items_info(): return [{"function_name": "Foo"}, {"function_name": "Bar"}]这些依赖项与普通依赖项的执行相同，但他们的返回值(如果有)不会被传递给路径操作函数。我们可以重复使用已经声明的依赖项，无论他们是否有返回值，都不会影响依赖项的执行。
2.5
带有 yield 功能依赖项
fastapi 支持依赖项在请求结束后做一些额外的工作。要实现这个功能，我们需要用 yield 代替 return，然后其后添加一些额外的工作。我们举个例子:创建一个数据库链接，然后在请求结束后关闭这个链接。当然处理这个 case 的解决方案也是有很多的，本次我们就基于使用 yield 功能的依赖项进行实现
# -*- encoding: utf-8 -*-async def get_db_info(): """ 获取数据库连接信息 """ # 创建链接 db = DBSession() try: yield db # 数据库使用 finally: db.close() # 关闭数据库链接
简单说明一下:yield db 后面的值 db 会注入给路径操作或者其他依赖项。yield db 后面的代码在 response 提交之后才会执行。使用 try 语句来捕获可能发生的异常。为了确保无论是否有异常发生都能执行退出逻辑，我们这里在 finally 语句中执行退出逻辑。但是要注意，如果尝试在 yield 后面抛出 HTTPException，不会起到任何作用。yield 之后的退出代码是在异常处理器之后被执行的，因此无法捕捉异常的发生。
2.6
可参数化依赖项
我们前面使用的依赖项都是固定的函数或者类，但有时想在依赖项中设置不同的参数，同时又不用声明不同的函数或类。此时可利用一个可调用的类实例来实现这个功能。注意:类本身就是可调用的，而它的实例需要实现一个特定类方法才是可调用的：call
一句话介绍 call：call 是 Python 的一个黑魔法方法，核心功能是通过对象实例可以直接触发 call 中的逻辑，在 Tornado 源码中这种用法很多。
# -*- encoding: utf-8 -*-from fastapi import Depends, FastAPIapp = FastAPI()class FixedContentQueryCheck(object): def __init__(self, fixed_content: str): self.fixed_content = fixed_content def __call__(self, q: str = ""): if q: return self.fixed_content in q return Falsechecker = FixedContentQueryCheck("bar")@app.get("/query-checker/")async def read_query_check(fixed_content_included: bool = Depends(checker)): return {"fixed_content_in_query": fixed_content_included}我们系统的梳理了实现依赖注入的各种方式供大家在生产环境中进行使用
看完上述内容，你们对python fastapi中依赖注入系统的工作原理以及使用方法有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注创新互联行业资讯频道，感谢大家的支持。
