高级依赖

参数化依赖

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

但是，在某些情况下，您可能希望能够为依赖项设置参数，而无需声明许多不同的函数或类。

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

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

“可调用”实例

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

不是类本身（它已经是可调用的），而是该类的实例。

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

Python 3.9+

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.9+ - 非注解

提示

如果可能，请优先使用 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.9+

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.9+ - 非注解

提示

如果可能，请优先使用 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.9+

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.9+ - 非注解

提示

如果可能，请优先使用 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。

将实例用作依赖项

然后，我们可以将此 checker 用在 Depends(checker) 中，而不是 Depends(FixedContentQueryChecker)，因为依赖项是实例 checker，而不是类本身。

在解析依赖项时，**FastAPI** 将像这样调用此 checker

checker(q="somequery")

... 并将该函数返回的任何内容作为依赖项的值传递给我们的路由处理函数中的参数 fixed_content_included

Python 3.9+

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.9+ - 非注解

提示

如果可能，请优先使用 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 讨论区提问，并附上您的具体用例以及为什么您会受益于带 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 传递给后台任务函数，然后在后台任务函数内部再次获取该对象。