"""API Routes""" import html import traceback from fastapi import APIRouter, Request, Depends from fastapi import HTTPException, status from pydantic import BaseModel, Field from mab import MatrixBot from database import Database from datatypes import * router = APIRouter(prefix="/api") # # MODELS # class ModelGetNotify(BaseModel): channel: str = Field( ..., min_length=11, max_length=11, description="Notification channel (use `!info` bot command)" ) text: str = Field( ..., min_length=1, max_length=1024, description="Text to use as notification body" ) service: str | None = Field( None, min_length=1, max_length=1024, description="Service name to use (ignored if token has name)" ) # # ENDPOINTS # @router.get("/notify") async def _get_notify(request: Request, params: ModelGetNotify = Depends()): # improve readability database: Database = request.app.state.database matrix: MatrixBot = request.app.state.matrix token: ObjectToken = request.state.token # check if room exists room = await database.room_get(params.channel) if room is None: raise HTTPException(status.HTTP_200_OK, "No such channel") # prepare service name service = token.name or params.service or "Unnamed" # prepare text of the notification text = f"Уведомление от службы {html.escape(service)}" text += "
" * 2 text += html.escape(params.text) # try to send the notification try: notification_id = await matrix.send_text_to_room( room.matrix_id, text, is_html=True ) except: traceback.print_exc() raise HTTPException(status.HTTP_200_OK, "Failed to send matrix message") # success return {"notification_id": notification_id}