Зависимости

FastAPI имеет очень мощную, но интуитивно понятную систему ввода зависимостей.

Она разработана для простоты использования и лёгкого интегрирования различных компонентов с FastAPI для любого разработчика.

Что такое "ввод зависимостей"

"Ввод зависимостей" в программировании означает, что ваш код (в данном случае ваши функции обработки путей) может объявлять необходимые для работы элементы — "зависимости".

Затем система (в данном случае FastAPI) позаботится о предоставлении коду этих зависимостей ("введет" зависимости).

Это очень полезно, когда необходимо:

• Использовать общую логику (один и тот же код многократно).
• Использовать общие подключения к базе данных.
• Обеспечить безопасность, аутентификацию, требования к ролям и т. д.
• И многое другое...

Всё это при минимальном повторении кода.

Первые шаги

Давайте рассмотрим очень простой пример. Пока он не очень полезен.

Но таким образом мы можем сосредоточиться на том, как работает система ввода зависимостей.

Создание зависимости

Сначала сосредоточимся на зависимости.

Это просто функция, которая может принимать все те же параметры, что и функция обработки пути:

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

from typing import Annotated, 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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

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

Всё.

2 строки.

И она имеет такую же форму и структуру, как и все ваши функции обработки пути.

Можно рассматривать её как функцию обработки пути без "декоратора" (без @app.get("/some-path")).

И она может возвращать всё, что вам нужно.

В этом случае эта зависимость ожидает:

• Необязательный параметр запроса q, который является str.
• Необязательный параметр запроса skip, который является int, и по умолчанию равен 0.
• Необязательный параметр запроса limit, который является int, и по умолчанию равен 100.

И затем просто возвращает dict, содержащий эти значения.

Импортировать Depends

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

from typing import Annotated, 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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

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

Объявление зависимости в "зависимом" коде

Так же, как вы используете Body, Query, и т. д., с параметрами ваших функций обработки пути, используйте Depends с новым параметром:

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

from typing import Annotated, 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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

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

Хотя вы используете Depends в параметрах вашей функции так же, как Body, Query, и т. д., Depends работает немного иначе.

Вы передаёте Depends только один параметр.

Этот параметр должен быть чем-то похожим на функцию.

Вы не вызываете её напрямую (не добавляйте скобки в конце), а просто передаёте её как параметр в Depends().

И эта функция принимает параметры так же, как и функции обработки пути.

При каждом новом запросе FastAPI позаботится о:

• Вызове вашей функции зависимости ("зависимой") с правильными параметрами.
• Получении результата от вашей функции.
• Присвоении этого результата параметру в вашей функции обработки пути.

graph TB

common_parameters(["common_parameters"])
read_items["/items/"]
read_users["/users/"]

common_parameters --> read_items
common_parameters --> read_users

Таким образом, вы пишете общий код один раз, а FastAPI позаботится о его вызове для ваших операций с путями.

Поделиться зависимостями Annotated

В приведенных выше примерах вы видите небольшое повторение кода.

Когда вам нужно использовать зависимость common_parameters(), вы должны написать весь параметр с аннотацией типа и Depends():

commons: Annotated[dict, Depends(common_parameters)]

Но так как мы используем Annotated, мы можем сохранить это значение Annotated в переменной и использовать его в нескольких местах:

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}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

from typing import Annotated, 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}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

Зависимости будут продолжать работать как ожидается, и самое лучшее — это то, что информация о типе будет сохранена, что означает, что ваш редактор сможет предоставлять вам автодополнение, встроенные ошибки и т. д. То же самое относится и к другим инструментам, таким как mypy.

Это будет особенно полезно, когда вы используете его в большом кодовом основании, где вы используете одни и те же зависимости снова и снова во многих операциях с путями.

Использовать async или нет async

Так как зависимости также будут вызываться FastAPI (так же, как и ваши функции обработки пути), применяются те же правила при определении ваших функций.

Вы можете использовать async def или обычные def.

Вы можете объявлять зависимости с async def внутри обычных def функций обработки пути, или зависимости def внутри async def функций обработки пути и т. д.

Это не имеет значения. FastAPI будет знать, что делать.

Интеграция с OpenAPI

Все объявления, валидации и требования к вашим зависимостям (и подзависимостям) будут интегрированы в одну схему OpenAPI.

Таким образом, в интерактивных документах также будет отображаться вся информация из этих зависимостей:

Простое использование

Если вы посмотрите, функции обработки путей объявляются для использования всякий раз, когда совпадают путь и операция, а затем FastAPI позаботится о вызове функции с правильными параметрами, извлекая данные из запроса.

Фактически, все (или большинство) веб-фреймворков работают таким же образом.

Вы никогда не вызываете эти функции напрямую. Их вызывает ваш фреймворк (в данном случае, FastAPI).

С помощью системы внедрения зависимостей вы также можете указать FastAPI, что ваша функция обработки пути также «зависит» от чего-то еще, что должно быть выполнено до вашей функции обработки пути, и FastAPI позаботится об его выполнении и «внедрении» результатов.

Другие распространенные термины для этой же идеи «внедрения зависимостей»:

• ресурсы
• поставщики
• сервисы
• вводимые элементы
• компоненты

Плагины FastAPI

Интеграции и «плагины» могут быть созданы с помощью системы внедрения зависимостей. Но на самом деле, нет необходимости создавать «плагины», так как с помощью зависимостей можно объявить бесконечное количество интеграций и взаимодействий, которые станут доступны вашим функциям обработки путей.

И зависимости могут быть созданы очень простым и интуитивно понятным способом, который позволяет просто импортировать необходимые Python-пакеты и интегрировать их с вашими функциями API в пару строк кода, буквально.

Вы увидите примеры этого в следующих главах, посвященных реляционным и NoSQL базам данных, безопасности и т. д.

Совместимость FastAPI

Простота системы внедрения зависимостей делает FastAPI совместимым с:

• всеми реляционными базами данных
• NoSQL базами данных
• внешними пакетами
• внешними API
• системами аутентификации и авторизации
• системами мониторинга использования API
• системами внедрения данных ответа
• и т. д.

Просто и мощно

Хотя иерархическая система внедрения зависимостей очень проста в определении и использовании, она по-прежнему очень мощная.

Вы можете определять зависимости, которые, в свою очередь, могут определять зависимости.

В конечном итоге создается иерархическое дерево зависимостей, и система внедрения зависимостей позаботится о решении всех этих зависимостей за вас (и их подзависимостей) и предоставлении (внедрении) результатов на каждом шаге.

Например, предположим, что у вас есть 4 конечные точки API (операции с путями):

• /items/public/
• /items/private/
• /users/{user_id}/activate
• /items/pro/

тогда вы могли бы добавить различные требования к разрешениям для каждой из них просто с помощью зависимостей и подзависимостей:

graph TB

current_user(["current_user"])
active_user(["active_user"])
admin_user(["admin_user"])
paying_user(["paying_user"])

public["/items/public/"]
private["/items/private/"]
activate_user["/users/{user_id}/activate"]
pro_items["/items/pro/"]

current_user --> active_user
active_user --> admin_user
active_user --> paying_user

current_user --> public
active_user --> private
admin_user --> activate_user
paying_user --> pro_items

Интегрировано с OpenAPI

Все эти зависимости, объявляя свои требования, также добавляют параметры, проверки и т. д. к вашим операциям с путями.

FastAPI позаботится о добавлении всего этого в схему OpenAPI, чтобы она отображалась в интерактивных системах документации.