跳转至

类作为依赖项

在深入探究 依赖注入 系统之前,让我们升级之前的例子。

来自前一个例子的dict

在前面的例子中, 我们从依赖项 ("可依赖对象") 中返回了一个 dict:

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
from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[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 提供很多支持(比如补全),因为编辑器不知道 dict 的键和值类型。

对此,我们可以做的更好...

什么构成了依赖项?

到目前为止,您看到的依赖项都被声明为函数。

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

关键因素是依赖项应该是 "可调用对象"。

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

在这个例子中, fluffy 是一个 Cat 类的实例。

为了创建 fluffy,你调用了 Cat

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

因此,在 FastAPI 中,你可以使用一个 Python 类作为一个依赖项。

实际上 FastAPI 检查的是它是一个 "可调用对象"(函数,类或其他任何类型)以及定义的参数。

如果您在 FastAPI 中传递一个 "可调用对象" 作为依赖项,它将分析该 "可调用对象" 的参数,并以处理路径操作函数的参数的方式来处理它们。包括子依赖项。

这也适用于完全没有参数的可调用对象。这与不带参数的路径操作函数一样。

所以,我们可以将上面的依赖项 "可依赖对象" common_parameters 更改为类 CommonQueryParams:

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
from typing import Union

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: Union[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 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
from typing import Union

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: Union[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 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
from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[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 schema 上文档化,等等。

使用它

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

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
from typing import Union

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: Union[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: CommonQueryParams = Depends(CommonQueryParams)

最后的 CommonQueryParams:

... = Depends(CommonQueryParams)

...实际上是 Fastapi 用来知道依赖项是什么的。

FastAPI 将从依赖项中提取声明的参数,这才是 FastAPI 实际调用的。


在本例中,第一个 CommonQueryParams

commons: CommonQueryParams ...

...对于 FastAPI 没有任何特殊的意义。FastAPI 不会使用它进行数据转换、验证等 (因为对于这,它使用 = Depends(CommonQueryParams))。

你实际上可以只这样编写:

commons = Depends(CommonQueryParams)

..就像:

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
from typing import Union

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: Union[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: CommonQueryParams = Depends(CommonQueryParams)

FastAPI 为这些情况提供了一个快捷方式,在这些情况下,依赖项 明确地 是一个类,FastAPI 将 "调用" 它来创建类本身的一个实例。

对于这些特定的情况,您可以跟随以下操作:

不是写成这样:

commons: CommonQueryParams = Depends(CommonQueryParams)

...而是这样写:

commons: CommonQueryParams = Depends()

您声明依赖项作为参数的类型,并使用 Depends() 作为该函数的参数的 "默认" 值(在 = 之后),而在 Depends() 中没有任何参数,而不是在 Depends(CommonQueryParams) 编写完整的类。

同样的例子看起来像这样:

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
from typing import Union

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: Union[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 会知道怎么处理。

Tip

如果这看起来更加混乱而不是更加有帮助,那么请忽略它,你不需要它。

这只是一个快捷方式。因为 FastAPI 关心的是帮助您减少代码重复。