Система плагинов, создание собственных расширений, интеграция со сторонними библиотеками, хуки жизненного цикла
Система плагинов, создание собственных расширений, интеграция со сторонними библиотеками
Starlite имеет мощную систему плагинов для расширения функциональности.
Некоторые популярные плагины:
from starlite import Starlite
from starlite.plugins.sqlalchemy import SQLAlchemyPlugin
from starlite.plugins.jwt import JWTPlugin
sqlalchemy_plugin = SQLAlchemyPlugin(
connection_string="postgresql+asyncpg://localhost/mydb"
)
jwt_plugin = JWTPlugin(
secret="your-secret-key",
algorithm="HS256",
)
app = Starlite(
route_handlers=[...],
plugins=[sqlalchemy_plugin, jwt_plugin],
)from starlite.plugins import PluginProtocol
from starlite import Starlite
class MyPlugin(PluginProtocol):
def __init__(self, config: dict):
self.config = config
def on_app_init(self, app_config: dict) -> dict:
"""Вызывается при инициализации приложения"""
# Добавляем middleware
if "middleware" not in app_config:
app_config["middleware"] = []
app_config["middleware"].append(self.custom_middleware)
# Добавляем зависимости
if "dependencies" not in app_config:
app_config["dependencies"] = {}
app_config["dependencies"]["my_service"] = self.get_service
return app_config
async def get_service(self):
"""Провайдер зависимости"""
return MyService(self.config)
def custom_middleware(self, request, call_next):
"""Middleware плагина"""
request.state.plugin_data = self.config
return await call_next(request)from starlite.plugins import PluginProtocol
from starlite.types import ASGIApp, Receive, Scope, Send
class LifecyclePlugin(PluginProtocol):
def __init__(self):
self.initialized = False
def on_app_init(self, app_config: dict) -> dict:
# Добавляем хуки жизненного цикла
if "on_startup" not in app_config:
app_config["on_startup"] = []
app_config["on_startup"].append(self.on_startup)
if "on_shutdown" not in app_config:
app_config["on_shutdown"] = []
app_config["on_shutdown"].append(self.on_shutdown)
return app_config
async def on_startup(self):
"""Вызывается при запуске приложения"""
print("Plugin starting up...")
self.initialized = True
# Инициализация ресурсов
await self.init_resources()
async def on_shutdown(self):
"""Вызывается при остановке приложения"""
print("Plugin shutting down...")
self.initialized = False
# Очистка ресурсов
await self.cleanup_resources()
async def init_resources(self):
...
async def cleanup_resources(self):
...from starlite.plugins import PluginProtocol
from starlite import Router, get
class AdminPlugin(PluginProtocol):
def __init__(self, admin_users: list[str]):
self.admin_users = admin_users
def on_app_init(self, app_config: dict) -> dict:
# Добавляем роутер с админ-эндпоинтами
if "route_handlers" not in app_config:
app_config["route_handlers"] = []
app_config["route_handlers"].append(self.create_router())
return app_config
def create_router(self) -> Router:
@get("/health")
def health_check(self) -> dict:
return {"status": "ok", "plugin": "admin"}
@get("/stats")
def stats(self) -> dict:
return {
"admin_users": len(self.admin_users),
"initialized": True,
}
return Router(
path="/admin",
route_handlers=[health_check, stats],
)from starlite import Starlite
async def init_database():
"""Инициализация БД при запуске"""
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text
engine = create_async_engine("postgresql+asyncpg://localhost/mydb")
async with engine.connect() as conn:
await conn.execute(text("CREATE TABLE IF NOT EXISTS ..."))
app = Starlite(
route_handlers=[...],
on_startup=[init_database],
)async def cleanup_database():
"""Очистка при остановке"""
# Закрытие соединений
await db_engine.dispose()
app = Starlite(
route_handlers=[...],
on_shutdown=[cleanup_database],
)from starlite import Request
async def before_request_hook(request: Request) -> None:
"""Вызывается перед каждым запросом"""
# Логирование
logger.info(f"{request.method} {request.url}")
# Измерение времени
request.state.start_time = time.time()
# Аутентификация
request.state.user = await get_user_from_token(
request.headers.get("Authorization")
)
app = Starlite(
route_handlers=[...],
before_request=before_request_hook,
)from starlite import Request, Response
async def after_request_hook(
request: Request,
response: Response,
) -> Response:
"""Вызывается после каждого запроса"""
# Добавляем заголовки
response.headers["X-Request-ID"] = request.state.request_id
# Логирование времени выполнения
duration = time.time() - request.state.start_time
logger.info(f"Request completed in {duration:.3f}s")
return response
app = Starlite(
route_handlers=[...],
after_request=after_request_hook,
)from starlite.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemyConfig
sqlalchemy_config = SQLAlchemyConfig(
connection_string="postgresql+asyncpg://user:pass@localhost/dbname",
engine_options={
"echo": True,
"pool_size": 10,
"max_overflow": 20,
},
)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
app = Starlite(
route_handlers=[...],
plugins=[sqlalchemy_plugin],
)
# Использование в хендлере
@get("/users")
async def get_users(
db: AsyncSession, # Из плагина
) -> list[User]:
result = await db.execute(select(User))
return result.scalars().all()from starlite import Provide
import redis.asyncio as redis
async def get_redis() -> redis.Redis:
return redis.Redis(
host="localhost",
port=6379,
db=0,
decode_responses=True,
)
app = Starlite(
route_handlers=[...],
dependencies={
"redis": Provide(get_redis, scope=Scope.APP),
},
)
@get("/cache/{key:str}")
async def get_cache(key: str, redis: redis.Redis) -> str | None:
return await redis.get(f"cache:{key}")from celery import Celery
celery_app = Celery(
"myapp",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/0",
)
@celery_app.task
def process_data_task(data: dict):
# Фоновая обработка
...
# В хендлере
@post("/process")
async def process_data(data: DataCreate) -> dict:
# Запуск фоновой задачи
task = process_data_task.delay(data.dict())
return {"task_id": task.id, "status": "processing"}
@get("/task/{task_id:str}")
async def get_task_status(task_id: str) -> dict:
from celery.result import AsyncResult
task = AsyncResult(task_id, app=celery_app)
return {"task_id": task_id, "status": task.status}import sentry_sdk
from starlite import Starlite
sentry_sdk.init(
dsn="https://...@sentry.io/...",
traces_sample_rate=1.0,
)
app = Starlite(
route_handlers=[...],
# Sentry автоматически перехватывает исключения
)from prometheus_client import Counter, Histogram, generate_latest
from starlite import get, Response
# Метрики
REQUEST_COUNT = Counter(
"http_requests_total",
"Total HTTP requests",
["method", "endpoint", "status"],
)
REQUEST_DURATION = Histogram(
"http_request_duration_seconds",
"HTTP request duration",
["method", "endpoint"],
)
# Middleware для метрик
async def metrics_middleware(request, call_next):
import time
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.url.path,
status=response.status_code,
).inc()
REQUEST_DURATION.labels(
method=request.method,
endpoint=request.url.path,
).observe(duration)
return response
# Эндпоинт для сбора метрик
@get("/metrics")
def metrics() -> Response:
return Response(
content=generate_latest(),
media_type="text/plain",
)
app = Starlite(
route_handlers=[...],
middleware=[metrics_middleware],
)from dataclasses import dataclass
from typing import Optional
@dataclass
class DatabasePluginConfig:
connection_string: str
pool_size: int = 10
max_overflow: int = 20
echo: bool = False
class DatabasePlugin(PluginProtocol):
def __init__(self, config: DatabasePluginConfig):
self.config = config
def on_app_init(self, app_config: dict) -> dict:
# Использование типизированной конфигурации
...
return app_configfrom pydantic import BaseModel, Field
class JWTPluginConfig(BaseModel):
secret: str = Field(min_length=32)
algorithm: str = "HS256"
expire_minutes: int = Field(ge=1, le=1440)
issuer: str | None = None
class JWTPlugin(PluginProtocol):
def __init__(self, config: JWTPluginConfig):
# Конфигурация уже валидирована Pydantic
self.config = config
...# ✅ Хорошо — независимый плагин
class CachePlugin(PluginProtocol):
def __init__(self, config: CacheConfig):
self.config = config
def on_app_init(self, app_config: dict) -> dict:
...
# ❌ Плохо — плагин зависит от других плагинов
class CachePlugin(PluginProtocol):
def __init__(self, db_plugin, redis_plugin):
# Жёсткая зависимость
...# ✅ Хорошо — документированная конфигурация
@dataclass
class MyPluginConfig:
"""Configuration for MyPlugin.
Attributes:
api_key: API key for external service
timeout: Request timeout in seconds
retry_count: Number of retries on failure
"""
api_key: str
timeout: int = 30
retry_count: int = 3
# ❌ Плохо — непонятные параметры
class MyPlugin(PluginProtocol):
def __init__(self, key, t, r):
...# ✅ Хорошо — обработка ошибок
class ExternalAPIPlugin(PluginProtocol):
async def call_external_api(self):
try:
return await self.client.get(...)
except ConnectionError:
logger.warning("External API unavailable")
return None # Fallback
except TimeoutError:
logger.warning("External API timeout")
return self.get_cached_data()
# ❌ Плохо — ошибки пробрасываются
class ExternalAPIPlugin(PluginProtocol):
async def call_external_api(self):
return await self.client.get(...) # Может упасть# ✅ Хорошо — ленивая инициализация
class HeavyPlugin(PluginProtocol):
def __init__(self, config):
self.config = config
self._resource = None
@property
def resource(self):
if self._resource is None:
self._resource = self._create_resource()
return self._resource
# ❌ Плохо — тяжёлая инициализация в __init__
class HeavyPlugin(PluginProtocol):
def __init__(self, config):
self.resource = self._create_resource() # Долго# ✅ Хорошо — тестирование плагина
def test_plugin_on_app_init():
plugin = MyPlugin(config=TestConfig())
app_config = {"route_handlers": []}
result = plugin.on_app_init(app_config)
assert "middleware" in result
# ❌ Плохо — тестирование только с приложением
def test_plugin():
app = Starlite(plugins=[MyPlugin()])
# Непонятно, что именно тестируетсяСистема плагинов Starlite включает:
В следующей теме мы изучим production и деплой.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.