文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >专栏 >FastAPI后台开发基础（15）： 依赖注入

FastAPI后台开发基础（15）： 依赖注入

原创
作者头像
密码学人CipherHUB
修改于 2024-11-18 12:44:30
修改于 2024-11-18 12:44:30
9581
举报
文章被收录于专栏：编码视界编码视界

基础

在 FastAPI 中，依赖项可以是任何可调用的对象，如函数、类方法等。这些依赖项可以接受参数，这些参数同样可以是其他依赖项，从而形成依赖关系链。FastAPI 会自动处理这些依赖关系，确保在调用主功能（如API路由处理函数）之前，所有依赖项都已正确解析和调用。

使用依赖注入的优势

代码重用：通过依赖注入，可以在多个API端点中重用相同的功能或数据访问逻辑。

解耦：依赖注入有助于将应用程序的不同部分解耦，使其更容易管理和扩展。

易于测试：依赖注入使得单元测试更加简单，因为你可以轻松地为特定的依赖提供模拟（mock）或替代实现。

灵活性和可扩展性：可以根据需要轻松添加或修改依赖项，而不会影响到使用这些依赖项的主功能。

示例

代码语言：python
复制
def check_item_id(item_id: int, x: int) -> int:
    """
    check_item_id 函数有两个参数：item_id 和 x

    其中 item_id 是一个路径参数，因为它直接与路径操作 @app.get('/items/{item_id}') 中的 {item_id} 对应

    而 x 没有在路径中定义，也没有提供默认值，因此 FastAPI 会将其视为查询参数
    """
    print('check item id')
    if item_id > 100:
        return item_id * x
    else:
        raise HTTPException(status_code = 404, detail = "User not found")


@app.get('/items/{item_id}')
async def read_item(item_id: int = Depends(check_item_id)):
    print('read item')
    return {'item_id': item_id}
依赖注入示例
依赖注入示例

更多参数类型作为依赖注入函数的参数

代码语言：python
复制
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None


def query_extractor(depend_q: str = Query(..., min_length = 3)):
    return hashlib.sha256(depend_q.encode('utf-8')).hexdigest()


def path_extractor(depend_item_id: int = Path(..., gt = 0)):
    return hashlib.sha256(depend_item_id.to_bytes(4, 'big', signed = True)).hexdigest()


def body_extractor(depend_item: Item = Body(...)):
    return hashlib.sha256(depend_item.model_dump_json().encode('utf-8')).hexdigest()


@app.post("/items/{depend_item_id}")
def read_item_2(query: str = Depends(query_extractor),
                depend_item_id: int = Depends(path_extractor),
                item: Item = Depends(body_extractor)):
    """
    :param query: 查询参数作为query_extractor的入参，请求时的参数需要为depend_q

    :param depend_item_id: 路径参数作为path_extractor的入参，请求时的参数需要为depend_item_id

    :param item: Body 参数作为body_extractor的入参，请求时的参数需要为depend_item

    请求:
    curl -X 'POST' \
    'http://127.0.0.1:18081/items/12334?depend_q=hello%2Cworld' \
    -H 'accept: application/json' \
    -H 'Content-Type: application/json' \
    -d '{
        "name": "测试",
        "description": "描述信息",
        "price": 12.34,
        "tax": 56.78
    }'

    响应:
    {
        "item_id": "17c73e617248212e4adc5e83c87e20e51df159532284996a19c28e15eac3326f",
        "query": "77df263f49123356d28a4a8715d25bf5b980beeeb503cab46ea61ac9f3320eda",
        "item": "4765dec528b96cb04e46f0dd6b039b8dad29c797e4a6e4455cd244b7062e1009"
    }
    """
    return {"item_id": depend_item_id, "query": query, "item": item}
多种参数类型的依赖注入
多种参数类型的依赖注入
运行效果
运行效果

常见的 FastAPI 参数作为依赖注入函数的参数

代码语言：python
复制
class User(BaseModel):
    username: str
    full_name: str = None


# 依赖函数使用路径参数
def extract_item_id(item_id: int = Path(...)):
    return item_id


# 依赖函数使用查询参数
def pagination(skip: int = Query(0, ge = 0), limit: int = Query(10, ge = 0)):
    return {"skip": skip, "limit": limit}


# 依赖函数使用Header参数
def get_user_agent(user_agent: str = Header(...)):
    return user_agent


# 依赖函数使用Cookie参数
def get_session_id(session_id: str = Cookie(None)):
    return session_id


# 依赖函数使用Body参数
def user_data(user: User = Body(...)):
    return user


@app.get("/items/{item_id}")
def read_item(item_id: int = Depends(extract_item_id),
              pagination_params: dict = Depends(pagination),
              user_agent_str: str = Depends(get_user_agent),
              session: str = Depends(get_session_id)):
    return {
        "item_id": item_id,
        "pagination": pagination_params,
        "user_agent": user_agent_str,
        "session_id": session
    }


@app.post("/users/")
def create_user(user: User = Depends(user_data)):
    return {"username": user.username, "full_name": user.full_name}

依赖注入函数运行的流程

依赖注入函数运行的流程
依赖注入函数运行的流程

依赖树

在 FastAPI 中，依赖树是指依赖项之间的层次结构或关系网，其中一个依赖项可以依赖于另一个依赖项，形成一个树状结构。这种结构允许开发者构建复杂而高效的应用程序，通过将逻辑和功能分解成更小、更可管理的部分来提高代码的可维护性和重用性。

代码语言：python
复制
# 模拟的用户数据库
users_db = {
    "alice": {"username": "alice", "role": "admin"},
    "bob": {"username": "bob", "role": "user"}
}


class User:
    def __init__(self, username: str, role: str):
        self.username = username
        self.role = role


def get_current_user(username: str):
    if username in users_db:
        user_data = users_db[username]
        return User(username = user_data["username"], role = user_data["role"])
    else:
        raise HTTPException(status_code = 404, detail = "User not found")


def verify_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code = 403, detail = "Admin privileges required")
    return user


@app.get("/admin-data/")
def admin_data(user: User = Depends(verify_admin)):
    """
    GET /admin-data/：这个路径操作依赖于 verify_admin 函数

    verify_admin 函数本身依赖于 get_current_user 函数来获取当前用户

    verify_admin 函数是 admin_data 路径操作的直接依赖，而 get_current_user 是 verify_admin 的子依赖

    这种方式允许你构建复杂的依赖关系，每个依赖项负责处理特定的任务，从而保持代码的清晰和模块化
    """
    return {"message": "Admin data accessed", "user": user.username}


def get_user_id(test_id: int = Path(..., alias = "aliased_user_id")):
    # 这里的 id 对应于路径中的 user_id
    return hashlib.sha3_512(test_id.to_bytes(8, 'big', signed = True)).hexdigest()


@app.get("/users/{aliased_user_id}")
def read_user(user_id: int = Depends(get_user_id)):
    """
    Path 和 Query 的参数名不必总是与 Depends 函数中的参数名保持一致

    可以通过在 Path 和 Query 的定义中使用 alias 参数来指定如何从请求中提取这些值，并将它们映射到依赖函数的参数上

    在这段代码中，路径参数 aliased_user_id 与 get_user_id 函数中的 test_id 参数

    通过 alias = "aliased_user_id" 保持一致的
    """
    return {"user_id": user_id}

在上面的代码示例中，我们可以看到一个依赖树的实际应用：

  • 顶层依赖：API 路径操作函数（如 admin_data 和 read_user）
  • 中间层依赖：这些是被路径操作直接依赖的函数（如 verify_admin 和 get_user_id）
  • 底层依赖：这些依赖项被中间层依赖所依赖（如 get_current_user）

以 /admin-data/ 路径操作为例，我们可以看到以下依赖关系：

  • admin_data 函数 依赖于 verify_admin 函数。
  • verify_admin 函数 依赖于 get_current_user 函数。

这里，get_current_user 是底层依赖，它直接与数据交互（从模拟的用户数据库中获取用户信息）。verify_admin 是中间层依赖，它使用 get_current_user 的输出来验证用户是否具有管理员权限。最后，admin_data 是顶层依赖，它依赖于 verify_admin 的验证结果来决定是否允许用户访问特定的数据。

依赖树的优势

  • 模块化：每个依赖项负责处理特定的任务，使得代码更加模块化。
  • 代码重用：依赖项可以在多个地方重用，减少代码重复。
  • 易于测试：每个依赖项可以独立测试，简化了单元测试的过程。
  • 清晰的责任分离：每个依赖项只关注自己的责任范围，提高了代码的清晰度和可维护性。

其他用法

使用类作为依赖注入参数

代码语言：python
复制
class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    """
     Class-Based Dependency Injection

     FastAPI will:
        Create an instance of CommonQueryParams
        Extract query parameters from the request
        Pass them to the class constructor
        Inject the resulting instance into your endpoint

     Key benefits of class-based dependencies:
        Encapsulation: Group related parameters and methods
        Reusability: Use the same dependency in multiple endpoints
        Maintainability: Organize code in a more structured way
        Testability: Easier to mock and test
        Hierarchical Dependencies: Classes can depend on other classes
    """
    response = {}
    class_json_str = json.dumps(commons, ensure_ascii = False, default = lambda o: o.__dict__)
    h_str = hashlib.sha256(class_json_str.encode('utf-8')).hexdigest()
    response.update({"sha256_str": h_str})
    return response
Class-Based Dependency Injection
Class-Based Dependency Injection

路径操作中的多个依赖项

代码语言：python
复制
async def verify_token(x_token: Annotated[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: Annotated[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():
    """
    Header-Based Authentication using multiple dependencies
    """
    return [{"item": "Foo"}, {"item": "Bar"}]
Header-Based Authentication using multiple dependencies
Header-Based Authentication using multiple dependencies

全局依赖注入

代码语言：python
复制
from __future__ import annotations

import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated


async def verify_token(x_token: Annotated[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: Annotated[str, Header()]):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code = 400, detail = "X-Key header invalid")
    return x_key


"""
Global Dependencies
"""
app = FastAPI(dependencies = [Depends(verify_token), Depends(verify_key)])


@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


if __name__ == '__main__':
    uvicorn.run(app, host = '127.0.0.1', port = 18081)
Global Dependencies
Global Dependencies
Global Dependencies
Global Dependencies

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

fastapi

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

fastapi
#FastAPI
#PyDantic
#Python3
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
目录
  • 基础
    • 使用依赖注入的优势
    • 示例
    • 更多参数类型作为依赖注入函数的参数
    • 常见的 FastAPI 参数作为依赖注入函数的参数
    • 依赖注入函数运行的流程
  • 依赖树
  • 其他用法
    • 使用类作为依赖注入参数
    • 路径操作中的多个依赖项
    • 全局依赖注入
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2026 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论
1