高级依赖项

参数化依赖项

我们目前所见到的所有依赖项都是固定的函数或类。

但有时你可能希望能够为依赖项设置参数，而无需声明许多不同的函数或类。

设想我们需要一个依赖项，用于检查查询参数 q 是否包含某些固定的内容。

但我们希望能够对该固定内容进行参数化。

“可调用”实例

在 Python 中，有一种方法可以使类的实例成为“可调用对象”。

不是类本身（类已经是可调用对象），而是该类的一个实例。

为此，我们声明一个 __call__ 方法。

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

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

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

在这种情况下，FastAPI 将使用这个 __call__ 来检查额外的参数和子依赖项，并且稍后会在调用时将值传递给你的路径操作函数中的参数。

参数化实例

现在，我们可以使用 __init__ 来声明实例的参数，从而“参数化”该依赖项。

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

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

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

在这种情况下，FastAPI 永远不会触碰或关心 __init__，我们将直接在代码中使用它。

创建实例

我们可以通过以下方式创建该类的实例：

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

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

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

这样我们就能够“参数化”我们的依赖项了，现在它内部包含 "bar"，作为属性 checker.fixed_content。

将实例用作依赖项

然后，我们可以使用 Depends(checker) 而不是 Depends(FixedContentQueryChecker) 来使用这个 checker，因为依赖项是实例 checker，而不是类本身。

当解析依赖项时，FastAPI 将会像这样调用该 checker：

checker(q="somequery")

……并将它的返回值作为依赖项的值，传递给我们的路径操作函数中的参数 fixed_content_included。

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

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

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

提示

这一切看起来可能有些复杂，而且目前还不太清楚有什么用处。

这些示例故意设计得很简单，但展示了它是如何工作的。

在关于安全性的章节中，有一些实用工具函数就是通过这种方式实现的。

如果你理解了这一切，你就已经掌握了那些安全工具在底层的实现原理。

带有 yield、HTTPException、except 和后台任务的依赖项

警告

你很可能不需要了解这些技术细节。

这些细节主要适用于你使用的 FastAPI 版本低于 0.121.0，并且在使用带有 yield 的依赖项时遇到问题的情况。

带有 yield 的依赖项随着时间的推移不断演进，以解决不同的用例和修复一些问题，以下是变更的总结。

带有 yield 和 scope 的依赖项

在 0.121.0 版本中，FastAPI 增加了对带有 yield 的依赖项使用 Depends(scope="function") 的支持。

使用 Depends(scope="function") 时，yield 之后的退出代码会在路径操作函数完成执行后、响应发回客户端之前执行。

而使用 Depends(scope="request")（默认值）时，yield 之后的退出代码会在响应发送后执行。

你可以在 带有 yield 的依赖项 - 提前退出与 scope 文档中阅读更多相关信息。

带有 yield 和 StreamingResponse 的依赖项，技术细节

在 FastAPI 0.118.0 之前，如果你使用了带有 yield 的依赖项，它会在路径操作函数返回后，但在发送响应之前执行退出代码。

这样做是为了避免资源持有时间超过必要长度，不必等待响应通过网络传输完成。

此项变更也意味着，如果你返回了一个 StreamingResponse，带有 yield 的依赖项的退出代码将已经执行完毕。

例如，如果你在一个带有 yield 的依赖项中拥有一个数据库会话，StreamingResponse 将无法在流式传输数据时使用该会话，因为会话已经在 yield 后的退出代码中关闭了。

这一行为在 0.118.0 版本中进行了回退，使得 yield 之后的退出代码在响应发送后执行。

信息

如下文所示，这与 0.106.0 版本之前的行为非常相似，但针对边缘情况进行了多项改进和错误修复。

提前退出的用例

有些特定条件下的用例可能受益于在发送响应之前运行 yield 依赖项退出代码的旧行为。

例如，假设你编写的代码在一个带有 yield 的依赖项中使用了数据库会话，但仅是为了验证用户，且该数据库会话在路径操作函数中不再使用，并且响应发送需要很长时间（例如缓慢发送数据的 StreamingResponse），但出于某种原因不使用数据库。

在这种情况下，数据库会话会一直被持有直到响应发送完毕，但如果你实际上并没有使用它，那么就没有必要持有它。

以下是它的表现形式：

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

其中的退出代码（自动关闭 Session 的部分）

Python 3.10+

# Code above omitted 👆

def get_session():
    with Session(engine) as session:
        yield session

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

……将在缓慢数据传输结束后执行。

Python 3.10+

# Code above omitted 👆

def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

👀 完整文件预览

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

但由于 generate_stream() 并不使用数据库会话，因此在发送响应时保持会话打开并不是必要的。

如果你在使用 SQLModel（或 SQLAlchemy）时有这种特定用例，你可以在不再需要它时显式关闭会话：

Python 3.10+

# Code above omitted 👆

def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

这样会话就会释放数据库连接，以便其他请求可以使用它。

如果你有其他需要从带有 yield 的依赖项中提前退出的用例，请创建一个 GitHub Discussion 问题，详细说明你的用例以及为什么你需要为带有 yield 的依赖项实现提前关闭。

如果有令人信服的、需要为带有 yield 的依赖项实现提前关闭的用例，我会考虑添加一种新的方式来选择启用提前关闭。

带有 yield 和 except 的依赖项，技术细节

在 FastAPI 0.110.0 之前，如果你使用了带有 yield 的依赖项，并在该依赖项中通过 except 捕获了异常，且没有再次抛出该异常，那么该异常会自动被抛出/转发给任何异常处理器或内部服务器错误处理器。

这一行为在 0.110.0 版本中被更改，目的是修复因未处理的转发异常（内部服务器错误）而导致的内存消耗问题，并使其与常规 Python 代码的行为保持一致。

后台任务与带有 yield 的依赖项，技术细节

在 FastAPI 0.106.0 之前，在 yield 之后抛出异常是不可能的，带有 yield 的依赖项中的退出代码是在响应发送之后执行的，因此 异常处理器 此时已经运行过了。

这种设计主要是为了允许在后台任务中使用依赖项“yield”出的相同对象，因为退出代码会在后台任务完成后执行。

这在 FastAPI 0.106.0 中被更改，目的是在等待响应通过网络传输时不再持有资源。

提示

此外，后台任务通常是一组独立的逻辑，应该单独处理，并拥有自己的资源（例如，它自己的数据库连接）。

因此，通过这种方式，你的代码可能会更整洁。

如果你曾经依赖过这种行为，现在你应该在后台任务本身内部创建后台任务所需的资源，并且在内部只使用不依赖于带有 yield 的依赖项资源的数据。

例如，与其使用同一个数据库会话，不如在后台任务内部创建一个新的数据库会话，并使用这个新会话从数据库获取对象。然后，与其将数据库对象作为参数传递给后台任务函数，不如传递该对象的 ID，然后在后台任务函数内部重新获取该对象。