blob: c37370e91d2f75baed674e6938b6f40c32fcc1e3 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
"""
Adds new admin user.
"""
import logging
from pydantic import BaseModel, Field
from spectree import Response
from starlette.authentication import requires
from starlette.requests import Request
from starlette.responses import JSONResponse
from backend.route import Route
from backend.validation import ErrorMessage, OkayResponse, api
logger = logging.getLogger(__name__)
class AdminModel(BaseModel):
id: str = Field(alias="_id")
class AdminRoute(Route):
"""Adds new admin user."""
name = "admin"
path = "/admin"
@requires(["authenticated", "admin"])
@api.validate(
json=AdminModel,
resp=Response(HTTP_200=OkayResponse, HTTP_400=ErrorMessage),
tags=["admin"]
)
async def post(self, request: Request) -> JSONResponse:
"""Grant a user administrator privileges."""
data = await request.json()
admin = AdminModel(**data)
logger.info(f"Trying to add a new admin with ID: {admin.id}")
if await request.state.db.admins.find_one(
{"_id": admin.id}
):
return JSONResponse({"error": "already_exists"}, status_code=400)
await request.state.db.admins.insert_one(admin.dict(by_alias=True))
return JSONResponse({"status": "ok"})
|
