97 lines
3.8 KiB
Python
97 lines
3.8 KiB
Python
"""External notification channels endpoints (#168).
|
|
|
|
Channel CRUD is admin-only (secrets involved); sending a test notification
|
|
requires authentication. Responses mask secrets (``***`` + ``has_secret``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Body, Depends, HTTPException
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from backend import notify as _notify
|
|
from backend.auth.middleware import require_admin, require_auth
|
|
from backend.schemas import StatusResponse
|
|
|
|
router = APIRouter(prefix="/api/notify", tags=["notify"])
|
|
|
|
|
|
class NotifyChannel(BaseModel):
|
|
"""Public view of a notification channel (secrets masked)."""
|
|
|
|
model_config = ConfigDict(extra="allow")
|
|
id: str = Field(description="Channel id")
|
|
name: str = Field(description="Display name")
|
|
type: str = Field(description="discord | telegram | smtp | webhook")
|
|
enabled: bool = Field(description="Whether the channel receives broadcasts")
|
|
config: dict[str, Any] = Field(description="Channel config (secrets masked)")
|
|
has_secret: bool = Field(description="True when a secret is configured")
|
|
|
|
|
|
class NotifySendResult(BaseModel):
|
|
"""Outcome of a test send / broadcast."""
|
|
|
|
model_config = ConfigDict(extra="allow")
|
|
ok: bool = Field(description="True when every delivery succeeded")
|
|
deliveries: list[dict[str, Any]] = Field(default_factory=list)
|
|
|
|
|
|
@router.get("/channels", response_model=list[NotifyChannel])
|
|
async def api_notify_list(current_user=Depends(require_admin)):
|
|
"""List notification channels (admin)."""
|
|
return _notify.list_channels()
|
|
|
|
|
|
@router.post("/channels", response_model=NotifyChannel)
|
|
async def api_notify_create(body: dict = Body(...), current_user=Depends(require_admin)):
|
|
"""Create a channel (``{name, type, config}``). Secrets go to the secret store."""
|
|
try:
|
|
return _notify.create_channel(
|
|
str(body.get("name") or ""),
|
|
str(body.get("type") or ""),
|
|
dict(body.get("config") or {}),
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
|
|
|
|
@router.patch("/channels/{channel_id}", response_model=NotifyChannel)
|
|
async def api_notify_update(channel_id: str, body: dict = Body(...), current_user=Depends(require_admin)):
|
|
"""Update a channel (name / enabled / config)."""
|
|
try:
|
|
result = _notify.update_channel(channel_id, body)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
if result is None:
|
|
raise HTTPException(404, "Channel not found")
|
|
return result
|
|
|
|
|
|
@router.delete("/channels/{channel_id}", response_model=StatusResponse)
|
|
async def api_notify_delete(channel_id: str, current_user=Depends(require_admin)):
|
|
"""Delete a channel and its secret."""
|
|
if not _notify.delete_channel(channel_id):
|
|
raise HTTPException(404, "Channel not found")
|
|
return {"status": "deleted"}
|
|
|
|
|
|
@router.post("/test", response_model=NotifySendResult)
|
|
async def api_notify_test(body: dict = Body(...), current_user=Depends(require_auth)):
|
|
"""Send a test notification (broadcast or single ``channel_id``)."""
|
|
title = str(body.get("title") or "Test ObsiGate")
|
|
message = str(body.get("message") or "Notification de test.")
|
|
channel_id = str(body.get("channel_id") or "")
|
|
if channel_id:
|
|
channel = next((c for c in _notify._read_channels() if c.get("id") == channel_id), None)
|
|
if channel is None:
|
|
raise HTTPException(404, "Channel not found")
|
|
try:
|
|
_notify.send_via_channel(channel, title, message, "manual")
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Envoi échoué : {e}") from e
|
|
return {"ok": True, "deliveries": [{"channel_id": channel_id, "ok": True}]}
|
|
deliveries = _notify.broadcast("manual", title, message)
|
|
return {"ok": all(d.get("ok") for d in deliveries), "deliveries": deliveries}
|