Продвинутые зависимости¶
Параметризованные зависимости¶
Все зависимости, которые мы видели, — это конкретная функция или класс.
Но бывают случаи, когда нужно задавать параметры зависимости, не объявляя много разных функций или классов.
Представим, что нам нужна зависимость, которая проверяет, содержит ли query-параметр q некоторое фиксированное содержимое.
Но при этом мы хотим иметь возможность параметризовать это фиксированное содержимое.
«Вызываемый» экземпляр¶
В Python есть способ сделать экземпляр класса «вызываемым» объектом.
Не сам класс (он уже является вызываемым), а экземпляр этого класса.
Для этого объявляем метод __call__:
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}
🤓 Other versions and variants
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
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}
Tip
Prefer to use the Annotated version if possible.
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}
В этом случае именно __call__ FastAPI использует для проверки дополнительных параметров и подзависимостей, и именно он будет вызван, чтобы позже передать значение параметру в вашей функции-обработчике пути.
Параметризуем экземпляр¶
Теперь мы можем использовать __init__, чтобы объявить параметры экземпляра, с помощью которых будем «параметризовать» зависимость:
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}
🤓 Other versions and variants
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
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}
Tip
Prefer to use the Annotated version if possible.
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__ и не зависит от него — мы используем его напрямую в нашем коде.
Создаём экземпляр¶
Мы можем создать экземпляр этого класса так:
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}
🤓 Other versions and variants
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
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}
Tip
Prefer to use the Annotated version if possible.
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:
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}
🤓 Other versions and variants
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
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}
Tip
Prefer to use the Annotated version if possible.
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.118.0 и вы столкнулись с проблемами зависимостей с yield.
Зависимости с yield со временем изменялись, чтобы учитывать разные случаи применения и исправлять проблемы. Ниже — краткое резюме изменений.
Зависимости с yield и StreamingResponse, технические детали¶
До FastAPI 0.118.0, если вы использовали зависимость с yield, код после yield выполнялся после возврата из функции-обработчика пути, но прямо перед отправкой ответа.
Идея состояла в том, чтобы не удерживать ресурсы дольше необходимого, пока ответ «путешествует» по сети.
Это изменение также означало, что если вы возвращали StreamingResponse, код после yield в зависимости уже успевал выполниться.
Например, если у вас была сессия базы данных в зависимости с yield, StreamingResponse не смог бы использовать эту сессию во время стриминга данных, потому что сессия уже была закрыта в коде после yield.
В версии 0.118.0 это поведение было возвращено к тому, что код после yield выполняется после отправки ответа.
Информация
Как вы увидите ниже, это очень похоже на поведение до версии 0.106.0, но с несколькими улучшениями и исправлениями краевых случаев.
Сценарии с ранним выполнением кода после yield¶
Есть некоторые сценарии со специфическими условиями, которым могло бы помочь старое поведение — выполнение кода после yield перед отправкой ответа.
Например, представьте, что вы используете сессию базы данных в зависимости с yield только для проверки пользователя, а в самой функции-обработчике пути эта сессия больше не используется, и при этом ответ отправляется долго, например, это StreamingResponse, который медленно отправляет данные и по какой-то причине не использует базу данных.
В таком случае сессия базы данных будет удерживаться до завершения отправки ответа, хотя если вы её не используете, удерживать её не требуется.
Это могло бы выглядеть так:
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))
Код после yield, автоматическое закрытие Session в:
# Code above omitted 👆
def get_session():
with Session(engine) as session:
yield session
# Code below omitted 👇
👀 Full file preview
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))
…будет выполнен после того, как ответ закончит отправку медленных данных:
# 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))
👀 Full file preview
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), вы можете явно закрыть сессию, когда она больше не нужна:
# 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 👇
👀 Full file preview
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 Discussions с описанием конкретного кейса и почему вам было бы полезно иметь раннее закрытие для зависимостей с yield.
Если появятся веские причины для раннего закрытия в зависимостях с yield, я рассмотрю добавление нового способа опционально включать раннее закрытие.
Зависимости с yield и except, технические детали¶
До FastAPI 0.110.0, если вы использовали зависимость с yield, затем перехватывали исключение с except в этой зависимости и не пробрасывали исключение снова, исключение автоматически пробрасывалось дальше к обработчикам исключений или к обработчику внутренней ошибки сервера.
В версии 0.110.0 это было изменено, чтобы исправить неконтролируемое потребление памяти из‑за проброшенных исключений без обработчика (внутренние ошибки сервера) и привести поведение в соответствие с обычным поведением Python-кода.
Фоновые задачи и зависимости с yield, технические детали¶
До FastAPI 0.106.0 вызывать исключения после yield было невозможно: код после yield в зависимостях выполнялся уже после отправки ответа, поэтому Обработчики исключений к тому моменту уже отработали.
Так было сделано в основном для того, чтобы можно было использовать те же объекты, «отданные» зависимостями через yield, внутри фоновых задач, потому что код после yield выполнялся после завершения фоновых задач.
В FastAPI 0.106.0 это изменили, чтобы не удерживать ресурсы, пока ответ передаётся по сети.
Совет
Кроме того, фоновая задача обычно — это самостоятельный фрагмент логики, который следует обрабатывать отдельно, со своими ресурсами (например, со своим подключением к базе данных).
Так код, скорее всего, будет чище.
Если вы полагались на прежнее поведение, теперь ресурсы для фоновых задач следует создавать внутри самой фоновой задачи и использовать внутри неё только данные, которые не зависят от ресурсов зависимостей с yield.
Например, вместо использования той же сессии базы данных, создайте новую сессию в фоновой задаче и получите объекты из базы данных с помощью этой новой сессии. И затем, вместо передачи объекта из базы данных параметром в функцию фоновой задачи, передавайте идентификатор этого объекта и заново получайте объект внутри функции фоновой задачи.