146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
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_BULK_DELETED: str = Events.ITEM_BULK_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
|