类作为依赖项¶
在深入研究依赖注入系统之前,让我们先升级一下上一个示例。
来自上一个示例的 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")
在这种情况下,fluffy 是 Cat 类的一个实例。
而为了创建 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 旨在帮助你减少代码重复。