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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
"""
Returns all form responses by form ID.
"""
import logging
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.models import FormResponse, ResponseList
from backend.route import Route
from backend.validation import ErrorMessage, OkayResponse, api
logger = logging.getLogger(__name__)
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", "admin"])
@api.validate(
resp=Response(HTTP_200=ResponseList, HTTP_404=ErrorMessage),
tags=["forms", "responses"]
)
async def get(self, request: Request) -> JSONResponse:
"""Returns all form responses by form ID."""
if not await request.state.db.forms.find_one(
{"_id": request.path_params["form_id"]}
):
return JSONResponse({"error": "not_found"}, 404)
cursor = request.state.db.responses.find(
{"form_id": request.path_params["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."""
form_id = {"_id": request.path_params["form_id"]}
logger.info(f"Bulk deleting responses for form with ID: {form_id.get('_id')}")
if not await request.state.db.forms.find_one(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": list(
entry.id for entry in entries
if entry.id != request.path_params["form_id"]
)
},
status_code=400
)
logger.debug(f"Executing deletion for the following responses: {','.join(actual_ids)}")
await request.state.db.responses.delete_many(
{
"_id": {"$in": list(actual_ids)}
}
)
return JSONResponse({"status": "ok"})
|
