43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
import asyncpg
|
|
from dishka.integrations.fastapi import DishkaRoute, FromDishka
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from utils.read import bots
|
|
from utils.read.models import BotView
|
|
from watcher import tokens
|
|
|
|
router = APIRouter(prefix="/api/bots", tags=["bots"], route_class=DishkaRoute)
|
|
|
|
INVALID_TOKEN = "token must look like 123456:ABC-DEF..." # noqa: S105
|
|
|
|
|
|
class BotAdd(BaseModel):
|
|
token: str
|
|
|
|
|
|
@router.get("")
|
|
async def list_bots(pool: FromDishka[asyncpg.Pool]) -> list[BotView]:
|
|
return await bots.list_bots(pool)
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def add_bot(pool: FromDishka[asyncpg.Pool], body: BotAdd) -> BotView:
|
|
bot_id = tokens.bot_id_of(body.token)
|
|
if bot_id is None:
|
|
raise HTTPException(status_code=422, detail=INVALID_TOKEN)
|
|
tokens.save(bot_id, body.token)
|
|
await bots.register_bot(pool, bot_id)
|
|
await bots.notify_bots_changed(pool)
|
|
bot = await bots.get_bot(pool, bot_id)
|
|
if bot is None:
|
|
raise HTTPException(status_code=500, detail="bot was not registered")
|
|
return bot
|
|
|
|
|
|
@router.delete("/{bot_id}", status_code=204)
|
|
async def remove_bot(pool: FromDishka[asyncpg.Pool], bot_id: int) -> None:
|
|
if not await bots.remove_bot(pool, bot_id):
|
|
raise HTTPException(status_code=404, detail="unknown bot")
|
|
await bots.notify_bots_changed(pool)
|