FastAPI и Dependency Injection: правда или вымысел?16.12.2024 17:30

В свое время FastAPI прогремел как гром среди ясного неба — тут тебе и минималистичный API аля-Flask (все устали от Django, диктующего свои правила), и OpenAPI документация из коробки, и удобное тестирование, и хайповая асинхронность. Буквально все, что нужно для свободы творчества и никаких ограничений! Да еще и Depends завезли! В тот момент это был культрурный шок — Dependency Injection в Python? Разве это не что-то из Java?

FastAPI показал, что DI — это паттерн, упрощающий разработку вне зависимости от языка программирования. Теперь DI как фича является практически неотъемлимым элементом любого нового Python-фреймворка (Litestar/Blacksheep/FastStream/etc), ведь людям это нужно. Все хотят «как в FastAPI».

Но дьявол кроется в деталях. А вы уверены, что те самые Depends == Dependency Injection? Уверены, что пишете код на FastAPI правильно?

Что Tiangolo (создатель FastAPI) прививает вам «лучшие практики»?

В рамках статьи мы рассмотрим различные подходы к организации зависимостей в рамках FastAPI проекта, оценим их с точки зрения удобства использования и постараемся разобраться, как же все-таки «правильно» готовить DI в FastAPI.

Что такое DI и зачем он нам нужен?

Dependency Injection — это паттерн, напрямую реализующий принцип Инверсии зависимостей (DIP — Dependency Inversion Principle) из soliD. Иногда можно встретить формулировку Inversion of Control (IoC), что суть — о том же самом.

DIP принцип заключается в том, что наша бизнес-логика не должна зависеть от деталей реализации (базы данных, протокола взаимодействия, конкретных библиотек). Вместо этого она должна запрашивать абстрактные интерфейсы, декларирующие методы, которые ей необходимы. Эти «абстрактные интерфейсы» находятся в ядре вашей системы, т.к. жизненно необходимы для ее функционирования.

А вот с помощью паттерна Dependency Injection «реальные» имплементации этих интерфейсов (которые знают про конкретные базы данных и тд) будут доставляться в вашу логику извне при инициализации проекта (тот самый Injection).

Т.е. вместо подобного кода: 

class TokenChecker:
    def __init__(self) -> None:
        self.storage = Redis()

    def check_token(self, token: str) -> bool:
        return self.storage.get(token) is not None

checker = TokenChecker()

Мы должны писать нечто такое: 

from typing import Protocol

# находится в БЛ, так как нужен для ее функционирования
class Storage(Protocol):
    def get(self, token: str) -> str | None:
        ...

class TokenChecker:
    def __init__(self, storage: Storage) -> None:
        self.storage = storage

    def check_token(self, token: str) -> bool:
        return self.storage.get(token) is not None

real_storage = Redis() # объект Redis подходит под интерфейс Storage
checker = TokenChecker(real_storage)

Кода стало больше, но зачем? — Теперь TokenChecker больше не знает о том, что работает с Redis, а это позволяет нам

  1. Заменить Redis на Memcached или даже хранение в памяти при необходимости

  2. Поместить в качестве Storage mock-объект в тестах удобным и понятным способом

Изначальная мотивация действительно пришла из Java и других компилируемых языков. Смысл в том, что внешний слой подвергается изменениям часто, а вот внутренний — редко. Если мы зависим от внешнего слоя в нашей бизнес-логике (банально делаем импорты оттуда), то при повторной компиляции проекта эти модули также придется перекомпилировать, хотя изменений в них не произошло (изменения были в модулях, от которых они зависят). Неконтролируемые зависимости приводят к тому, что весь проект пересобирается при изменении любой строки в любом файле и тем самым многочасовым «код компилируется».

Однако, DI — это хорошая практика, которая приносит ощутимую пользу в любых языках.

DI в FastAPI по Tiangolo

Одна из основных фич FastAPI — его Depends, которая как раз позиционируется как реализация Dependency Injection принципа. Давайте посмотрим, как Tiangolo предлагает ее использовать: 

from typing import Annotated

from fastapi import Depends

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

«Так вы можете переиспользовать логику между разными эндпоинтами» — вот как аргументирует использование Depends Tiangolo. Однако, это не Dependency Injection.

Просто давайте взглянем на следующий код: 

from typing import Annotated

from fastapi import Request

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(request: Request):
    commons = await common_parameters(**request.query_params)
    return commons

@app.get("/users")
async def read_users(request: Request):
    commons = await common_parameters(**request.query_params)
    return commons

Разве это не то же самое «переиспользование логики», с которым нам хочет помочь Tiangolo? Кажется, его помощь — это просто еще один слой синтаксического сахара (не бесплатного, конечно).

Однако, Dependency Injection тут все-таки есть, т.к. есть возможность заменить зависимость через механизм dependency-overrides

async def override_dependency(q: str | None = None):
    return {"q": q, "skip": 5, "limit": 10}

app.dependency_overrides[common_parameters] = override_dependency

В варианте с прямым использованием функции это невозможно.

Правда, механизм позиционируется «для тестов» и все еще не помогает соблюсти DIP — мы подменяем зависимость от реализации на другую зависимость от реализации. Что может только путать людей, работающих с кодовой базой.

Но не все потеряно и мы можем доработать Depends так, чтобы это был настоящий DI с соблюдением DIP.

«Настоящий» DI в FastAPI

Не претендую на авторство данного подхода, но готов принять все шишки за его использование, т.к. не нашел способа сделать DI лучше.

Так вот: помним, что в DI нам нужно завязывать на абстракцию, а реализацию Inject'ить?

В FastAPI МОЖНО реализовать Dependency Injection с соблюдением DIP. Но не совсем тем способом, которым планировал Tiangolo.

В FastAPI у нас есть глобальный словарь app.dependency_overrides, который предлагается использовать для «тестирования зависимостей» (в документации). Однако, по всем внешним признакам — это контейнер зависимостей. И мы можем его использовать как раз по прямому назначению IoC контейнера — Inject’ить зависимости.

Давайте разбираться по порядку.

Вводим абстракцию

Давайте представим, что нам нужно идентифицировать пользователя по наличию токена в кеше? Код будет несколько упрощен относительно реального, но смысл от этого становится только яснее.

Для начала введем интерфейс объекта, с помощью которого мы как раз будем валидировать токен: 

from typing import Protocol

class TokenRepo(Protocol):
    async def get_user_by_token(self, token: str) -> str | None:
        ...

    async def set_user_token(self, token: str, username: str) -> None:
        ...

Зависим от абстракции

Теперь нам нужно «завязаться» на эту абстракцию в нашем эндпоинте: 

from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: Annotated[TokenRepo, Depends()],  # "запрашиваем" абстракцию
) -> str | None:
    return await token_repo.get_user_by_token(token)

Пишем реализацию

Нам остается только реализовать где-то заданный интерфейс и поместить эту реализацию в наш контейнер зависимостей (откуда она попадет в исполняемый код вместо абстракции).

Реализация протокола для работы с Redis: 

from redis.asyncio import Redis

class RedisTokenRepo(TokenRepo):
    def __init__(
        self,
        redis: Redis,
        expiration: str,
    ) -> None:
        self.redis = redis
        self.token_expiration = expiration

    async def get_user_by_token(self, token: str) -> str | None:
        if username := await self.redis.get(token):
            return username.decode()

    async def set_user_token(self, token: str, username: str) -> None:
        await self.redis.set(
            name=token,
            value=username,
            ex=self.token_expiration,
        )

Используем реализацию вместо абстракции

Ну и «помещаем» нашу реализацию в FastAPI IoC Container: 

def setup_ioc_container(
    app: FastAPI,
) -> FastAPI:
    settings_object = {  # mock настроек
        "redis_url": "redis://localhost:6379",
        "token_expiration": 300,
    }

    redis_repo = RedisTokenRepo(
        redis=Redis.from_url(settings_object["redis_url"]),
        expiration=settings_object["token_expiration"],
    )

    app.dependency_overrides.update({
        TokenRepo: lambda: redis_repo,
    })

    return app

Реальной зависимостью в нашем случае является lambda: redis_repo. Именно эта функция будет вызываться при каждом запросе с Annotated[TokenRepo, Depends()] зависимостью.

Мы реализовали ее через lambda для того, чтобы избежать вызова конструктора RedisTokenRepo на каждый вызов, а сделать этот объект «синглтоном».

Так выглядит DI в FastAPI «здорового человека». Но не совсем.

Боремся с FastAPI

К сожалению, Tiangolo не планировал использование Depends таким образом. Он не хочет, чтобы мы зависили от «абстракции». Поэтому в нашу OpenAPI схему просочилось что-то странное (args, kwargs?):

ee8a91b8f021cbfaf0e6949cc4e292ed.png

Для того, чтобы от этого избавить нужно «спрятать» от FastAPI сигнатуру исходной «абстракции»

Сделать это можно следующим образом: 

from typing import Callable, Any

class Stub:
    def __init__(self, dependency: Callable[..., Any]) -> None:
        """Сохраняем нашу абстракцию."""
        self._dependency = dependency

    def __call__(self) -> None:
        """Выкинем ошибку, если забыли подменить реализацию при старте приложения."""
        raise NotImplementedError(f"You forgot to register `{self._dependency}` implementation.")

    def __hash__(self) -> int:
        """Обманываем app.dependency_overrides, чтобы он считал Stub реальной зависимостью"""
        return hash(self._dependency)

    def __eq__(self, __value: object) -> bool:
        """Обманываем app.dependency_overrides, чтобы он считал Stub реальной зависимостью"""
        if isinstance(__value, Stub):
            return self._dependency == __value._dependency
        else:
            return self._dependency == __value

Теперь мы должны «запрашивать» зависимость следующим образом: 

@app.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: Annotated[TokenRepo, Depends(Stub(TokenRepo))]
): ...

Уже не так «сахарно», зато в схему ничего не течет.

e93ef01eea6d61d77790beb823b536ef.png

Резюме

Dependency Injection в FastAPI возможен, однако:

  • требует дополнительных приседаний, чтобы ничего не утекало в схему

  • требует дополнительных приседаний для реализации Application-level зависимостей (объктов, которые создаются 1 раз при старте приложения)

  • требует дополнительных приседаний для регистрации зависимостей

Альтернатива?

Допустим, DI в FastAPI нам не сильно нравится (а он нам не нравится) и мы хотим взять стороннюю библиотеку. Наверное, первое, что приходит на ум — это Dependency Injector. Но, кажется, создатель отказался от его сопровождения (да и библиотека имеет множество минусов, которые нам тоже не нравятся).

А что нам остается?

Все как-то не то. Слабое распространение, скудный функционал, нестабильный API.

В общем, в ходе продолжительных баталий дискуссий опытный разработчик Андрей Тихонов (автор канала Советы разработчикам, администратор ru-python, fastapi-ru и прочих крупных TG-групп) решил создать собственное решение — dishka!

Полное сравнение с другими библиотеками вы можете найти в документации библиотеки.

Но я пределагаю сначала взглянуть, как он работает, а потом уже сравнить с FastAPI Depends.

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

Концепция проста и незамысловата:

  1. Мы пишем «провайдеры» — классы, которые содержат в себе фабрики зависимостей

  2. Затем объединяем их в «контейнер», откуда они уже будут доставляться в конечные функции

  3. Используем интеграции с фреймворками для бесшовного встраивания контейнера в ваше приложение

Пишем «провайдер»

from dataclasses import dataclass

from dishka import Provider, Scope, provide, from_context

@dataclass
class Settings:
    redis_url: str
    token_expiration: int

class RepoProvider(Provider):
    # говорим, что объект типа Settings будем помещаться в контейнер пользователем
    settings = from_context(provides=Settings, scope=Scope.APP)

    @provide(scope=Scope.APP)  # зависимость уровня приложения (синглтон)
    def get_redis_token_repo(
        self,
        settings: Settings,  # "запрашиваем" другую зависимость
    ) -> TokenRepo:
        return RedisTokenRepo(
            redis=Redis.from_url(settings.redis_url),
            expiration=settings.token_expiration,
        )

Затем мы должны собрать из провайдера (у нас он один) контейнер

container = make_async_container(
    RepoProvider(),
    context={  # помещаем Settings в контейнер вручную
        Settings: Settings(
            redis_url="redis://localhost:6379",
            token_expiration=300,
        ),
    },
)

И наконец — используем этот контейнер в нашем FastAPI приложении! 

from fastapi import APIRouter, FastAPI
from dishka.integrations.fastapi import FromDishka, DishkaRoute, setup_dishka

router = APIRouter(route_class=DishkaRoute)  # используем специальный route_class

@router.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: FromDishka[TokenRepo],  # используем вместо Depends
) -> str | None:
    return await token_repo.get_user_by_token(token)

app = FastAPI()
app.include_router(router)
setup_dishka(container, app)

Как видите, приседаний стало меньше, сайд-эффектов — тоже, а контейнер можно переиспользовать и для других фреймворков/библиотек, если они запущены в том же рантайме (например, FastStream).

Выводы по dishka

По моему субъективному мнению dishka значительно комфортнее для реализации DI принципа в FastAPI проекте (относительно нативного Depends) по следуюшим причинам:

  • Имеет четкое разделение на Application-level (синглтоны в рамках приложения) и Request-level (создаются на каждый запрос) зависимости. Нативный Depends работает только для Request зависимостей, а Application-level (самые частые) приходится изобретать самостоятельно вокруг main (как в моем примере) и/или lifespan request.state.* (как советует Starlette). Также dishka поддерживает и другие Scope’ы, в т.ч. и кастомные, что позволяет использовать его в совершенно разных кейсах.

  • Финализация Application-level зависимостей. В FastAPI отдельной головной болью стоит вопрос о том, как их финализировать, а для асинхронных зависимостей — еще и инициализировать (асинхронный main? извращения с lifespan?). Dishka поддерживает как асинхронные фабрики зависимостей, так и фабрики с yield, так что обе проблемы для него просто не существуют.

  • Помогает организовать логику управления графом зависимостей в одном месте, не размазывая ее по разным функциям и частям приложения (а также избавляет от различных служебных функций-оберток, необходимых для победы над FastAPI)

  • Позволяет переиспользовать контейнер зависимостей в рамках всего приложения (и других фреймворков/библиотек), а не только handler'ах FastAPI (аккуратнее с этим). Также, вы без труда сможете мигрировать на другой веб-фреймворк без переписывания логики DI. HTTP-фреймворк в таком случае остается только на транспортном уровне, где мы и хотим его видеть.

  • Работает несколько быстрее стандартного Depends

Однако, у использования dishka есть и свои минусы

  • Придется потратить 20 минут на изучение новой библиотеки

  • +1 зависимость (и библиотека в вашем портфолио)

  • В уже созданном контейнере нельзя переопределить зависимости, поэтому организация main должна учитывать, что в тестах вам потребуется использовать другие контейнеры под разные сценарии

  • Возможность разделения фабрик зависимостей на логические группы (по разным провайдерам) может вскружить голову и вы сделаете хуже, чем было до dishka. Поэтому рекомендую начинать с 1 го провайдера на приложение, а там — как пойдет.

  • Нет возможности учитывать зависимости при генерации OpenAPI

Выводы

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

В данной статье мы рассмотрели самый популярный и «правильный» подход к реализации принципа внедрения зависимостей в рамках FastAPI приложения, а также познакомились с dishka — великолепной библиотекой, которая позволяет реализовать DI в рамках любого приложения (в т.ч. и FastAPI).

Лично я рекомендую вам как минимум обратить внимение на эту библиотеку, а еще:

  • поддержать ее автора, поставив звезду на GitHub

  • вступить в телеграм чат, где вы можете пообщаться с ее создателем лично

  • прочитать статью про использование dishka с Litestar и FastStream

  • посмотреть доклад автора dishka на Podlodka Python Crew

  • подписаться на мой телеграм канал, если вам интересен подобный материал

© Habrahabr.ru