FastAPI 依赖注入

什么是依赖注入？, Depends

声明代码（本文中为路径操作函数 ）运行所需的，或要使用的「依赖」的一种方式。

依赖注入常用于以下场景：

• 共享业务逻辑（复用相同的代码逻辑）

• 共享数据库连接

• 实现安全、验证、角色权限

• 等……

• 使用依赖注入，将代码重复最小化。

示例一

# coding: utf8

from fastapi import FastAPI, Depends
from typing import Optional

app = FastAPI()

# 依赖，类似公共组件
async def common_args(
        q: Optional[str] = None,
        skip: int = 0,
        limit: int = 100
    ):
    return {"q": q, "skip": skip, "limit": limit}

@app.get('/items/')
async def get_items(commons: dict = Depends(common_args)):
    return commons

@app.get('/users/')
async def get_users(commons: dict = Depends(common_args)):
    return commons

# # 启动: uvicorn main:app --reload

示例二

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"city": "beijing"}, {"city": "shanghai"},
                 {"city": "heze"}]


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

@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
    response = {}
    if commons.desc:
        response.update({"desc": commons.desc})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

嵌套依赖

from fastapi import Depends, FastAPI
from  typing import Optional

app = FastAPI()

fake_items_db = [{"city": "beijing"}, {"city": "shanghai"},
                 {"city": "heze"}]

def query_extractor(desc: Optional[str] = None):
    return desc

def query__extractor(
    desc: str = Depends(query_extractor),
        name: Optional[str]="" ,
):
    if not desc:
        return name
   return desc

@app.get("/items/")
def read_items(query__extractor: str = Depends(query__extractor)):

    return query__extractor

路径操作装饰器依赖项

• X-Token 存在且必须等于 token123,

• Host 存在且必须等于 google.com

# coding: utf8

from fastapi import FastAPI, Depends, Header, HTTPException,status
from typing import Optional

app = FastAPI()


def verify_token(token: str= Header(..., alias="X-Token")):
    if token != 'token123':
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Token header invalid')

def verify_host(host: str = Header(..., alias='Host')):
    if host != 'google.com':
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Host header invalid')

@app.get("/items", dependencies=[Depends(verify_token), Depends(verify_host)])
async def read_items():
    return {"code": status.HTTP_200_OK}

# # 启动: uvicorn main:app --reload

全局依赖

• 比如我们全局都需要校验token

# coding: utf8

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



def verify_token(token: str= Header(..., alias="X-Token")):
    if token != 'token123':
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Token header invalid')

app = FastAPI(dependencies=Depends(verify_token))

@app.get("/items")
async def read_items():
    return {"code": status.HTTP_200_OK}

# # 启动: uvicorn main:app --reload

FastAPI 兼容性

依赖注入系统如此简洁的特性，让 FastAPI 可以与下列系统兼容：

• 关系型数据库

• NoSQL 数据库

• 外部支持库

• 外部 API

• 认证和鉴权系统

• API 使用监控系统

• 响应数据注入系统