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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
"""Returns all form responses by form ID."""
from pydantic import BaseModel
from spectree import Response
from starlette.authentication import requires
from starlette.requests import Request
from starlette.responses import JSONResponse
from backend import discord
from backend.models import FormResponse, ResponseList
from backend.route import Route
from backend.validation import ErrorMessage, OkayResponse, api
class ResponseIdList(BaseModel):
ids: list[str]
class Responses(Route):
"""Returns all form responses by form ID."""
name = "form_responses"
path = "/{form_id:str}/responses"
@requires(["authenticated"])
@api.validate(
resp=Response(HTTP_200=ResponseList),
tags=["forms", "responses"],
)
async def get(self, request: Request) -> JSONResponse:
"""Returns all form responses by form ID."""
form_id = request.path_params["form_id"]
await discord.verify_response_access(form_id, request)
cursor = request.state.db.responses.find(
{"form_id": form_id},
)
responses = [FormResponse(**response) for response in await cursor.to_list(None)]
return JSONResponse([response.dict() for response in responses])
@requires(["authenticated", "admin"])
@api.validate(
json=ResponseIdList,
resp=Response(
HTTP_200=OkayResponse,
HTTP_404=ErrorMessage,
HTTP_400=ErrorMessage,
),
tags=["forms", "responses"],
)
async def delete(self, request: Request) -> JSONResponse:
"""Bulk deletes form responses by IDs."""
if not await request.state.db.forms.find_one(
{"_id": request.path_params["form_id"]},
):
return JSONResponse({"error": "not_found"}, status_code=404)
data = await request.json()
response_ids = ResponseIdList(**data)
# Convert IDs to set to remove duplicates
ids = set(response_ids.ids)
cursor = request.state.db.responses.find(
{"_id": {"$in": list(ids)}}, # Convert here back to list, may throw error.
)
entries = [FormResponse(**submission) for submission in await cursor.to_list(None)]
actual_ids = {entry.id for entry in entries}
if len(ids) != len(actual_ids):
return JSONResponse(
{
"error": "responses_not_found",
"ids": list(ids - actual_ids),
},
status_code=404,
)
if any(entry.form_id != request.path_params["form_id"] for entry in entries):
return JSONResponse(
{
"error": "wrong_form",
"ids": [
entry.id for entry in entries if entry.id != request.path_params["form_id"]
],
},
status_code=400,
)
await request.state.db.responses.delete_many(
{
"_id": {"$in": list(actual_ids)},
},
)
return JSONResponse({"status": "ok"})
|
