跳到内容

类作为依赖项

在深入研究依赖注入系统之前,让我们先升级一下上一个示例。

来自上一个示例的 dict

在上一个示例中,我们从依赖项(“可依赖对象”)中返回了一个 dict

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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)):
    return commons


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

但随后我们在路径操作函数的参数 commons 中得到了一个 dict

我们知道,编辑器无法为 dict 提供太多的支持(例如自动补全),因为它们无法预知其中的键和值类型。

我们可以做得更好……

是什么构成了依赖项

到目前为止,你看到的依赖项都是以函数形式声明的。

但这并不是声明依赖项的唯一方式(尽管这可能是最常见的方式)。

关键因素在于,依赖项必须是一个“可调用对象”(callable)。

Python 中的“可调用对象”是指任何 Python 可以像函数一样“调用”的东西。

所以,如果你有一个对象 something(它可能不是一个函数),并且你可以像这样“调用”(执行)它:

something()

或者

something(some_argument, some_keyword_argument="foo")

那么它就是一个“可调用对象”。

类作为依赖项

你可能注意到,创建 Python 类的实例时,使用的语法也是一样的。

例如

class Cat:
    def __init__(self, name: str):
        self.name = name


fluffy = Cat(name="Mr Fluffy")

在这种情况下,fluffyCat 类的一个实例。

而为了创建 fluffy,你正在“调用” Cat

所以,Python 类也是一个可调用对象

因此,在 FastAPI 中,你可以将 Python 类用作依赖项。

FastAPI 实际检查的是它是否是一个“可调用对象”(函数、类或其他任何东西)及其定义的参数。

如果你在 FastAPI 中将一个“可调用对象”作为依赖项传入,它将分析该“可调用对象”的参数,并以与路径操作函数参数相同的方式处理它们。包括子依赖项。

这也适用于没有任何参数的可调用对象。就像没有任何参数的路径操作函数一样。

那么,我们可以将上述依赖项“可依赖对象” common_parameters 更改为 CommonQueryParams 类。

from typing import Annotated

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 = 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)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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 = 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 = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

注意用于创建类实例的 __init__ 方法

from typing import Annotated

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 = 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)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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 = 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 = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

...它拥有与我们之前的 common_parameters 相同的参数。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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)):
    return commons


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

这些参数正是 FastAPI 将用于“解析”依赖项的依据。

在这两种情况下,它都将包含:

  • 一个可选的 q 查询参数,类型为 str
  • 一个 skip 查询参数,类型为 int,默认值为 0
  • 一个 limit 查询参数,类型为 int,默认值为 100

在这两种情况下,数据都会被转换、校验,并记录在 OpenAPI 模式中等等。

使用它

现在你可以使用这个类来声明你的依赖项。

from typing import Annotated

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 = 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)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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 = 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 = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

FastAPI 会调用 CommonQueryParams 类。这会创建一个该类的“实例”,并且该实例将作为 commons 参数传递给你的函数。

类型注解 vs Depends

注意在上面的代码中我们是如何两次编写 CommonQueryParams 的:

commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]

提示

如果可能,请优先使用 Annotated 版本。

commons: CommonQueryParams = Depends(CommonQueryParams)

最后的 CommonQueryParams,位于

... Depends(CommonQueryParams)

...正是 FastAPI 实际用来识别依赖项的内容。

FastAPI 正是从这里提取声明的参数,并实际调用它。


在这种情况下,第一个 CommonQueryParams,位于

commons: Annotated[CommonQueryParams, ...

提示

如果可能,请优先使用 Annotated 版本。

commons: CommonQueryParams ...

...对 FastAPI 没有任何特殊意义。FastAPI 不会将其用于数据转换、校验等(因为它使用的是 Depends(CommonQueryParams) 来完成这些工作)。

你实际上可以直接写

commons: Annotated[Any, Depends(CommonQueryParams)]

提示

如果可能,请优先使用 Annotated 版本。

commons = Depends(CommonQueryParams)

...就像在

from typing import Annotated, Any

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 = 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[Any, Depends(CommonQueryParams)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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 = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


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

但我们鼓励声明类型,因为这样你的编辑器就能知道传递给 commons 参数的内容,从而为你提供代码补全、类型检查等帮助。

快捷方式

但你可以看到这里存在一些代码重复,即写了两次 CommonQueryParams

commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]

提示

如果可能,请优先使用 Annotated 版本。

commons: CommonQueryParams = Depends(CommonQueryParams)

FastAPI 为这些情况提供了一个快捷方式,即当依赖项专门是一个类,且 FastAPI 会“调用”该类来创建其自身实例时。

对于这些特定情况,你可以执行以下操作:

与其写:

commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]

提示

如果可能,请优先使用 Annotated 版本。

commons: CommonQueryParams = Depends(CommonQueryParams)

...不如写:

commons: Annotated[CommonQueryParams, Depends()]

提示

如果可能,请优先使用 Annotated 版本。

commons: CommonQueryParams = Depends()

你将依赖项声明为参数的类型,并使用不带任何参数的 Depends(),而不必在 Depends(CommonQueryParams) 内部再次完整地写出类名。

同样的示例看起来就是:

from typing import Annotated

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 = 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()]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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 = 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()):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

...这样 FastAPI 就会知道该做什么。

提示

如果这看起来比有帮助更让人困惑,请忽略它,你不需要使用它。

这只是一个快捷方式。因为 FastAPI 旨在帮助你减少代码重复。