Встроенная система внедрения зависимостей, провайдеры, области видимости, кэширование зависимостей
Встроенная система внедрения зависимостей с поддержкой scopes и кэширования
Dependency Injection (DI) — паттерн, при котором зависимости (объекты, сервисы) передаются в функцию или класс извне, а не создаются внутри.
Преимущества DI:
Starlite имеет встроенную DI-систему, которая превосходит FastAPI по гибкости и возможностям.
from starlite import Starlite, get, Provide
# Провайдер — функция, создающая зависимость
async def get_redis() -> Redis:
return await aioredis.create_redis_pool("redis://localhost")
# Хендлер с зависимостью
@get("/cache/{key:str}", dependencies={"redis": Provide(get_redis)})
def get_cache(self, key: str, redis: Redis) -> str:
return redis.get(key)
app = Starlite(route_handlers=[get_cache])Provide — обёртка для провайдеров:
from starlite import Provide
def sync_provider() -> str:
return "value"
async def async_provider() -> str:
return "value"
# Для синхронных провайдеров
Provide(sync_provider, sync_to_thread=True)
# Для асинхронных провайдеров
Provide(async_provider, sync_to_thread=False)Параметры Provide:
dependency — сама функция-провайдерsync_to_thread — запускать ли в thread pool (для синхронных функций)scope — область видимости (см. ниже)cache — кэшировать ли результатStarlite поддерживает три уровня scopes:
from starlite import Scope
# APP — один экземпляр на всё приложение
Provide(get_redis, scope=Scope.APP)
# REQUEST — новый экземпляр для каждого запроса
Provide(get_db_session, scope=Scope.REQUEST)
# ROUTER — один экземпляр на роутер
Provide(get_service, scope=Scope.ROUTER)Зависимость создаётся один раз при запуске приложения:
async def get_database() -> Database:
db = Database()
await db.connect()
return db
# Один экземпляр БД на всё приложение
db_provider = Provide(get_database, scope=Scope.APP)
@get("/users", dependencies={"db": db_provider})
def get_users(db: Database) -> list[User]:
return db.query(User).all()Когда использовать:
Зависимость создаётся заново для каждого запроса:
async def get_db_session(
db: Database # Зависимость от APP-скопа
) -> AsyncSession:
async with db.session() as session:
yield session
# Новая сессия для каждого запроса
session_provider = Provide(get_db_session, scope=Scope.REQUEST)
@get("/users/{id:int}", dependencies={"session": session_provider})
def get_user(id: int, session: AsyncSession) -> User:
return session.get(User, id)Когда использовать:
Зависимость создаётся один раз на роутер:
router = Router(
path="/api",
route_handlers=[...],
dependencies={
"service": Provide(get_service, scope=Scope.ROUTER)
}
)Когда использовать:
По умолчанию Starlite кэширует результат провайдера в пределах scope:
# Кэширование включено по умолчанию
Provide(get_redis, cache=True)
# Отключить кэширование
Provide(get_random_value, cache=False)Можно установить время жизни кэша:
from datetime import timedelta
Provide(
get_expensive_resource,
cache=True,
cache_key="expensive_resource",
cache_ttl=timedelta(minutes=5)
)Провайдеры могут зависеть от других провайдеров:
async def get_database_url() -> str:
return "postgresql://localhost/mydb"
async def get_database(
url: str # Зависимость от get_database_url
) -> Database:
db = Database()
await db.connect(url)
return db
async def get_user_repository(
db: Database # Зависимость от get_database
) -> UserRepository:
return UserRepository(db)
@get(
"/users",
dependencies={
"db_url": Provide(get_database_url),
"db": Provide(get_database),
"repo": Provide(get_user_repository),
}
)
def get_users(repo: UserRepository) -> list[User]:
return repo.find_all()Starlite автоматически разрешает зависимости в правильном порядке.
Зависимости можно объявлять глобально для всего приложения:
app = Starlite(
route_handlers=[...],
dependencies={
"db": Provide(get_database, scope=Scope.APP),
"redis": Provide(get_redis, scope=Scope.APP),
"config": Provide(get_config, scope=Scope.APP),
}
)Теперь все хендлеры могут использовать эти зависимости:
@get("/users")
def get_users(db: Database) -> list[User]:
...
@get("/cache/{key}")
def get_cache(key: str, redis: Redis) -> str:
...Контроллеры могут объявлять собственные зависимости:
class UserController(Controller):
path = "/users"
dependencies = {
"repo": Provide(UserRepository),
"service": Provide(UserService),
}
@get()
def list_users(self, repo: UserRepository) -> list[User]:
return repo.find_all()
@post()
def create_user(
self,
data: UserCreate,
service: UserService
) -> User:
return service.create(data)Роутеры также поддерживают зависимости:
api_router = Router(
path="/api",
dependencies={
"auth_service": Provide(AuthService),
},
route_handlers=[UserController, ItemController]
)Для тестирования можно переопределить провайдеры:
# В тестах
async def get_mock_db() -> Database:
return MockDatabase()
app = Starlite(
route_handlers=[...],
dependencies={
"db": Provide(get_database),
}
)
# Переопределение для тестов
app.dependency_overrides["db"] = Provide(get_mock_db)Пример DI для SQLAlchemy:
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
engine = create_async_engine("postgresql+asyncpg://localhost/mydb")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
app = Starlite(
dependencies={
"db": Provide(get_db, scope=Scope.REQUEST)
}
)DI работает с Pydantic-моделями:
from pydantic import BaseModel
class Config(BaseModel):
api_key: str
debug: bool
async def get_config() -> Config:
return Config(api_key="secret", debug=True)
@get("/status")
def get_status(config: Config) -> dict:
return {"debug": config.debug}# ✅ Хорошо — БД на APP, сессия на REQUEST
Provide(get_database, scope=Scope.APP)
Provide(get_session, scope=Scope.REQUEST)
# ❌ Плохо — новая БД на каждый запрос
Provide(get_database, scope=Scope.REQUEST)# ✅ Хорошо
Provide(get_user_repository, dependencies={"repo": ...})
# ❌ Плохо
Provide(get_user_repository, dependencies={"x": ...})# ✅ Хорошо — явные зависимости
async def get_service(repo: UserRepository, cache: Redis):
...
# ❌ Плохо — создание внутри
async def get_service():
repo = UserRepository()
cache = Redis()# ✅ Хорошо
Provide(get_expensive_config, cache=True)
# ❌ Плохо — каждый раз заново
Provide(get_expensive_config, cache=False)| Возможность | FastAPI | Starlite |
|---|---|---|
| Scopes | Нет | APP, REQUEST, ROUTER |
| Кэширование | Ограниченное | Полное с TTL |
| Переопределение | app.dependency_overrides | app.dependency_overrides |
| Синтаксис | Depends() | Provide() в декораторе |
| Цепочки | Поддерживаются | Поддерживаются |
DI-система Starlite предлагает:
В следующей теме мы изучим валидацию данных с Pydantic.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.