danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: migrate notifications to the new db model

Browse source
This commit is contained in:
arabcoders 2026-01-21 18:29:16 +03:00
parent ea1fc59e88
commit 5b7d9ddbde
23 changed files with 1846 additions and 1780 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
app/features/core/schemas.py
View file

@ -19,6 +19,7 @@ class CEFeature(str, Enum):
PRESETS = "presets"
DL_FIELDS = "dl_fields"
CONDITIONS = "conditions"
NOTIFICATIONS = "notifications"
def __str__(self) -> str:
return self.value

5
app/features/notifications/deps.py Normal file
View file

@ -0,0 +1,5 @@
from app.features.notifications.repository import NotificationsRepository
def get_notifications_repo() -> NotificationsRepository:
return NotificationsRepository.get_instance()

168
app/features/notifications/migration.py Normal file
View file

@ -0,0 +1,168 @@
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from app.features.core.migration import Migration as FeatureMigration
from app.features.notifications.schemas import NotificationEvents
from app.library.config import Config
from app.library.Presets import Presets
if TYPE_CHECKING:
from app.features.notifications.repository import NotificationsRepository
LOG: logging.Logger = logging.getLogger(__name__)
class Migration(FeatureMigration):
name: str = "notifications"
def __init__(self, repo: NotificationsRepository, config: Config | None = None):
self._config: Config = config or Config.get_instance()
super().__init__(config=self._config)
self._repo: NotificationsRepository = repo
self._source_file: Path = Path(self._config.config_path) / "notifications.json"
async def should_run(self) -> bool:
return self._source_file.exists()
async def migrate(self) -> None:
if await self._repo.count() > 0:
LOG.warning("Notification targets already exist in the database; skipping migration.")
await self._move_file(self._source_file)
return
try:
items: list[dict] | None = json.loads(self._source_file.read_text())
except Exception as exc:
LOG.exception("Failed to read %s: %s. Ignoring", self._source_file, exc)
await self._move_file(self._source_file)
return
if items is None:
LOG.warning("No notification targets found in %s; skipping migration.", self._source_file)
await self._move_file(self._source_file)
return
inserted = 0
for index, item in enumerate(items):
normalized = self._normalize(item, index)
if not normalized:
continue
try:
await self._repo.create(normalized)
inserted += 1
except Exception as exc:
LOG.exception("Failed to insert notification '%s': %s", normalized.get("name"), exc)
LOG.info("Migrated %s notification target(s) from %s.", inserted, self._source_file)
await self._move_file(self._source_file)
def _normalize(self, item: Any, index: int) -> dict[str, Any] | None:
if not isinstance(item, dict):
LOG.warning("Skipping notification at index %s due to invalid type.", index)
return None
name: str | None = item.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping notification at index %s due to missing name.", index)
return None
request = item.get("request") if isinstance(item.get("request"), dict) else {}
url: str | None = request.get("url")
if not url or not isinstance(url, str):
LOG.warning("Skipping notification '%s' due to missing request url.", name)
return None
method = request.get("method") if isinstance(request.get("method"), str) else "POST"
method = method.upper()
if method not in {"POST", "PUT"}:
LOG.warning("Skipping notification '%s' due to invalid method '%s'.", name, method)
return None
req_type = request.get("type") if isinstance(request.get("type"), str) else "json"
req_type = req_type.lower()
if req_type not in {"json", "form"}:
LOG.warning("Skipping notification '%s' due to invalid type '%s'.", name, req_type)
return None
data_key = request.get("data_key") if isinstance(request.get("data_key"), str) else "data"
if not data_key:
data_key = "data"
headers = self._normalize_headers(request.get("headers"))
enabled: bool = item.get("enabled") if isinstance(item.get("enabled"), bool) else True
events = self._normalize_events(item.get("on"))
if events is None:
LOG.warning("Skipping notification '%s' due to invalid events.", name)
return None
presets = self._normalize_presets(item.get("presets"))
if presets is None:
LOG.warning("Skipping notification '%s' due to invalid presets.", name)
return None
return {
"name": name,
"on": events,
"presets": presets,
"enabled": enabled,
"request_url": url,
"request_method": method,
"request_type": req_type,
"request_data_key": data_key,
"request_headers": headers,
}
def _normalize_events(self, events: Any) -> list[str] | None:
if events is None:
return []
if not isinstance(events, list):
return []
allowed = set(NotificationEvents.get_events().values())
valid = [event for event in events if event in allowed]
invalid = [event for event in events if event not in allowed]
if len(invalid) > 0 and len(valid) < 1:
return None
if len(invalid) > 0:
LOG.warning("Dropping invalid notification events: %s", ", ".join(invalid))
return valid
def _normalize_presets(self, presets: Any) -> list[str] | None:
if presets is None:
return []
if not isinstance(presets, list):
return []
allowed = {preset.name for preset in Presets.get_instance().get_all()}
valid = [preset for preset in presets if preset in allowed]
invalid = [preset for preset in presets if preset not in allowed]
if len(invalid) > 0 and len(valid) < 1:
return None
if len(invalid) > 0:
LOG.warning("Dropping invalid notification presets: %s", ", ".join(invalid))
return valid
def _normalize_headers(self, headers: Any) -> list[dict[str, str]]:
if not isinstance(headers, list):
return []
normalized: list[dict[str, str]] = []
for header in headers:
if not isinstance(header, dict):
continue
key = str(header.get("key", "")).strip()
value = str(header.get("value", "")).strip()
if key and value:
normalized.append({"key": key, "value": value})
return normalized

39
app/features/notifications/models.py Normal file
View file

@ -0,0 +1,39 @@
from __future__ import annotations
from datetime import datetime # noqa: TC003
from sqlalchemy import JSON, Boolean, Index, Integer, String, Text
from sqlalchemy import Enum as SQLAEnum
from sqlalchemy.orm import Mapped, mapped_column
from app.features.core.models import Base, UTCDateTime, utcnow
from app.features.notifications.schemas import NotificationRequestMethod, NotificationRequestType
class NotificationModel(Base):
__tablename__: str = "notifications"
__table_args__: tuple[Index, ...] = (
Index("ix_notifications_name", "name"),
Index("ix_notifications_enabled", "enabled"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
on: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
presets: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
request_url: Mapped[str] = mapped_column(Text, nullable=False)
request_method: Mapped[NotificationRequestMethod] = mapped_column(
SQLAEnum(NotificationRequestMethod, name="notification_request_method", native_enum=False),
nullable=False,
default=NotificationRequestMethod.POST,
)
request_type: Mapped[NotificationRequestType] = mapped_column(
SQLAEnum(NotificationRequestType, name="notification_request_type", native_enum=False),
nullable=False,
default=NotificationRequestType.JSON,
)
request_data_key: Mapped[str] = mapped_column(String(255), nullable=False, default="data")
request_headers: Mapped[list[dict]] = mapped_column(JSON, nullable=False, default=list)
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, nullable=False)
updated_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, onupdate=utcnow, nullable=False)

162
app/features/notifications/repository.py Normal file
View file

@ -0,0 +1,162 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
from sqlalchemy import func, or_, select
from app.features.core.deps import get_session
from app.features.notifications.migration import Migration
from app.features.notifications.models import NotificationModel
from app.library.Singleton import Singleton
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from sqlalchemy.engine.result import Result
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.selectable import Select
LOG: logging.Logger = logging.getLogger(__name__)
class NotificationsRepository(metaclass=Singleton):
def __init__(self, session: AsyncGenerator[AsyncSession] | None = None) -> None:
self._migrated = False
self.session: AsyncGenerator[AsyncSession] = session or get_session
async def run_migrations(self) -> None:
if self._migrated:
return
self._migrated = True
await Migration(repo=self, config=None).run()
@staticmethod
def get_instance() -> NotificationsRepository:
return NotificationsRepository()
async def list(self) -> list[NotificationModel]:
async with self.session() as session:
result: Result[tuple[NotificationModel]] = await session.execute(
select(NotificationModel).order_by(NotificationModel.name.asc())
)
return list(result.scalars().all())
async def list_paginated(self, page: int, per_page: int) -> tuple[list[NotificationModel], int, int, int]:
async with self.session() as session:
total: int = await self.count()
total_pages: int = (total + per_page - 1) // per_page if total > 0 else 1
if page > total_pages and total > 0:
page = total_pages
query: Select[tuple[NotificationModel]] = (
select(NotificationModel)
.order_by(NotificationModel.name.asc())
.limit(per_page)
.offset((page - 1) * per_page)
)
result: Result[tuple[NotificationModel]] = await session.execute(query)
return list(result.scalars().all()), total, page, total_pages
async def count(self) -> int:
async with self.session() as session:
result: Result[tuple[int]] = await session.execute(select(func.count()).select_from(NotificationModel))
return int(result.scalar_one())
async def get(self, identifier: int | str) -> NotificationModel | None:
async with self.session() as session:
if not identifier:
return None
if isinstance(identifier, int):
clause: ColumnElement[bool] = NotificationModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier)
else:
clause = NotificationModel.name == identifier
result: Result[tuple[NotificationModel]] = await session.execute(
select(NotificationModel).where(clause).limit(1)
)
return result.scalar_one_or_none()
async def get_by_name(self, name: str, exclude_id: int | None = None) -> NotificationModel | None:
async with self.session() as session:
query: Select[tuple[NotificationModel]] = select(NotificationModel).where(NotificationModel.name == name)
if exclude_id is not None:
query = query.where(NotificationModel.id != exclude_id)
result: Result[tuple[NotificationModel]] = await session.execute(query.limit(1))
return result.scalar_one_or_none()
async def create(self, payload: NotificationModel | dict) -> NotificationModel:
async with self.session() as session:
model: NotificationModel = NotificationModel(**payload) if isinstance(payload, dict) else payload
if model.id is not None:
model.id = None
if await self.get_by_name(name=model.name) is not None:
msg: str = f"Notification target with name '{model.name}' already exists."
raise ValueError(msg)
session.add(model)
await session.commit()
await session.refresh(model)
return model
async def update(self, identifier: int | str, payload: dict[str, Any]) -> NotificationModel:
async with self.session() as session:
if isinstance(identifier, int):
clause: ColumnElement[bool] = NotificationModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier)
else:
clause = NotificationModel.name == identifier
result: Result[tuple[NotificationModel]] = await session.execute(
select(NotificationModel).where(clause).limit(1)
)
model: NotificationModel | None = result.scalar_one_or_none()
if not model:
msg: str = f"Notification target '{identifier}' not found."
raise KeyError(msg)
payload.pop("id", None)
if "name" in payload and await self.get_by_name(name=payload["name"], exclude_id=model.id) is not None:
msg = f"Notification target with name '{payload['name']}' already exists."
raise ValueError(msg)
for key, value in payload.items():
if hasattr(model, key):
setattr(model, key, value)
await session.commit()
await session.refresh(model)
return model
async def delete(self, identifier: int | str) -> NotificationModel:
async with self.session() as session:
if isinstance(identifier, int):
clause: ColumnElement[bool] = NotificationModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier)
else:
clause = NotificationModel.name == identifier
result: Result[tuple[NotificationModel]] = await session.execute(
select(NotificationModel).where(clause).limit(1)
)
model = result.scalar_one_or_none()
if not model:
msg: str = f"Notification target '{identifier}' not found."
raise KeyError(msg)
await session.delete(model)
await session.commit()
return model

218
app/features/notifications/router.py Normal file
View file

@ -0,0 +1,218 @@
import logging
from typing import Any
from aiohttp import web
from aiohttp.web import Request, Response
from pydantic import ValidationError
from app.features.core.schemas import CEAction, CEFeature, ConfigEvent, Pagination
from app.features.core.utils import build_pagination, format_validation_errors, normalize_pagination
from app.features.notifications.schemas import Notification, NotificationEvents, NotificationList, NotificationPatch
from app.features.notifications.service import Notifications
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.router import route
LOG: logging.Logger = logging.getLogger(__name__)
def _model(model: Any) -> Notification:
return Notifications.get_instance().model_to_schema(model)
def _serialize(model: Any) -> dict:
return _model(model).model_dump()
@route("GET", "api/notifications/", "notifications_list")
async def notifications_list(request: Request, encoder: Encoder) -> Response:
page, per_page = normalize_pagination(request)
items, total, current_page, total_pages = await Notifications.get_instance().list_paginated(page, per_page)
return web.json_response(
data=NotificationList(
items=[_model(model) for model in items],
pagination=Pagination.model_validate(build_pagination(total, current_page, per_page, total_pages)),
),
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("GET", "api/notifications/events/", "notifications_events")
async def notifications_events(encoder: Encoder) -> Response:
return web.json_response(
data={"events": list(NotificationEvents.get_events().values())},
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("POST", "api/notifications/test/", "notification_test")
async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("POST", "api/notifications/", "notification_add")
async def notifications_add(request: Request, encoder: Encoder, notify: EventBus) -> Response:
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
item: Notification = Notification.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
try:
saved = _serialize(await Notifications.get_instance().create(item))
except ValueError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code)
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.CREATE, data=saved),
)
return web.json_response(data=saved, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("GET", "api/notifications/{id}", "notification_get")
async def notifications_get(request: Request, encoder: Encoder) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await Notifications.get_instance().get(identifier)):
return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code)
return web.json_response(data=_serialize(model), status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("DELETE", "api/notifications/{id}", "notification_delete")
async def notifications_delete(request: Request, encoder: Encoder, notify: EventBus) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
try:
deleted = _serialize(await Notifications.get_instance().delete(identifier))
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.DELETE, data=deleted),
)
return web.json_response(
data=deleted,
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
except KeyError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code)
@route("PATCH", "api/notifications/{id}", "notification_patch")
async def notifications_patch(request: Request, encoder: Encoder, notify: EventBus) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await Notifications.get_instance().get(identifier)):
return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code)
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
validated = NotificationPatch.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
current = _model(model).model_dump()
if validated.name is not None:
current["name"] = validated.name
if validated.on is not None:
current["on"] = validated.on
if validated.presets is not None:
current["presets"] = validated.presets
if validated.enabled is not None:
current["enabled"] = validated.enabled
if validated.request is not None:
request_payload = current.get("request", {})
if validated.request.url is not None:
request_payload["url"] = validated.request.url
if validated.request.method is not None:
request_payload["method"] = validated.request.method
if validated.request.type is not None:
request_payload["type"] = validated.request.type
if validated.request.headers is not None:
request_payload["headers"] = [header.model_dump() for header in validated.request.headers]
if validated.request.data_key is not None:
request_payload["data_key"] = validated.request.data_key
current["request"] = request_payload
try:
updated = _serialize(await Notifications.get_instance().update(model.id, current))
except ValueError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code)
except KeyError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code)
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.UPDATE, data=updated),
)
return web.json_response(
data=updated,
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("PUT", "api/notifications/{id}", "notification_update")
async def notifications_update(request: Request, encoder: Encoder, notify: EventBus) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await Notifications.get_instance().get(identifier)):
return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code)
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
validated = Notification.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
try:
updated = _serialize(await Notifications.get_instance().update(model.id, validated))
except ValueError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code)
except KeyError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code)
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.UPDATE, data=updated),
)
return web.json_response(data=updated, status=web.HTTPOk.status_code, dumps=encoder.encode)

145
app/features/notifications/schemas.py Normal file
View file

@ -0,0 +1,145 @@
from __future__ import annotations
from enum import Enum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.features.core.schemas import Pagination
from app.library.Events import Events
class NotificationEvents:
TEST: str = Events.TEST
ITEM_ADDED: str = Events.ITEM_ADDED
ITEM_COMPLETED: str = Events.ITEM_COMPLETED
ITEM_CANCELLED: str = Events.ITEM_CANCELLED
ITEM_DELETED: str = Events.ITEM_DELETED
ITEM_PAUSED: str = Events.ITEM_PAUSED
ITEM_RESUMED: str = Events.ITEM_RESUMED
ITEM_MOVED: str = Events.ITEM_MOVED
PAUSED: str = Events.PAUSED
RESUMED: str = Events.RESUMED
LOG_INFO: str = Events.LOG_INFO
LOG_SUCCESS: str = Events.LOG_SUCCESS
LOG_WARNING: str = Events.LOG_WARNING
LOG_ERROR: str = Events.LOG_ERROR
TASK_DISPATCHED: str = Events.TASK_DISPATCHED
@staticmethod
def get_events() -> dict[str, str]:
return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)}
@staticmethod
def events() -> list[str]:
return [
getattr(NotificationEvents, ev)
for ev in dir(NotificationEvents)
if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev))
]
@staticmethod
def is_valid(event: str) -> bool:
return event in NotificationEvents.get_events().values()
class NotificationRequestType(str, Enum):
JSON = "json"
FORM = "form"
def __str__(self) -> str:
return self.value
class NotificationRequestMethod(str, Enum):
POST = "POST"
PUT = "PUT"
def __str__(self) -> str:
return self.value
class NotificationRequestHeader(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
key: str = Field(min_length=1)
value: str = Field(min_length=1)
class NotificationRequest(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
type: NotificationRequestType = NotificationRequestType.JSON
method: NotificationRequestMethod = NotificationRequestMethod.POST
url: str = Field(min_length=1)
headers: list[NotificationRequestHeader] = Field(default_factory=list)
data_key: str = Field(default="data", min_length=1)
@field_validator("method", mode="before")
@classmethod
def _normalize_method(cls, value: Any) -> Any:
if value is None:
return value
if isinstance(value, NotificationRequestMethod):
return value
return str(value).upper()
@field_validator("type", mode="before")
@classmethod
def _normalize_type(cls, value: Any) -> Any:
if value is None:
return value
if isinstance(value, NotificationRequestType):
return value
return str(value).lower()
class NotificationRequestPatch(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
type: NotificationRequestType | None = None
method: NotificationRequestMethod | None = None
url: str | None = None
headers: list[NotificationRequestHeader] | None = None
data_key: str | None = None
class Notification(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
id: int | None = None
name: str = Field(min_length=1)
on: list[str] = Field(default_factory=list)
presets: list[str] = Field(default_factory=list)
enabled: bool = True
request: NotificationRequest
@field_validator("on", "presets", mode="before")
@classmethod
def _normalize_list(cls, value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(item).strip() for item in value if str(item).strip()]
return [str(value).strip()]
class NotificationPatch(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
name: str | None = None
on: list[str] | None = None
presets: list[str] | None = None
enabled: bool | None = None
request: NotificationRequestPatch | None = None
class NotificationList(BaseModel):
model_config = ConfigDict(from_attributes=True)
items: list[Notification] = Field(default_factory=list)
pagination: Pagination

347
app/features/notifications/service.py Normal file
View file

@ -0,0 +1,347 @@
from __future__ import annotations
import asyncio
import logging
import traceback
from pathlib import Path
from typing import TYPE_CHECKING, Any
from app.features.notifications.models import NotificationModel
from app.features.notifications.repository import NotificationsRepository
from app.features.notifications.schemas import (
Notification,
NotificationEvents,
NotificationRequest,
NotificationRequestHeader,
NotificationRequestType,
)
from app.library.BackgroundWorker import BackgroundWorker
from app.library.config import Config
from app.library.encoder import Encoder
from app.library.Events import Event, EventBus, Events
from app.library.httpx_client import async_client
from app.library.ItemDTO import Item, ItemDTO
from app.library.Presets import Preset, Presets
from app.library.Singleton import Singleton
if TYPE_CHECKING:
from collections.abc import Awaitable
import httpx
from aiohttp import web
LOG: logging.Logger = logging.getLogger("feature.notifications")
class Notifications(metaclass=Singleton):
def __init__(
self,
repo: NotificationsRepository | None = None,
client: httpx.AsyncClient | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
background_worker: BackgroundWorker | None = None,
) -> None:
self._repo: NotificationsRepository = repo or NotificationsRepository.get_instance()
config = config or Config.get_instance()
self._debug: bool = config.debug
self._version: str = config.app_version
self._client: httpx.AsyncClient = client or async_client()
self._encoder: Encoder = encoder or Encoder()
self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance()
@staticmethod
def get_instance() -> Notifications:
return Notifications()
async def on_shutdown(self, _: web.Application) -> None:
pass
def attach(self, _: web.Application) -> None:
async def handle_event(_, __):
await self._repo.run_migrations()
EventBus.get_instance().subscribe(Events.STARTED, handle_event, "NotificationsRepository.run_migrations")
EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit")
async def list(self) -> list[NotificationModel]:
return await self._repo.list()
async def list_paginated(self, page: int, per_page: int) -> tuple[list[NotificationModel], int, int, int]:
return await self._repo.list_paginated(page, per_page)
async def get(self, identifier: int | str) -> NotificationModel | None:
return await self._repo.get(identifier)
async def create(self, item: Notification | dict) -> NotificationModel:
if not isinstance(item, Notification):
item = Notification.model_validate(item)
normalized = self._normalize(item)
payload = self._payload_from_schema(normalized)
return await self._repo.create(payload)
async def update(self, identifier: int | str, payload: Notification | dict) -> NotificationModel:
if not isinstance(payload, Notification):
payload = Notification.model_validate(payload)
normalized = self._normalize(payload)
update_payload = self._payload_from_schema(normalized)
return await self._repo.update(identifier, update_payload)
async def delete(self, identifier: int | str) -> NotificationModel:
return await self._repo.delete(identifier)
async def send(self, ev: Event, wait: bool = True) -> list[dict] | list[Awaitable[dict]]:
targets = await self._repo.list()
if len(targets) < 1:
return []
if not isinstance(ev.data, (ItemDTO, Item, dict)):
LOG.debug("Received invalid item type '%s' with event '%s'.", type(ev.data), ev.event)
return []
tasks: list[Awaitable[dict]] = []
apprise_targets: list[Notification] = []
for target_model in targets:
target = self.model_to_schema(target_model)
if not target.enabled:
continue
if len(target.on) > 0 and ev.event not in target.on and NotificationEvents.TEST != ev.event:
continue
if NotificationEvents.TEST != ev.event and not self._check_preset(target, ev):
continue
if not target.request.url.startswith("http"):
apprise_targets.append(target)
else:
tasks.append(self._send(target, ev))
if len(apprise_targets) > 0:
tasks.append(self._apprise(apprise_targets, ev))
if wait:
return await asyncio.gather(*tasks)
return tasks
def emit(self, e: Event, _, **__) -> None:
if not NotificationEvents.is_valid(e.event):
return
self._offload.submit(self.send, e)
def _normalize(self, item: Notification) -> Notification:
if item.enabled is not None and not isinstance(item.enabled, bool):
msg: str = "Enabled must be a boolean."
raise ValueError(msg)
item.on = self._filter_events(item.on)
item.presets = self._filter_presets(item.presets)
item.request.headers = [
NotificationRequestHeader(key=header.key, value=header.value)
for header in item.request.headers
if header.key and header.value
]
return item
def _filter_events(self, events: list[str]) -> list[str]:
if not events:
return []
allowed = set(NotificationEvents.get_events().values())
valid = [event for event in events if event in allowed]
invalid = [event for event in events if event not in allowed]
if len(invalid) > 0 and len(valid) < 1:
msg: str = f"Invalid notification target. Invalid events '{', '.join(invalid)}' found."
raise ValueError(msg)
if len(invalid) > 0:
LOG.warning("Dropping invalid notification events: %s", ", ".join(invalid))
return valid
def _filter_presets(self, presets: list[str]) -> list[str]:
if not presets:
return []
all_presets: list[Preset] = Presets.get_instance().get_all()
allowed = {preset.name for preset in all_presets}
valid = [preset for preset in presets if preset in allowed]
invalid = [preset for preset in presets if preset not in allowed]
if len(invalid) > 0 and len(valid) < 1:
msg: str = f"Invalid notification target. Invalid presets '{', '.join(invalid)}' found."
raise ValueError(msg)
if len(invalid) > 0:
LOG.warning("Dropping invalid notification presets: %s", ", ".join(invalid))
return valid
def model_to_schema(self, model: NotificationModel) -> Notification:
headers: list[NotificationRequestHeader] = []
if isinstance(model.request_headers, list):
for header in model.request_headers:
if not isinstance(header, dict):
continue
key = str(header.get("key", "")).strip()
value = str(header.get("value", "")).strip()
if key and value:
headers.append(NotificationRequestHeader(key=key, value=value))
return Notification(
id=model.id,
name=model.name,
on=list(model.on or []),
presets=list(model.presets or []),
enabled=model.enabled,
request=NotificationRequest.model_validate(
{
"type": model.request_type,
"method": model.request_method,
"url": model.request_url,
"headers": headers,
"data_key": model.request_data_key,
}
),
)
def _payload_from_schema(self, item: Notification) -> dict[str, Any]:
return {
"name": item.name,
"on": list(item.on),
"presets": list(item.presets),
"enabled": item.enabled,
"request_url": item.request.url,
"request_method": str(item.request.method),
"request_type": str(item.request.type),
"request_data_key": item.request.data_key,
"request_headers": [header.model_dump() for header in item.request.headers],
}
def _check_preset(self, target: Notification, ev: Event) -> bool:
if len(target.presets) < 1:
return True
if not isinstance(ev.data, (Item, ItemDTO, dict)):
return False
preset_name: str | None = None
if isinstance(ev.data, Item):
preset_name = ev.data.preset
if isinstance(ev.data, ItemDTO):
preset_name = ev.data.preset
if isinstance(ev.data, dict):
preset_name = ev.data.get("preset", None)
if not preset_name:
return False
return preset_name in target.presets
async def _apprise(self, targets: list[Notification], ev: Event) -> dict:
if not targets:
return {}
import apprise
try:
notify = apprise.Apprise(debug=self._debug)
apr_config = Path(Config.get_instance().apprise_config)
if apr_config.exists():
apprise_file = apprise.AppriseConfig()
apprise_file.add(str(apr_config))
notify.add(apprise_file)
for target in targets:
notify.add(target.request.url)
status = await notify.async_notify(
body=ev.message or self._encoder.encode(ev.serialize()),
title=ev.title or f"YTPTube Event: {ev.event}",
)
if not status:
msg = "Apprise failed to send notification."
raise RuntimeError(msg) # noqa: TRY301
except Exception as exc:
LOG.exception(exc)
LOG.error("Error sending Apprise notification: %s", exc)
return {"error": str(exc), "event": ev.event, "id": ev.id, "targets": [t.name for t in targets]}
return {}
async def _send(self, target: Notification, ev: Event) -> dict:
try:
LOG.info("Sending notification event '%s: %s' to '%s'.", ev.event, ev.id, target.name)
headers: dict[str, str] = {
"User-Agent": f"YTPTube/{self._version}",
"X-Event-Id": ev.id,
"X-Event": ev.event,
"Content-Type": "application/json"
if NotificationRequestType.JSON == target.request.type
else "application/x-www-form-urlencoded",
}
if len(target.request.headers) > 0:
headers.update({h.key: h.value for h in target.request.headers if h.key and h.value})
payload_data: dict[str, Any] = ev.serialize()
if "data" != target.request.data_key:
payload_data[target.request.data_key] = payload_data["data"]
payload_data.pop("data", None)
if NotificationRequestType.FORM == target.request.type:
payload_data[target.request.data_key] = self._encoder.encode(payload_data[target.request.data_key])
data_payload: dict[str, Any] | None = payload_data
content_payload: str | None = None
else:
data_payload = None
content_payload = self._encoder.encode(payload_data)
response = await self._client.request(
method=str(target.request.method).upper(),
url=target.request.url,
headers=headers,
data=data_payload,
content=content_payload,
)
resp_data: dict[str, Any] = {
"url": target.request.url,
"status": response.status_code,
"text": response.text,
}
msg: str = (
f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' "
f"with status '{response.status_code}'."
)
if self._debug and resp_data.get("text"):
msg += f" body '{resp_data.get('text', '??')}'."
LOG.info(msg)
return resp_data
except Exception as exc:
err_msg = str(exc) or type(exc).__name__
tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
LOG.error(
"Error sending notification event '%s: %s' to '%s'. '%s'. %s",
ev.event,
ev.id,
target.name,
err_msg,
tb,
)
return {"url": target.request.url, "status": 500, "text": str(ev)}

154
app/features/notifications/tests/test_notifications_repository.py Normal file
View file

@ -0,0 +1,154 @@
"""Tests for NotificationsRepository."""
from __future__ import annotations
import pytest
import pytest_asyncio
from app.features.notifications.repository import NotificationsRepository
from app.library.sqlite_store import SqliteStore
@pytest_asyncio.fixture
async def repo(tmp_path):
"""Provide a fresh repository instance with initialized database for each test."""
NotificationsRepository._reset_singleton()
SqliteStore._reset_singleton()
store = SqliteStore(db_path=":memory:")
await store.get_connection()
repository = NotificationsRepository.get_instance()
yield repository
if store._conn:
await store._conn.close()
if store._engine:
await store._engine.dispose()
NotificationsRepository._reset_singleton()
SqliteStore._reset_singleton()
class TestNotificationsRepository:
"""Test suite for NotificationsRepository database operations."""
@pytest.mark.asyncio
async def test_list_empty(self, repo):
"""List returns empty when no notifications exist."""
notifications = await repo.list()
assert notifications == [], "Should return empty list when no notifications exist"
@pytest.mark.asyncio
async def test_create_notification(self, repo):
"""Create notification with valid data."""
payload = {
"name": "Webhook",
"on": ["item_completed"],
"presets": [],
"enabled": True,
"request_url": "https://example.com/webhook",
"request_method": "POST",
"request_type": "json",
"request_data_key": "data",
"request_headers": [{"key": "Authorization", "value": "Bearer token"}],
}
model = await repo.create(payload)
assert model.id is not None, "Should generate ID for new notification"
assert model.name == "Webhook", "Should store name correctly"
assert model.request_url == "https://example.com/webhook", "Should store request url"
assert model.request_method == "POST", "Should store request method"
@pytest.mark.asyncio
async def test_get_by_id(self, repo):
"""Get notification by integer ID."""
created = await repo.create(
{
"name": "Get Test",
"on": [],
"presets": [],
"enabled": True,
"request_url": "https://example.com",
"request_method": "POST",
"request_type": "json",
"request_data_key": "data",
"request_headers": [],
}
)
retrieved = await repo.get(created.id)
assert retrieved is not None, "Should retrieve created notification"
assert retrieved.id == created.id, "Should match created notification id"
@pytest.mark.asyncio
async def test_update_notification(self, repo):
"""Update existing notification."""
created = await repo.create(
{
"name": "Update Test",
"on": [],
"presets": [],
"enabled": True,
"request_url": "https://example.com",
"request_method": "POST",
"request_type": "json",
"request_data_key": "data",
"request_headers": [],
}
)
updated = await repo.update(created.id, {"name": "Updated Name", "enabled": False})
assert updated.name == "Updated Name", "Should update name"
assert updated.enabled is False, "Should update enabled flag"
@pytest.mark.asyncio
async def test_delete_notification(self, repo):
"""Delete existing notification."""
created = await repo.create(
{
"name": "Delete Test",
"on": [],
"presets": [],
"enabled": True,
"request_url": "https://example.com",
"request_method": "POST",
"request_type": "json",
"request_data_key": "data",
"request_headers": [],
}
)
deleted = await repo.delete(created.id)
assert deleted.id == created.id, "Should return deleted notification"
assert await repo.get(created.id) is None, "Deleted notification should not be retrievable"
@pytest.mark.asyncio
async def test_list_paginated(self, repo):
"""List paginated returns correct subset."""
for i in range(4):
await repo.create(
{
"name": f"Item {i}",
"on": [],
"presets": [],
"enabled": True,
"request_url": "https://example.com",
"request_method": "POST",
"request_type": "json",
"request_data_key": "data",
"request_headers": [],
}
)
items, total, page, total_pages = await repo.list_paginated(page=1, per_page=2)
assert len(items) == 2, "Should return 2 items per page"
assert total == 4, "Should report total count of 4"
assert page == 1, "Should be on page 1"
assert total_pages == 2, "Should have 2 pages total"

559
app/library/Notifications.py
View file

@ -1,559 +0,0 @@
import asyncio
import json
import logging
import traceback
from collections.abc import Awaitable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import httpx
from aiohttp import web
from .ag_utils import ag
from .BackgroundWorker import BackgroundWorker
from .config import Config
from .encoder import Encoder
from .Events import Event, EventBus, Events
from .httpx_client import async_client
from .ItemDTO import Item, ItemDTO
from .Presets import Preset, Presets
from .Singleton import Singleton
from .Utils import validate_uuid
LOG = logging.getLogger("notifications")
@dataclass(kw_only=True)
class TargetRequestHeader:
"""Request header details for a notification target."""
key: str
value: str
def serialize(self) -> dict:
return {"key": self.key, "value": self.value}
def json(self) -> str:
return Encoder().encode(self.serialize())
def get(self, key: str, default: Any = None) -> Any:
return self.serialize().get(key, default)
@dataclass(kw_only=True)
class TargetRequest:
"""Request details for a notification target."""
type: str
method: str
url: str
headers: list[TargetRequestHeader] = field(default_factory=list)
data_key: str = "data"
def serialize(self) -> dict:
return {
"type": self.type,
"method": self.method,
"url": self.url,
"data_key": self.data_key,
"headers": [h.serialize() for h in self.headers],
}
def json(self) -> str:
return Encoder().encode(self.serialize())
def get(self, key: str, default: Any = None) -> Any:
return ag(self.serialize(), key, default)
@dataclass(kw_only=True)
class Target:
"""Notification target details."""
id: str
name: str
on: list[str] = field(default_factory=list)
presets: list[str] = field(default_factory=list)
request: TargetRequest
enabled: bool = True
def serialize(self) -> dict:
return {
"id": self.id,
"name": self.name,
"on": self.on,
"presets": self.presets,
"request": self.request.serialize(),
"enabled": self.enabled,
}
def json(self) -> str:
return Encoder().encode(self.serialize())
def get(self, key: str, default: Any = None) -> Any:
return self.serialize().get(key, default)
class NotificationEvents:
TEST: str = Events.TEST
ITEM_ADDED: str = Events.ITEM_ADDED
ITEM_COMPLETED: str = Events.ITEM_COMPLETED
ITEM_CANCELLED: str = Events.ITEM_CANCELLED
ITEM_DELETED: str = Events.ITEM_DELETED
ITEM_PAUSED: str = Events.ITEM_PAUSED
ITEM_RESUMED: str = Events.ITEM_RESUMED
ITEM_MOVED: str = Events.ITEM_MOVED
PAUSED: str = Events.PAUSED
RESUMED: str = Events.RESUMED
LOG_INFO: str = Events.LOG_INFO
LOG_SUCCESS: str = Events.LOG_SUCCESS
LOG_WARNING: str = Events.LOG_WARNING
LOG_ERROR: str = Events.LOG_ERROR
TASK_DISPATCHED: str = Events.TASK_DISPATCHED
@staticmethod
def get_events() -> dict[str, str]:
return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)}
def events() -> list:
return [
getattr(NotificationEvents, ev)
for ev in dir(NotificationEvents)
if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev))
]
@staticmethod
def is_valid(event: str) -> bool:
return event in NotificationEvents.get_events().values()
class Notification(metaclass=Singleton):
def __init__(
self,
file: str | None = None,
client: httpx.AsyncClient | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
background_worker: BackgroundWorker | None = None,
):
self._targets: list[Target] = []
"Notification targets to send events to."
config: Config = config or Config.get_instance()
self._debug: bool = config.debug
"Debug mode."
self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json")
"File to store notification targets."
self._client: httpx.AsyncClient = client or async_client()
"HTTP client to send requests."
self._encoder: Encoder = encoder or Encoder()
"Encoder to encode data."
self._version: str = config.app_version
"Application version."
self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance()
"Background worker to offload tasks to."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
self._file.chmod(0o600)
except Exception:
pass
@staticmethod
def get_instance(
file: str | None = None,
client: httpx.AsyncClient | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
background_worker: BackgroundWorker | None = None,
) -> "Notification":
"""
Get the instance of the class.
"""
return Notification(
file=file, client=client, encoder=encoder, config=config, background_worker=background_worker
)
def attach(self, _: web.Application):
"""
Attach the class to the application.
Args:
_ (web.Application): The aiohttp application.
"""
self.load()
EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit")
def get_targets(self) -> list[Target]:
"""Get the list of notification targets."""
return self._targets
def clear(self) -> "Notification":
"""Clear the list of notification targets."""
self._targets.clear()
return self
def save(self, targets: list[Target]) -> "Notification":
"""
Save notification targets to the file.
Args:
targets (list[Target]|None): The list of targets to save.
Returns:
Notification: The Notification instance.
"""
try:
self._file.write_text(json.dumps([t.serialize() for t in targets], indent=4))
LOG.info(f"Updated '{self._file}'.")
except Exception as e:
LOG.exception(e)
LOG.error(f"Error saving '{self._file}'. '{e!s}'")
return self
def load(self) -> "Notification":
"""Load or reload notification targets from the file."""
if len(self._targets) > 0:
self.clear()
if not self._file.exists() or self._file.stat().st_size < 1:
return self
targets: list = []
try:
LOG.info(f"Loading '{self._file}'.")
targets = json.loads(self._file.read_text())
except Exception as e:
LOG.error(f"Error loading '{self._file}'. '{e!s}'")
if not targets or len(targets) < 1:
LOG.debug(f"No targets were found in '{self._file}'.")
return self
need_save = False
for target in targets:
try:
if "enabled" not in target:
target["enabled"] = True
need_save = True
try:
Notification.validate(target)
target: Target = self.make_target(target)
except ValueError as e:
name = target.get("name") or target.get("id") or target.get("request", {}).get("url") or "unknown"
LOG.error(f"Invalid notification target '{name}'. '{e!s}'")
continue
self._targets.append(target)
LOG.info(
f"Send '{target.request.type}' request on '{', '.join(target.on) if len(target.on) > 0 else 'all events'}' to '{target.name}'."
)
except Exception as e:
LOG.error(f"Error loading notification target '{target}'. '{e!s}'")
if need_save:
LOG.warning("Saving notifications due to schema changes.")
self.save(self._targets)
return self
def make_target(self, target: dict) -> Target:
"""
Make a notification target from a dictionary.
Args:
target (dict): The target details.
Returns:
Target: The notification target.
"""
return Target(
id=target.get("id"),
name=target.get("name"),
on=target.get("on", []),
presets=target.get("presets", []),
enabled=target.get("enabled", True),
request=TargetRequest(
type=target.get("request", {}).get("type", "json"),
method=target.get("request", {}).get("method", "POST"),
url=target.get("request", {}).get("url"),
data_key=target.get("request", {}).get("data_key", "data"),
headers=[
TargetRequestHeader(
key=str(h.get("key", "")).strip(),
value=str(h.get("value", "")).strip(),
)
for h in target.get("request", {}).get("headers", [])
],
),
)
@staticmethod
def validate(target: Target | dict) -> bool:
"""
Validate a notification target.
Args:
target (Target|dict): The target to validate.
Returns:
bool: True if the target is valid, False otherwise.
"""
if not isinstance(target, dict):
target = target.serialize()
if "id" not in target or validate_uuid(target["id"], version=4) is False:
msg = "Invalid notification target. No ID found."
raise ValueError(msg)
if "name" not in target:
msg = "Invalid notification target. No name found."
raise ValueError(msg)
if "request" not in target:
msg = "Invalid notification target. No request details found."
raise ValueError(msg)
if "url" not in target["request"]:
msg = "Invalid notification target. No URL found."
raise ValueError(msg)
if "data_key" not in target["request"]:
target["request"]["data_key"] = "data"
if "enabled" not in target:
target["enabled"] = True
if target.get("enabled") is not None and not isinstance(target.get("enabled"), bool):
msg = "Enabled must be a boolean."
raise ValueError(msg)
if "method" in target["request"] and target["request"]["method"].upper() not in ["POST", "PUT"]:
msg = "Invalid notification target. Invalid method found."
raise ValueError(msg)
if "type" in target["request"] and target["request"]["type"].lower() not in ["json", "form"]:
msg = "Invalid notification target. Invalid type found."
raise ValueError(msg)
if "on" in target:
if not isinstance(target["on"], list):
msg = "Invalid notification target. Invalid 'on' event list found."
raise ValueError(msg)
removed_events: list = []
all_events: dict[str, str] = NotificationEvents.get_events().values()
for e in target["on"]:
if e not in all_events:
removed_events.append(e)
target["on"].remove(e)
continue
if len(removed_events) > 0 and len(target["on"]) < 1:
msg: str = f"Invalid notification target. Invalid events '{', '.join(removed_events)}' found."
raise ValueError(msg)
if "presets" in target:
if not isinstance(target["presets"], list):
msg = "Invalid notification target. Invalid 'presets' list found."
raise ValueError(msg)
removed_presets: list = []
all_presets: list[Preset] = Presets.get_instance().get_all()
for p in target["presets"]:
if p not in [ap.name for ap in all_presets]:
removed_presets.append(p)
target["presets"].remove(p)
continue
if len(removed_presets) > 0 and len(target["presets"]) < 1:
msg: str = f"Invalid notification target. Invalid presets '{', '.join(removed_presets)}' found."
raise ValueError(msg)
if "headers" in target["request"]:
if not isinstance(target["request"]["headers"], list):
msg = "Invalid notification target. Invalid headers list found."
raise ValueError(msg)
for h in target["request"]["headers"]:
if "key" not in h:
msg = "Invalid notification target. No header key found."
raise ValueError(msg)
if "value" not in h:
msg = "Invalid notification target. No header value found."
raise ValueError(msg)
return True
async def send(self, ev: Event, wait: bool = True) -> list[dict] | Awaitable[list[dict]]:
if len(self._targets) < 1:
return []
if not isinstance(ev.data, ItemDTO) and not isinstance(ev.data, dict):
LOG.debug(f"Received invalid item type '{type(ev.data)}' with event '{ev.event}'.")
return []
tasks = []
apprise_targets: list[Target] = []
for target in self._targets:
if not target.enabled:
continue
if len(target.on) > 0 and ev.event not in target.on and "test" != ev.event:
continue
if "test" != ev.event and not self._check_preset(target, ev):
continue
if not target.request.url.startswith("http"):
apprise_targets.append(target)
else:
tasks.append(self._send(target, ev))
if len(apprise_targets) > 0:
tasks.append(self._apprise(apprise_targets, ev))
if wait:
return await asyncio.gather(*tasks)
return tasks
def _check_preset(self, target: Target, ev: Event) -> bool:
if len(target.presets) < 1:
return True
if not isinstance(ev.data, (Item, ItemDTO, dict)):
return False
preset_name: str | None = None
if isinstance(ev.data, Item):
preset_name = ev.data.preset
if isinstance(ev.data, ItemDTO):
preset_name = ev.data.preset
if isinstance(ev.data, dict):
preset_name = ev.data.get("preset", None)
if not preset_name:
return False
return preset_name in target.presets
async def _apprise(self, target: list[Target], ev: Event) -> dict:
if not target or not isinstance(target, list):
return {}
import apprise
try:
notify = apprise.Apprise(debug=self._debug)
apr_config = Path(Config.get_instance().apprise_config)
if apr_config.exists():
apprise_config = notify.AppriseConfig()
apprise_config.add(apr_config)
notify.add(apprise_config)
for t in target:
notify.add(t.request.url)
status = await notify.async_notify(
body=ev.message or self._encoder.encode(ev.serialize()),
title=ev.title or f"YTPTube Event: {ev.event}",
)
if not status:
msg = "Apprise failed to send notification."
raise RuntimeError(msg) # noqa: TRY301
except Exception as e:
LOG.exception(e)
LOG.error(f"Error sending Apprise notification: {e!s}")
return {"error": str(e), "event": ev.event, "id": ev.id, "targets": [t.name for t in target]}
return {}
async def _send(self, target: Target, ev: Event) -> dict:
try:
LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.")
headers: dict[str, str] = {
"User-Agent": f"YTPTube/{self._version}",
"X-Event-Id": ev.id,
"X-Event": ev.event,
"Content-Type": "application/json"
if "json" == target.request.type.lower()
else "application/x-www-form-urlencoded",
}
if len(target.request.headers) > 0:
headers.update({h.key: h.value for h in target.request.headers if h.key and h.value})
payload: dict = ev.serialize()
if "data" != target.request.data_key:
payload[target.request.data_key] = payload["data"]
payload.pop("data", None)
if "form" == target.request.type.lower():
payload[target.request.data_key] = self._encoder.encode(payload[target.request.data_key])
else:
payload = self._encoder.encode(payload)
response = await self._client.request(
method=target.request.method.upper(),
url=target.request.url,
headers=headers,
data=payload if "form" == target.request.type.lower() else None,
content=payload if "json" == target.request.type.lower() else None,
)
respData: dict[str, Any] = {
"url": target.request.url,
"status": response.status_code,
"text": response.text,
}
msg: str = f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' with status '{response.status_code}'."
if self._debug and respData.get("text"):
msg += f" body '{respData.get('text', '??')}'."
LOG.info(msg)
return respData
except Exception as e:
err_msg = str(e)
if not err_msg:
err_msg: str = type(e).__name__
tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'. {tb}")
return {"url": target.request.url, "status": 500, "text": str(ev)}
def emit(self, e: Event, _, **__) -> None:
if len(self._targets) < 1 or not NotificationEvents.is_valid(e.event):
return
self._offload.submit(self.send, e)
return
async def noop(self) -> None:
return None

28
app/library/router.py
View file

@ -1,8 +1,9 @@
import logging
import re
from collections.abc import Awaitable
from collections.abc import Awaitable, Callable
from enum import Enum
from functools import wraps
from typing import Any
LOG: logging.Logger = logging.getLogger(__name__)
@ -25,15 +26,15 @@ class Route:
method (str): The HTTP method (GET, POST, etc.).
path (str): The path for the route.
name (str): The name of the route.
handler (Awaitable): The function that handles the route.
handler (Callable[..., Awaitable]): The function that handles the route.
"""
def __init__(self, method: str, path: str, name: str, handler: Awaitable):
def __init__(self, method: str, path: str, name: str, handler: Callable[..., Awaitable[Any]]):
self.method: str = method.upper()
self.path: str = path
self.name: str = name
self.handler: Awaitable = handler
self.handler: Callable[..., Awaitable[Any]] = handler
ROUTES: dict[str, dict[str, Route]] = {}
@ -55,7 +56,12 @@ def make_route_name(method: str, path: str) -> str:
return f"{method}:" + ".".join(segments or ["root"])
def route(method: RouteType | str | list[str], path: str, name: str | None = None, **kwargs) -> Awaitable:
def route(
method: RouteType | str | list[str],
path: str,
name: str | None = None,
**kwargs,
) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
"""
Decorator to mark a method as an HTTP route handler.
@ -71,7 +77,7 @@ def route(method: RouteType | str | list[str], path: str, name: str | None = Non
"""
methods = [method] if isinstance(method, (str, RouteType)) else method
def decorator(func):
def decorator(func: Callable[..., Awaitable[Any]]):
@wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
@ -97,7 +103,13 @@ def route(method: RouteType | str | list[str], path: str, name: str | None = Non
return decorator
def add_route(method: RouteType | str | list[str], path: str, handler: Awaitable, name: str | None = None, **kwargs):
def add_route(
method: RouteType | str | list[str],
path: str,
handler: Callable[..., Awaitable[Any]],
name: str | None = None,
**kwargs,
):
"""
Decorator to mark a method as an HTTP route handler.
@ -129,7 +141,7 @@ def add_route(method: RouteType | str | list[str], path: str, handler: Awaitable
)
def get_route(route_type: RouteType, name: str) -> dict[str, Route] | None:
def get_route(route_type: RouteType, name: str) -> Route | None:
"""
Get the route information by name.

4
app/main.py
View file

@ -16,6 +16,7 @@ from aiohttp import web
from app.features.conditions.service import Conditions
from app.features.dl_fields.service import DLFields
from app.features.notifications.service import Notifications
from app.library.BackgroundWorker import BackgroundWorker
from app.library.cache import Cache
from app.library.config import Config
@ -23,7 +24,6 @@ from app.library.downloads import DownloadQueue
from app.library.Events import EventBus, Events
from app.library.HttpAPI import HttpAPI
from app.library.HttpSocket import HttpSocket
from app.library.Notifications import Notification
from app.library.Presets import Presets
from app.library.Scheduler import Scheduler
from app.library.Services import Services
@ -118,7 +118,7 @@ class Main:
Presets.get_instance().attach(self._app)
Tasks.get_instance().attach(self._app)
Notification.get_instance().attach(self._app)
Notifications.get_instance().attach(self._app)
Conditions.get_instance().attach(self._app)
DLFields.get_instance().attach(self._app)
TaskDefinitions.get_instance().attach(self._app)

40
app/migrations/20260121134838_add_notifications_table.py Normal file
View file

@ -0,0 +1,40 @@
"""
This module contains a db migration.
Migration Name: add_notifications_table
Migration Version: 20260121134838
"""
from sqlalchemy import text
async def upgrade(c):
sql: list[str] = [
"""
CREATE TABLE IF NOT EXISTS "notifications" (
"id" INTEGER PRIMARY KEY AUTOINCREMENT,
"name" TEXT NOT NULL UNIQUE,
"on" JSON NOT NULL DEFAULT '[]',
"presets" JSON NOT NULL DEFAULT '[]',
"enabled" BOOLEAN NOT NULL DEFAULT 1,
"request_url" TEXT NOT NULL,
"request_method" TEXT NOT NULL DEFAULT 'POST',
"request_type" TEXT NOT NULL DEFAULT 'json',
"request_data_key" TEXT NOT NULL DEFAULT 'data',
"request_headers" JSON NOT NULL DEFAULT '[]',
"created_at" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
""",
'CREATE INDEX IF NOT EXISTS "ix_notifications_name" ON "notifications" ("name");',
'CREATE INDEX IF NOT EXISTS "ix_notifications_enabled" ON "notifications" ("enabled");',
]
for sql_stmt in sql:
await c.execute(text(sql_stmt))
async def downgrade(c):
sql = """
DROP TABLE IF EXISTS "notifications";
"""
await c.execute(text(sql))

109
app/routes/api/notifications.py
View file

@ -1,108 +1,3 @@
import logging
import uuid
"""Migrated API routes for feature submodule."""
from aiohttp import web
from aiohttp.web import Request, Response
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.Notifications import Notification, NotificationEvents
from app.library.router import route
from app.library.Utils import validate_uuid
LOG: logging.Logger = logging.getLogger(__name__)
@route("GET", "api/notifications/", "notifications_list")
async def notifications_list(encoder: Encoder) -> Response:
"""
Get the notification targets.
Args:
encoder (Encoder): The encoder instance.
Returns:
Response: The response object.
"""
return web.json_response(
data={
"notifications": Notification.get_instance().get_targets(),
"allowedTypes": list(NotificationEvents.get_events().values()),
},
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("PUT", "api/notifications/", "notification_add")
async def notification_add(request: Request, encoder: Encoder) -> Response:
"""
Add notification targets.
Args:
request (Request): The request object.
encoder (Encoder): The encoder instance.
Returns:
Response: The response object.
"""
post = await request.json()
if not isinstance(post, list):
return web.json_response(
{"error": "Invalid request body expecting list with dicts."},
status=web.HTTPBadRequest.status_code,
)
targets: list = []
ins: Notification = Notification.get_instance()
for item in post:
if not isinstance(item, dict):
return web.json_response(
{"error": "Invalid request body expecting list with dicts."},
status=web.HTTPBadRequest.status_code,
)
if not item.get("id", None) or validate_uuid(item.get("id"), version=4):
item["id"] = str(uuid.uuid4())
try:
Notification.validate(item)
except ValueError as e:
return web.json_response(
{"error": f"Invalid notification target settings. {e!s}", "data": item},
status=web.HTTPBadRequest.status_code,
)
targets.append(ins.make_target(item))
try:
ins.save(targets=targets)
ins.load()
except Exception as e:
LOG.exception(e)
return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code)
data = {"notifications": targets, "allowedTypes": list(NotificationEvents.get_events().values())}
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("POST", "api/notifications/test/", "notification_test")
async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
"""
Test the notification.
Args:
encoder (Encoder): The encoder instance.
notify (EventBus): The event bus instance.
Returns:
Response: The response object.
"""
notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode)
import app.features.notifications.router # noqa: F401

921
app/tests/test_notifications.py
View file

@ -1,921 +0,0 @@
import json
import tempfile
import uuid
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch
import httpx
import pytest
from aiohttp import web
from app.library.BackgroundWorker import BackgroundWorker
from app.library.encoder import Encoder
from app.library.Events import Event
from app.library.ItemDTO import ItemDTO
from app.library.Notifications import (
Notification,
NotificationEvents,
Target,
TargetRequest,
TargetRequestHeader,
)
class TestTargetRequestHeader:
"""Test the TargetRequestHeader dataclass."""
def test_target_request_header_creation(self):
"""Test creating a TargetRequestHeader."""
header = TargetRequestHeader(key="Authorization", value="Bearer token123")
assert header.key == "Authorization"
assert header.value == "Bearer token123"
def test_target_request_header_serialize(self):
"""Test serializing a TargetRequestHeader."""
header = TargetRequestHeader(key="Content-Type", value="application/json")
serialized = header.serialize()
expected = {"key": "Content-Type", "value": "application/json"}
assert serialized == expected
def test_target_request_header_json(self):
"""Test JSON serialization of TargetRequestHeader."""
header = TargetRequestHeader(key="X-API-Key", value="secret123")
json_str = header.json()
# Should be valid JSON
parsed = json.loads(json_str)
assert parsed["key"] == "X-API-Key"
assert parsed["value"] == "secret123"
def test_target_request_header_get(self):
"""Test get method of TargetRequestHeader."""
header = TargetRequestHeader(key="Authorization", value="Bearer token")
assert header.get("key") == "Authorization"
assert header.get("value") == "Bearer token"
assert header.get("nonexistent", "default") == "default"
class TestTargetRequest:
"""Test the TargetRequest dataclass."""
def test_target_request_creation_defaults(self):
"""Test creating a TargetRequest with defaults."""
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
assert request.type == "json"
assert request.method == "POST"
assert request.url == "https://example.com/webhook"
assert request.headers == []
assert request.data_key == "data"
def test_target_request_creation_with_headers(self):
"""Test creating a TargetRequest with headers."""
headers = [
TargetRequestHeader(key="Authorization", value="Bearer token"),
TargetRequestHeader(key="Content-Type", value="application/json"),
]
request = TargetRequest(
type="json", method="POST", url="https://example.com/webhook", headers=headers, data_key="payload"
)
assert len(request.headers) == 2
assert request.headers[0].key == "Authorization"
assert request.data_key == "payload"
def test_target_request_serialize(self):
"""Test serializing a TargetRequest."""
headers = [TargetRequestHeader(key="X-Token", value="abc123")]
request = TargetRequest(
type="json", method="PUT", url="https://api.example.com/notify", headers=headers, data_key="content"
)
serialized = request.serialize()
expected = {
"type": "json",
"method": "PUT",
"url": "https://api.example.com/notify",
"data_key": "content",
"headers": [{"key": "X-Token", "value": "abc123"}],
}
assert serialized == expected
def test_target_request_json(self):
"""Test JSON serialization of TargetRequest."""
request = TargetRequest(type="form", method="POST", url="https://webhook.site/test")
json_str = request.json()
parsed = json.loads(json_str)
assert parsed["type"] == "form"
assert parsed["method"] == "POST"
assert parsed["url"] == "https://webhook.site/test"
def test_target_request_get(self):
"""Test get method of TargetRequest."""
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
assert request.get("type") == "json"
assert request.get("method") == "POST"
assert request.get("nonexistent", "default") == "default"
class TestTarget:
"""Test the Target dataclass."""
def test_target_creation_minimal(self):
"""Test creating a Target with minimal required fields."""
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test Webhook", request=request)
assert target.name == "Test Webhook"
assert target.on == []
assert target.presets == []
assert target.enabled is True
assert target.request == request
def test_target_creation_full(self):
"""Test creating a Target with all fields."""
target_id = str(uuid.uuid4())
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(
id=target_id,
name="Full Test Webhook",
on=["item_completed", "item_failed"],
presets=["default", "audio_only"],
enabled=False,
request=request,
)
assert target.id == target_id
assert target.name == "Full Test Webhook"
assert target.on == ["item_completed", "item_failed"]
assert target.presets == ["default", "audio_only"]
assert target.enabled is False
def test_target_serialize(self):
"""Test serializing a Target."""
target_id = str(uuid.uuid4())
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=target_id, name="Test Target", on=["item_completed"], presets=["default"], request=request)
serialized = target.serialize()
assert serialized["id"] == target_id
assert serialized["name"] == "Test Target"
assert serialized["on"] == ["item_completed"]
assert serialized["presets"] == ["default"]
assert isinstance(serialized["request"], dict)
def test_target_json(self):
"""Test JSON serialization of Target."""
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="JSON Test", request=request)
json_str = target.json()
parsed = json.loads(json_str)
assert parsed["name"] == "JSON Test"
def test_target_get(self):
"""Test get method of Target."""
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Get Test", request=request)
assert target.get("name") == "Get Test"
assert target.get("nonexistent", "default") == "default"
class TestNotificationEvents:
"""Test the NotificationEvents class."""
def test_notification_events_constants(self):
"""Test that NotificationEvents has expected constants."""
assert hasattr(NotificationEvents, "TEST")
assert hasattr(NotificationEvents, "ITEM_ADDED")
assert hasattr(NotificationEvents, "ITEM_COMPLETED")
assert hasattr(NotificationEvents, "ITEM_CANCELLED")
assert hasattr(NotificationEvents, "ITEM_DELETED")
assert hasattr(NotificationEvents, "LOG_INFO")
assert hasattr(NotificationEvents, "LOG_ERROR")
def test_get_events(self):
"""Test get_events static method."""
events = NotificationEvents.get_events()
assert isinstance(events, dict)
assert "TEST" in events
assert "ITEM_ADDED" in events
assert "ITEM_COMPLETED" in events
def test_events_function(self):
"""Test events function."""
events_list = NotificationEvents.events()
assert isinstance(events_list, list)
assert len(events_list) > 0
def test_is_valid(self):
"""Test is_valid static method."""
assert NotificationEvents.is_valid("test"), "Valid events"
assert NotificationEvents.is_valid("item_added")
assert NotificationEvents.is_valid("item_completed")
assert not NotificationEvents.is_valid("invalid_event"), "Invalid events"
assert not NotificationEvents.is_valid("")
assert not NotificationEvents.is_valid(None)
class TestNotification:
"""Test the Notification singleton class."""
def setup_method(self):
"""Set up test environment before each test."""
# Reset singleton instance to ensure clean state
Notification._reset_singleton()
def teardown_method(self):
"""Clean up after each test."""
# Reset singleton instance to prevent test pollution
Notification._reset_singleton()
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
def test_notification_singleton(self, mock_background_worker, mock_config):
"""Test that Notification follows singleton pattern."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_background_worker.get_instance.return_value = Mock()
instance1 = Notification.get_instance()
instance2 = Notification.get_instance()
assert instance1 is instance2
assert isinstance(instance1, Notification)
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
def test_notification_init_default_params(self, mock_background_worker, mock_config):
"""Test Notification initialization with default parameters."""
mock_config_instance = Mock(debug=False, config_path="/tmp/test", app_version="1.0.0")
mock_config.get_instance.return_value = mock_config_instance
mock_background_worker.get_instance.return_value = Mock()
with patch("pathlib.Path.exists", return_value=False):
notification = Notification.get_instance()
assert notification._debug is False
assert notification._version == "1.0.0"
assert isinstance(notification._targets, list)
assert len(notification._targets) == 0
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
def test_notification_init_custom_params(self, mock_background_worker, mock_config):
"""Test Notification initialization with custom parameters."""
_ = mock_background_worker, mock_config # Suppress unused variable warnings
mock_config_instance = Mock(debug=True, config_path="/custom/path", app_version="2.0.0")
mock_client = Mock(spec=httpx.AsyncClient)
mock_encoder = Mock(spec=Encoder)
mock_worker = Mock(spec=BackgroundWorker)
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_path = temp_file.name
notification = Notification.get_instance(
file=temp_path,
client=mock_client,
encoder=mock_encoder,
config=mock_config_instance,
background_worker=mock_worker,
)
assert notification._debug is True
assert notification._version == "2.0.0"
assert notification._client == mock_client
assert notification._encoder == mock_encoder
assert notification._offload == mock_worker
def test_get_targets_empty(self):
"""Test get_targets when no targets are loaded."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
targets = notification.get_targets()
assert isinstance(targets, list)
assert len(targets) == 0
def test_clear_targets(self):
"""Test clearing notification targets."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
# Add a dummy target
notification._targets = ["dummy_target"]
assert len(notification._targets) == 1
# Clear targets
result = notification.clear()
assert result == notification # Should return self
assert len(notification._targets) == 0
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
def test_save_targets(self, mock_worker, mock_config):
"""Test saving targets to file."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
temp_path = temp_file.name
# Create a test target
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
notification = Notification.get_instance(file=temp_path)
result = notification.save([target])
assert result == notification # Should return self
# Verify file was written
with open(temp_path) as f:
saved_data = json.load(f)
assert isinstance(saved_data, list)
assert len(saved_data) == 1
assert saved_data[0]["name"] == "Test Target"
# Clean up
Path(temp_path).unlink()
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
def test_load_targets_empty_file(self, mock_worker, mock_config):
"""Test loading targets from empty file."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
temp_file.write("") # Empty file
temp_path = temp_file.name
notification = Notification.get_instance(file=temp_path)
result = notification.load()
assert result == notification # Should return self
assert len(notification._targets) == 0
# Clean up
Path(temp_path).unlink()
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
@patch("app.library.Notifications.Presets")
def test_load_targets_schema_update(self, mock_presets, mock_worker, mock_config):
"""Test that load automatically updates file schema when enabled field is missing."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
mock_preset = Mock()
mock_preset.name = "default"
mock_presets.get_instance.return_value.get_all.return_value = [mock_preset]
# Create target data without enabled field (old schema)
target_data = [
{
"id": str(uuid.uuid4()),
"name": "Old Schema Target",
"on": ["item_completed"],
"presets": ["default"],
"request": {"type": "json", "method": "POST", "url": "https://example.com/webhook"},
}
]
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
json.dump(target_data, temp_file)
temp_path = temp_file.name
notification = Notification.get_instance(file=temp_path)
# Mock the save method to verify it's called
with patch.object(notification, "save") as mock_save:
notification.load()
# Verify save was called due to schema update
mock_save.assert_called_once()
assert len(notification._targets) == 1, "Verify the target has enabled=True by default"
assert notification._targets[0].enabled is True
# Clean up
Path(temp_path).unlink()
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
@patch("app.library.Notifications.Presets")
def test_load_targets_valid_file(self, mock_presets, mock_worker, mock_config):
"""Test loading targets from valid file."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
# Mock preset with name attribute
mock_preset = Mock()
mock_preset.name = "default"
mock_presets.get_instance.return_value.get_all.return_value = [mock_preset]
target_data = [
{
"id": str(uuid.uuid4()),
"name": "Test Webhook",
"on": ["item_completed"],
"presets": ["default"],
"request": {
"type": "json",
"method": "POST",
"url": "https://example.com/webhook",
"data_key": "data",
"headers": [],
},
}
]
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
json.dump(target_data, temp_file)
temp_path = temp_file.name
notification = Notification.get_instance(file=temp_path)
result = notification.load()
assert result == notification # Should return self
assert len(notification._targets) == 1
assert notification._targets[0].name == "Test Webhook"
# Clean up
Path(temp_path).unlink()
def test_make_target(self):
"""Test make_target method."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
target_dict = {
"id": str(uuid.uuid4()),
"name": "Test Target",
"on": ["item_completed"],
"presets": ["default"],
"request": {
"type": "json",
"method": "POST",
"url": "https://example.com/webhook",
"data_key": "payload",
"headers": [{"key": "Authorization", "value": "Bearer token"}],
},
}
target = notification.make_target(target_dict)
assert isinstance(target, Target)
assert target.name == "Test Target"
assert target.on == ["item_completed"]
assert target.presets == ["default"]
assert target.request.url == "https://example.com/webhook"
assert target.request.data_key == "payload"
assert len(target.request.headers) == 1
assert target.request.headers[0].key == "Authorization"
def test_validate_target_valid(self):
"""Test validate method with valid target."""
target_dict = {
"id": str(uuid.uuid4()),
"name": "Valid Target",
"request": {"url": "https://example.com/webhook"},
}
with (
patch("app.library.Notifications.NotificationEvents.get_events") as mock_events,
patch("app.library.Notifications.Presets") as mock_presets,
):
mock_events.return_value.values.return_value = ["item_completed"]
mock_presets.get_instance.return_value.get_all.return_value = []
result = Notification.validate(target_dict)
assert result is True
def test_validate_target_missing_id(self):
"""Test validate method with missing ID."""
target_dict = {"name": "Missing ID Target", "request": {"url": "https://example.com/webhook"}}
with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."):
Notification.validate(target_dict)
def test_validate_target_invalid_id(self):
"""Test validate method with invalid UUID."""
target_dict = {
"id": "invalid-uuid",
"name": "Invalid ID Target",
"request": {"url": "https://example.com/webhook"},
}
with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."):
Notification.validate(target_dict)
def test_validate_target_missing_name(self):
"""Test validate method with missing name."""
target_dict = {"id": str(uuid.uuid4()), "request": {"url": "https://example.com/webhook"}}
with pytest.raises(ValueError, match=r"Invalid notification target\. No name found\."):
Notification.validate(target_dict)
def test_validate_target_missing_request(self):
"""Test validate method with missing request."""
target_dict = {"id": str(uuid.uuid4()), "name": "Missing Request Target"}
with pytest.raises(ValueError, match=r"Invalid notification target\. No request details found\."):
Notification.validate(target_dict)
def test_validate_target_missing_url(self):
"""Test validate method with missing URL."""
target_dict = {"id": str(uuid.uuid4()), "name": "Missing URL Target", "request": {}}
with pytest.raises(ValueError, match=r"Invalid notification target\. No URL found\."):
Notification.validate(target_dict)
def test_validate_target_invalid_method(self):
"""Test validate method with invalid HTTP method."""
target_dict = {
"id": str(uuid.uuid4()),
"name": "Invalid Method Target",
"request": {
"url": "https://example.com/webhook",
"method": "GET", # Only POST and PUT are allowed
},
}
with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid method found\."):
Notification.validate(target_dict)
def test_validate_target_invalid_type(self):
"""Test validate method with invalid request type."""
target_dict = {
"id": str(uuid.uuid4()),
"name": "Invalid Type Target",
"request": {
"url": "https://example.com/webhook",
"type": "xml", # Only json and form are allowed
},
}
with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid type found\."):
Notification.validate(target_dict)
def test_validate_target_invalid_enabled(self):
"""Test validating a target with invalid enabled field."""
target_dict = {
"id": str(uuid.uuid4()),
"name": "Test Target",
"enabled": "yes", # Should be boolean
"request": {
"url": "https://example.com/webhook",
"method": "POST",
"type": "json",
},
}
with pytest.raises(ValueError, match=r"Enabled must be a boolean\."):
Notification.validate(target_dict)
def test_validate_target_enabled_defaults_to_true(self):
"""Test that enabled defaults to True when not provided."""
target_dict = {
"id": str(uuid.uuid4()),
"name": "Test Target",
"request": {
"url": "https://example.com/webhook",
"method": "POST",
"type": "json",
},
}
result = Notification.validate(target_dict)
assert result is True
assert target_dict.get("enabled") is True
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
@patch("app.library.Notifications.EventBus")
def test_attach(self, mock_eventbus, mock_worker, mock_config):
"""Test attach method."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
mock_eventbus_instance = Mock()
mock_eventbus.get_instance.return_value = mock_eventbus_instance
notification = Notification.get_instance()
app = Mock(spec=web.Application)
with patch.object(notification, "load") as mock_load:
notification.attach(app)
mock_load.assert_called_once()
mock_eventbus_instance.subscribe.assert_called_once()
@pytest.mark.asyncio
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
async def test_send_no_targets(self, mock_worker, mock_config):
"""Test send method with no targets."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
event = Event(event="test", data={"test": "data"})
result = await notification.send(event)
assert result == []
@pytest.mark.asyncio
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
async def test_send_invalid_event_data(self, mock_worker, mock_config):
"""Test send method with invalid event data."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
# Add a target
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
notification._targets = [target]
event = Event(
event="test",
data="invalid_string_data", # Should be ItemDTO or dict
)
result = await notification.send(event)
assert result == []
@pytest.mark.asyncio
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
async def test_send_with_http_target(self, mock_worker, mock_config):
"""Test send method with HTTP target."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
# Mock HTTP client
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = "OK"
mock_client = AsyncMock()
mock_client.request = AsyncMock(return_value=mock_response)
notification = Notification.get_instance(client=mock_client)
# Add HTTP target
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test HTTP Target", request=request)
notification._targets = [target]
# Create test event
item_dto = ItemDTO(
id="test_id", url="https://youtube.com/watch?v=test", title="Test Video", folder="/downloads"
)
event = Event(event="item_completed", data=item_dto)
result = await notification.send(event)
assert len(result) == 1
assert result[0]["status"] == 200
assert result[0]["text"] == "OK"
mock_client.request.assert_called_once()
@pytest.mark.asyncio
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
async def test_send_skips_disabled_targets(self, mock_worker, mock_config):
"""Test that send method skips disabled targets."""
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
# Mock HTTP client
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = "OK"
mock_client = AsyncMock()
mock_client.request = AsyncMock(return_value=mock_response)
notification = Notification.get_instance(client=mock_client)
# Add enabled and disabled targets
enabled_request = TargetRequest(type="json", method="POST", url="https://example.com/enabled")
enabled_target = Target(id=str(uuid.uuid4()), name="Enabled Target", enabled=True, request=enabled_request)
disabled_request = TargetRequest(type="json", method="POST", url="https://example.com/disabled")
disabled_target = Target(id=str(uuid.uuid4()), name="Disabled Target", enabled=False, request=disabled_request)
notification._targets = [enabled_target, disabled_target]
# Create test event
item_dto = ItemDTO(
id="test_id", url="https://youtube.com/watch?v=test", title="Test Video", folder="/downloads"
)
event = Event(event="item_completed", data=item_dto)
result = await notification.send(event)
assert len(result) == 1, "Only enabled target should be called"
assert result[0]["status"] == 200
mock_client.request.assert_called_once()
@pytest.mark.asyncio
@patch("app.library.Notifications.Config")
@patch("app.library.Notifications.BackgroundWorker")
async def test_send_with_apprise_target(self, mock_worker, mock_config):
"""Test send method with Apprise target."""
mock_config.get_instance.return_value = Mock(
debug=False, config_path="/tmp", app_version="1.0.0", apprise_config="/tmp/apprise.conf"
)
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
# Add Apprise target (non-HTTP URL)
request = TargetRequest(type="json", method="POST", url="discord://webhook_id/webhook_token")
target = Target(id=str(uuid.uuid4()), name="Test Discord Target", request=request)
notification._targets = [target]
# Create test event
event = Event(event="item_completed", data={"test": "data"})
with patch("app.library.Notifications.Path") as mock_path, patch("builtins.__import__") as mock_import:
# Mock apprise config file doesn't exist
mock_path.return_value.exists.return_value = False
# Mock apprise import
mock_apprise = Mock()
mock_notify = Mock()
# Mock async_notify as an AsyncMock that returns True
mock_notify.async_notify = AsyncMock(return_value=True)
mock_apprise.Apprise.return_value = mock_notify
mock_import.return_value = mock_apprise
result = await notification.send(event)
assert len(result) == 1, "Should return empty dict from _apprise method"
assert result[0] == {}
def test_check_preset_no_presets(self):
"""Test _check_preset method with target having no preset filters."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(
id=str(uuid.uuid4()),
name="No Preset Filter",
presets=[], # No preset filter
request=request,
)
event = Event(event="item_completed", data={"preset": "default"})
result = notification._check_preset(target, event)
assert result is True # Should pass when no preset filter
def test_check_preset_with_matching_preset(self):
"""Test _check_preset method with matching preset."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(
id=str(uuid.uuid4()), name="Preset Filter", presets=["default", "audio_only"], request=request
)
# Test with ItemDTO
item_dto = ItemDTO(
id="test_id", url="https://youtube.com/test", title="Test Video", folder="/downloads", preset="default"
)
event = Event(event="item_completed", data=item_dto)
result = notification._check_preset(target, event)
assert result is True
def test_check_preset_with_non_matching_preset(self):
"""Test _check_preset method with non-matching preset."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Preset Filter", presets=["audio_only"], request=request)
event = Event(
event="item_completed",
data={"preset": "video_only"}, # Different preset
)
result = notification._check_preset(target, event)
assert result is False
def test_emit_invalid_event(self):
"""Test emit method with invalid event."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker_instance = Mock()
mock_worker.get_instance.return_value = mock_worker_instance
notification = Notification.get_instance()
# Add a target
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
notification._targets = [target]
# Create invalid event
event = Event(event="invalid_event", data={"test": "data"})
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False):
result = notification.emit(event, None)
assert result is None, "Should return None and not submit to background worker"
mock_worker_instance.submit.assert_not_called()
def test_emit_valid_event(self):
"""Test emit method with valid event."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker_instance = Mock()
mock_worker.get_instance.return_value = mock_worker_instance
notification = Notification.get_instance()
# Add a target
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
notification._targets = [target]
# Create valid event
event = Event(event="item_completed", data={"test": "data"})
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True):
result = notification.emit(event, None)
assert result is None, "Should return None but submit to background worker"
mock_worker_instance.submit.assert_called_once_with(notification.send, event)
@pytest.mark.asyncio
async def test_noop(self):
"""Test noop method."""
with (
patch("app.library.Notifications.Config") as mock_config,
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
):
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
mock_worker.get_instance.return_value = Mock()
notification = Notification.get_instance()
result = await notification.noop()
assert result is None

4
ui/app/components/ConditionForm.vue
View file

@ -2,7 +2,7 @@
<main class="columns mt-2 is-multiline">
<div class="column is-12">
<form autocomplete="off" id="addForm" @submit.prevent="checkInfo()">
<div class="card is-flex is-full-height is-flex-direction-column">
<div class="card">
<div class="card-header">
<div class="card-header-title is-text-overflow is-block">
<span class="icon-text">
@ -21,7 +21,7 @@
</div>
</div>
<div class="card-content is-flex-grow-1">
<div class="card-content">
<div class="columns is-multiline is-mobile">

58
ui/app/components/DLFieldForm.vue
View file

@ -2,7 +2,7 @@
<main class="columns mt-2 is-multiline">
<div class="column is-12">
<form autocomplete="off" id="dlFieldForm" @submit.prevent="checkInfo">
<div class="card is-flex is-full-height is-flex-direction-column">
<div class="card">
<div class="card-header">
<div class="card-header-title is-text-overflow is-block">
<span class="icon-text">
@ -21,8 +21,8 @@
</div>
</div>
<div class="card-content is-flex-grow-1">
<div class="columns is-multiline is-mobile">
<div class="card-content">
<div class="columns is-multiline">
<div class="column is-12" v-if="showImport || !reference">
<label class="label is-inline" for="import_string">
<span class="icon"><i class="fa-solid fa-file-import" /></span>
@ -46,6 +46,32 @@
<span>You can use this field to populate the data, using shared string.</span>
</span>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
<span class="icon"><i class="fas fa-font" /></span>
<span>Field Name</span>
</label>
<input type="text" v-model="form.name" class="input" :disabled="addInProgress" />
<span class="help is-bold">
The name of the field, it will be shown in the UI.
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
<span class="icon"><i class="fas fa-info-circle" /></span>
<span>Field Description</span>
</label>
<input type="text" v-model="form.description" class="input" :disabled="addInProgress" />
<span class="help is-bold">
A short description of the field, it will be shown in the UI.
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
@ -62,18 +88,7 @@
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
<span class="icon"><i class="fas fa-font" /></span>
<span>Field Name</span>
</label>
<input type="text" v-model="form.name" class="input" :disabled="addInProgress" />
<span class="help is-bold">
The name of the field, it will be shown in the UI.
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
@ -87,18 +102,7 @@
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">
<label class="label">
<span class="icon"><i class="fas fa-info-circle" /></span>
<span>Field Description</span>
</label>
<input type="text" v-model="form.description" class="input" :disabled="addInProgress" />
<span class="help is-bold">
A short description of the field, it will be shown in the UI.
</span>
</div>
</div>
<div class="column is-6 is-12-mobile">
<div class="field">

27
ui/app/components/NotificationForm.vue
View file

@ -92,7 +92,7 @@
</div>
</div>
<div class="column is-6-tablet is-12-mobile" v-if="!isApprise">
<div class="column is-6-tablet is-12-mobile" v-if="!isAppriseTarget">
<div class="field">
<label class="label is-inline" for="method">
Request method
@ -117,7 +117,7 @@
</div>
</div>
<div class="column is-6-tablet is-12-mobile" v-if="!isApprise">
<div class="column is-6-tablet is-12-mobile" v-if="!isAppriseTarget">
<div class="field">
<label class="label is-inline" for="type">
Request Type
@ -207,7 +207,7 @@
</div>
</div>
<div class="column" :class="isApprise ? 'is-12' : 'is-6-tablet is-12-mobile'">
<div class="column" :class="isAppriseTarget ? 'is-12' : 'is-6-tablet is-12-mobile'">
<div class="field">
<label class="label is-inline" for="enabled">
<span class="icon"><i class="fa-solid fa-power-off" /></span>
@ -227,7 +227,7 @@
</div>
</div>
<div class="column is-6-tablet is-12-mobile" v-if="!isApprise">
<div class="column is-6-tablet is-12-mobile" v-if="!isAppriseTarget">
<div class="field">
<label class="label is-inline" for="data_key">
Data field
@ -247,7 +247,7 @@
</div>
</div>
<div class="column is-12" v-if="!isApprise">
<div class="column is-12" v-if="!isAppriseTarget">
<div class="field">
<label class="label is-inline is-unselectable">
Optional Headers - <button type="button" class="has-text-link"
@ -327,22 +327,25 @@
<script setup lang="ts">
import { useStorage } from '@vueuse/core'
import type { notification, notificationImport } from '~/types/notification'
import type { notification } from '~/types/notification'
import { useConfirm } from '~/composables/useConfirm'
import { useNotification } from '~/composables/useNotification'
import type { ImportedItem } from '~/types'
const emitter = defineEmits(['cancel', 'submit'])
const toast = useNotification()
const box = useConfirm()
const config = useConfigStore()
const { isApprise } = useNotifications()
const props = defineProps({
reference: {
type: String,
type: Number,
required: false,
default: null,
},
allowedEvents: {
type: Array as () => string[],
type: Array as () => readonly string[],
required: true,
},
item: {
@ -374,7 +377,7 @@ onMounted(() => {
const checkInfo = async () => {
let required: string[]
if (!isApprise.value) {
if (!isAppriseTarget.value) {
required = ['name', 'request.url', 'request.method', 'request.type', 'request.data_key']
} else {
required = ['name', 'request.url']
@ -398,7 +401,7 @@ const checkInfo = async () => {
}
}
if (!isApprise.value) {
if (!isAppriseTarget.value) {
try {
new URL(form.request.url)
} catch {
@ -427,7 +430,7 @@ const importItem = async () => {
}
try {
const item = decode(val) as notificationImport
const item = decode(val) as notification & ImportedItem
if ('notification' !== item._type) {
toast.error(`Invalid import string. Expected type 'notification', got '${item._type}'.`)
@ -483,6 +486,6 @@ const importItem = async () => {
}
}
const isApprise = computed(() => form.request.url && !form.request.url.startsWith('http'))
const isAppriseTarget = computed(() => form.request.url && isApprise(form.request.url))
const filter_presets = (flag: boolean = true) => config.presets.filter(item => item.default === flag)
</script>

427
ui/app/composables/useNotifications.ts Normal file
View file

@ -0,0 +1,427 @@
import { ref, readonly } from 'vue'
import { useNotification } from '~/composables/useNotification'
import { request, parse_list_response, parse_api_response, parse_api_error } from '~/utils'
import type { notification } from '~/types/notification'
import type { APIResponse, Pagination } from '~/types/responses'
/**
* List of all notifications in memory.
*/
const notifications = ref<Array<notification>>([])
/**
* Pagination state for notifications list.
*/
const pagination = ref<Pagination>({
page: 1,
per_page: 50,
total: 0,
total_pages: 0,
has_next: false,
has_prev: false,
})
/**
* List of allowed notification events.
*/
const events = ref<Array<string>>([])
/**
* Indicates if a request is in progress.
*/
const isLoading = ref<boolean>(false)
/**
* Indicates if an add/update operation is in progress.
*/
const addInProgress = ref<boolean>(false)
/**
* Stores the last error message, if any.
*/
const lastError = ref<string | null>(null)
/**
* If true, methods will throw errors instead of returning null/false (for testing)
*/
const throwInstead = ref(false)
/**
* Notification composable for showing success/error messages.
*/
const notify = useNotification()
/**
* Sorts notifications by name (A-Z).
* @param items Array of notifications
* @returns Sorted array of notifications
*/
const sortNotifications = (items: Array<notification>): Array<notification> => {
return [...items].sort((a, b) => a.name.localeCompare(b.name))
}
/**
* Safely reads JSON from a Response, returns null on error.
* @param response Fetch Response object
* @returns Parsed JSON or null
*/
const readJson = async (response: Response): Promise<unknown> => {
try {
const clone = response.clone()
return await clone.json()
}
catch {
return null
}
}
/**
* Throws an error if the response is not OK, using API error message if available.
* @param response Fetch Response object
* @throws Error with message from API or status code
*/
const ensureSuccess = async (response: Response): Promise<void> => {
if (response.ok) {
return
}
const payload = await readJson(response)
const message = await parse_api_error(payload)
throw new Error(message)
}
/**
* Handles errors by updating lastError and showing a notification.
* @param error Error object or unknown
*/
const handleError = (error: unknown): void => {
const message = error instanceof Error ? error.message : 'Unexpected error occurred.'
lastError.value = message
notify.error(message)
}
/**
* Updates or adds a notification in the list, keeping sort order.
* Also increments pagination.total if it's a new notification.
* @param item Notification to update/add
*/
const updateNotifications = (item: notification): void => {
const isNew = !notifications.value.some(existing => existing.id === item.id)
notifications.value = sortNotifications([
...notifications.value.filter(existing => existing.id !== item.id),
item,
])
if (isNew) {
pagination.value.total++
}
}
/**
* Removes a notification from the list by ID.
* Also decrements pagination.total.
* @param id Notification ID
*/
const removeNotification = (id: number) => {
const initialLength = notifications.value.length
notifications.value = notifications.value.filter(item => item.id !== id)
if (notifications.value.length < initialLength) {
pagination.value.total = Math.max(0, pagination.value.total - 1)
}
}
/**
* Loads notification targets from the API with pagination support.
* @param page Page number
* @param perPage Items per page
*/
const loadNotifications = async (page: number = 1, perPage: number | undefined = undefined): Promise<void> => {
await loadNotificationEvents()
isLoading.value = true
try {
let url = `/api/notifications/?page=${page}`
if (perPage !== undefined) {
url += `&per_page=${perPage}`
}
const response = await request(url)
await ensureSuccess(response)
const json = await response.json()
const { items, pagination: paginationData } = await parse_list_response<notification>(json)
notifications.value = sortNotifications(items)
pagination.value = paginationData
lastError.value = null
}
catch (error) {
handleError(error)
if (throwInstead.value) throw error
}
finally {
isLoading.value = false
}
}
/**
* Loads notification events from the API.
*/
const loadNotificationEvents = async (): Promise<void> => {
if (events.value.length > 0) {
return
}
try {
const response = await request('/api/notifications/events/')
await ensureSuccess(response)
const json = await response.json()
events.value = Array.isArray(json?.events) ? json.events : []
lastError.value = null
}
catch (error) {
handleError(error)
if (throwInstead.value) throw error
}
}
/**
* Fetches a single notification by ID from the API.
* @param id Notification ID
* @returns Notification or null on error
*/
const getNotification = async (id: number): Promise<notification | null> => {
try {
const response = await request(`/api/notifications/${id}`)
await ensureSuccess(response)
const json = await response.json()
const item = await parse_api_response<notification>(json)
lastError.value = null
return item
}
catch (error) {
handleError(error)
if (throwInstead.value) throw error
return null
}
}
/**
* Creates a new notification via API.
* @param item Notification to create
* @param callback Optional callback with APIResponse result
* @returns Created notification or null on error
*/
const createNotification = async (
item: Omit<notification, 'id'>,
callback?: (response: APIResponse<notification>) => void,
): Promise<notification | null> => {
addInProgress.value = true
try {
const response = await request('/api/notifications/', {
method: 'POST',
body: JSON.stringify(item),
})
await ensureSuccess(response)
const json = await response.json()
const created = await parse_api_response<notification>(json)
updateNotifications(created)
notify.success('Notification target created.')
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: created })
}
return created
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Updates an existing notification via API (PUT - full update).
* @param id Notification ID
* @param item Updated notification data
* @param callback Optional callback with APIResponse result
* @returns Updated notification or null on error
*/
const updateNotification = async (
id: number,
item: notification,
callback?: (response: APIResponse<notification>) => void,
): Promise<notification | null> => {
addInProgress.value = true
try {
if (item.id) {
item.id = undefined
}
const response = await request(`/api/notifications/${id}`, {
method: 'PUT',
body: JSON.stringify(item),
})
await ensureSuccess(response)
const json = await response.json()
const updated = await parse_api_response<notification>(json)
updateNotifications(updated)
notify.success(`Notification target '${updated.name}' updated.`)
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: updated })
}
return updated
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Partially updates an existing notification via API (PATCH).
* @param id Notification ID
* @param patch Partial notification data to update
* @param callback Optional callback with APIResponse result
* @returns Updated notification or null on error
*/
const patchNotification = async (
id: number,
patch: Partial<notification>,
callback?: (response: APIResponse<notification>) => void,
): Promise<notification | null> => {
addInProgress.value = true
try {
if (patch.id) {
patch.id = undefined
}
const response = await request(`/api/notifications/${id}`, {
method: 'PATCH',
body: JSON.stringify(patch),
})
await ensureSuccess(response)
const json = await response.json()
const updated = await parse_api_response<notification>(json)
updateNotifications(updated)
notify.success(`Notification target '${updated.name}' updated.`)
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: updated })
}
return updated
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Deletes a notification by ID via API.
* @param id Notification ID
* @param callback Optional callback with APIResponse result
* @returns true if deleted, false on error
*/
const deleteNotification = async (
id: number,
callback?: (response: APIResponse<boolean>) => void,
): Promise<boolean> => {
try {
const response = await request(`/api/notifications/${id}`, { method: 'DELETE' })
await ensureSuccess(response)
removeNotification(id)
notify.success('Notification target deleted.')
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: true })
}
return true
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: false })
}
if (throwInstead.value) throw error
return false
}
}
/**
* Determines if a notification URL is for Apprise (non-HTTP).
* @param url Notification URL
* @returns true if Apprise URL, false otherwise
*/
const isApprise = (url: string) => !url.startsWith('http')
/**
* Clears the last error message.
*/
const clearError = () => lastError.value = null
/**
* useNotifications composable
*
* Returns reactive state and CRUD methods for notifications.
* @returns Object with state and API methods
*/
export const useNotifications = () => ({
notifications: readonly(notifications),
pagination: readonly(pagination),
events: readonly(events),
isLoading: readonly(isLoading),
addInProgress: readonly(addInProgress),
lastError: readonly(lastError),
loadNotifications,
loadNotificationEvents,
getNotification,
createNotification,
updateNotification,
patchNotification,
deleteNotification,
clearError,
throwInstead,
isApprise
})

2
ui/app/pages/conditions.vue
View file

@ -398,7 +398,7 @@ const updateItem = async ({ reference, item: updatedItem }: {
}
const editItem = (_item: Condition): void => {
item.value = { ..._item }
item.value = JSON.parse(JSON.stringify(_item)) as Condition
itemRef.value = _item.id
toggleForm.value = true
}

188
ui/app/pages/notifications.vue
View file

@ -56,7 +56,7 @@
</p>
<p class="control" v-if="notifications.length > 0">
<button class="button is-info" @click="reloadContent()" :class="{ 'is-loading': isLoading }"
<button class="button is-info" @click="loadContent(page)" :class="{ 'is-loading': isLoading }"
:disabled="isLoading || notifications.length < 1">
<span class="icon"><i class="fas fa-refresh" /></span>
<span v-if="!isMobile">Reload</span>
@ -71,8 +71,13 @@
</div>
</div>
<div class="column is-12" v-if="!toggleForm && paging?.total_pages > 1">
<Pager :page="paging.page" :last_page="paging.total_pages" :isLoading="isLoading"
@navigate="async (newPage) => { page = newPage; await loadContent(newPage); }" />
</div>
<div class="column is-12" v-if="toggleForm">
<NotificationForm :addInProgress="addInProgress" :reference="targetRef as string" :item="target"
<NotificationForm :addInProgress="addInProgress" :reference="targetRef" :item="target"
@cancel="resetForm(true);" @submit="updateItem" :allowedEvents="allowedEvents" />
</div>
</div>
@ -171,12 +176,6 @@
<span class="icon"><i class="fa-solid fa-file-export" /></span>
</a>
</div>
<div class="control">
<button @click="item.raw = !item.raw">
<span class="icon"><i class="fa-solid"
:class="{ 'fa-arrow-down': !item?.raw, 'fa-arrow-up': item?.raw }" /></span>
</button>
</div>
</div>
</div>
</header>
@ -196,11 +195,6 @@
</p>
</div>
</div>
<div class="card-content" v-if="item?.raw">
<div class="content">
<pre><code>{{ JSON.stringify(cleanObject(item, ['raw']), null, 2) }}</code></pre>
</div>
</div>
<div class="card-footer mt-auto">
<div class="card-footer-item">
<button class="button is-warning is-fullwidth" @click="editItem(item);">
@ -272,19 +266,28 @@
<script setup lang="ts">
import { useStorage } from '@vueuse/core'
import type { notification, notificationImport } from '~/types/notification'
import type { notification } from '~/types/notification'
import { useConfirm } from '~/composables/useConfirm'
type notificationItemWithUI = notification & { raw?: boolean }
import { useNotifications } from '~/composables/useNotifications'
import type { ImportedItem } from '~/types'
const toast = useNotification()
const socket = useSocketStore()
const box = useConfirm()
const display_style = useStorage<string>("tasks_display_style", "cards")
const display_style = useStorage<string>('notification_display_style', 'cards')
const isMobile = useMediaQuery({ maxWidth: 1024 })
const allowedEvents = ref<string[]>([])
const notifications = ref<notificationItemWithUI[]>([])
const notificationsStore = useNotifications()
const notifications = notificationsStore.notifications
const paging = notificationsStore.pagination
const allowedEvents = notificationsStore.events
const isLoading = notificationsStore.isLoading
const addInProgress = notificationsStore.addInProgress
const lastError = notificationsStore.lastError
const page = ref(1)
const targetRef = ref<number | undefined>(undefined)
const toggleForm = ref(false)
const sendingTest = ref(false)
const defaultState = (): notification => ({
name: '',
@ -295,21 +298,15 @@ const defaultState = (): notification => ({
})
const target = ref<notification>(defaultState())
const targetRef = ref<string | null>('')
const toggleForm = ref(false)
const isLoading = ref(false)
const initialLoad = ref(true)
const addInProgress = ref(false)
const sendingTest = ref(false)
const query = ref<string>('')
const toggleFilter = ref(false)
const filteredTargets = computed<notificationItemWithUI[]>(() => {
const q = query.value?.toLowerCase();
if (!q) return notifications.value;
return notifications.value.filter((item: notificationItemWithUI) => deepIncludes(item, q, new WeakSet()));
});
const filteredTargets = computed<notification[]>(() => {
const q = query.value?.toLowerCase()
const items = notifications.value.map(item => ({ ...item })) as notification[]
if (!q) return items
return items.filter((item: notification) => deepIncludes(item, q, new WeakSet()))
})
watch(toggleFilter, (val) => {
if (!val) {
@ -317,134 +314,55 @@ watch(toggleFilter, (val) => {
}
})
watch(() => socket.isConnected, async () => {
if (socket.isConnected && initialLoad.value) {
await reloadContent(true)
initialLoad.value = false
}
})
const reloadContent = async (fromMounted = false) => {
try {
isLoading.value = true
const response = await request('/api/notifications')
if (fromMounted && !response.ok) {
return
}
notifications.value = []
const data = await response.json()
if (data.length < 1) {
return
}
allowedEvents.value = data.allowedTypes
notifications.value = data.notifications
} catch (e) {
if (fromMounted) {
return
}
console.error(e)
toast.error('Failed to fetch notifications.')
} finally {
isLoading.value = false
}
const loadContent = async (pageNumber = page.value) => {
await notificationsStore.loadNotifications(pageNumber)
}
const resetForm = (closeForm = false) => {
target.value = defaultState()
targetRef.value = null
targetRef.value = undefined
if (closeForm) {
toggleForm.value = false
}
}
const updateData = async (items: notification[]) => {
const response = await request('/api/notifications', {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(items),
})
const data = await response.json()
if (200 !== response.status) {
toast.error(`Failed to update notifications. ${data.error}`)
return false
}
notifications.value = data.notifications
return true
}
const deleteItem = async (item: notification) => {
if (true !== (await box.confirm(`Delete '${item.name}'?`))) {
return
}
const index = notifications.value.findIndex(i => i?.id === item.id)
if (0 <= index) {
notifications.value.splice(index, 1)
} else {
if (!item.id) {
toast.error('Notification target not found.')
await reloadContent()
return
}
const status = await updateData(notifications.value)
if (!status) return
toast.success('Notification target deleted.')
await notificationsStore.deleteNotification(item.id)
}
const toggleEnabled = async (item: notification) => {
const index = notifications.value.findIndex(i => i?.id === item.id)
if (-1 === index) {
if (!item.id) {
toast.error('Notification target not found.')
return
}
const target = notifications.value[index]
if (!target) {
toast.error('Notification target not found.')
return
}
target.enabled = !target.enabled
const status = await updateData(notifications.value)
if (status) {
toast[target.enabled ? 'success' : 'warning'](`Notification is ${target.enabled ? 'enabled' : 'disabled'}.`)
}
await notificationsStore.patchNotification(item.id, { enabled: !item.enabled })
}
const updateItem = async ({ reference, item }: { reference: string | null, item: notification }) => {
const updateItem = async ({ reference, item }: { reference: number | undefined, item: notification }) => {
if (reference) {
const index = notifications.value.findIndex(i => i?.id === reference)
if (0 <= index) {
notifications.value[index] = item
}
await notificationsStore.updateNotification(reference, item)
} else {
notifications.value.push(item)
await notificationsStore.createNotification(item)
}
try {
const status = await updateData(notifications.value)
if (!status) return
toast.success(`Notification target ${reference ? 'updated' : 'added'}.`)
if (!lastError.value) {
resetForm(true)
} finally {
addInProgress.value = false
}
}
const editItem = (item: notification) => {
target.value = item
targetRef.value = item.id ?? null
target.value = JSON.parse(JSON.stringify(item)) as notification
targetRef.value = item.id ?? undefined
toggleForm.value = true
}
@ -460,31 +378,33 @@ const sendTest = async () => {
sendingTest.value = true
const response = await request('/api/notifications/test', { method: 'POST' })
if (200 !== response.status) {
if (!response.ok) {
const data = await response.json()
toast.error(`Failed to send test notification. ${data.error}`)
const message = await parse_api_error(data)
toast.error(`Failed to send test notification. ${message}`)
return
}
toast.success('Test notification sent.')
} catch (e: any) {
console.error(e)
toast.error(`Failed to send test notification. ${e.message}`)
} catch (error: any) {
console.error(error)
const message = error?.message || 'Unknown error'
toast.error(`Failed to send test notification. ${message}`)
} finally {
sendingTest.value = false
}
}
onMounted(async () => socket.isConnected ? await reloadContent(true) : '')
onMounted(async () => await notificationsStore.loadNotifications(page.value))
const exportItem = async (item: notification) => {
const data: notificationImport = {
const data: notification & ImportedItem = {
...JSON.parse(JSON.stringify(item)),
_type: 'notification',
_version: '1.0',
}
const keys = ['id', 'raw'] as const
const keys = ['id', 'raw']
keys.forEach(k => {
if (Object.prototype.hasOwnProperty.call(data, k)) {
const { [k]: _, ...rest } = data as any
@ -493,9 +413,7 @@ const exportItem = async (item: notification) => {
})
if (data.request?.headers?.length) {
data.request.headers = data.request.headers.filter(
h => 'authorization' !== h.key.toLowerCase(),
)
data.request.headers = data.request.headers.filter(h => 'authorization' !== h.key.toLowerCase())
}
copyText(encode(data))

1
ui/app/stores/NotificationStore.ts
View file

@ -67,7 +67,6 @@ export const useNotificationStore = defineStore('notifications', () => {
const markAllRead = () => notifications.value.forEach(n => n.seen = true)
const markRead = (id: string) => {
console.log(`Marking notification ${id} as read`)
const n = notifications.value.find(n => n.id === id)
if (!n) {
return

19
ui/app/types/notification.d.ts vendored
View file

@ -12,7 +12,7 @@ type notificationRequest = {
};
type notification = {
id?: string;
id?: number;
name: string;
request: notificationRequest;
on: Array<string>;
@ -20,9 +20,18 @@ type notification = {
enabled: boolean;
};
type notificationImport = notification & {
_type: 'notification';
_version: string;
type notificationPagination = {
page: number;
per_page: number;
total: number;
total_pages: number;
has_next: boolean;
has_prev: boolean;
};
export type { notificationRequestHeaderItem, notification, notificationRequest, notificationImport };
export type {
notificationRequestHeaderItem,
notification,
notificationRequest,
notificationPagination,
};