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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
"""Returns, updates or deletes a single form given an ID."""
import enum
import json.decoder
import deepmerge
from pydantic import BaseModel, ValidationError
from spectree.response import Response
from starlette.authentication import requires
from starlette.requests import Request
from starlette.responses import JSONResponse
from backend import constants, discord
from backend.models import Form
from backend.route import Route
from backend.routes.forms.discover import AUTH_FORM
from backend.validation import ErrorMessage, OkayResponse, api
PUBLIC_FORM_FEATURES = (constants.FormFeatures.OPEN, constants.FormFeatures.DISCOVERABLE)
class SubmissionPrecheckSeverity(enum.StrEnum):
SECONDARY = "secondary"
WARNING = "warning"
DANGER = "danger"
class SubmissionProblem(BaseModel):
severity: SubmissionPrecheckSeverity
message: str
class SubmissionPrecheck(BaseModel):
problems: list[SubmissionProblem] = []
can_submit: bool = True
class FormWithAncillaryData(Form):
"""
Form model with ancillary data for the form.
This is used to return the form with additional information such as
whether the user has edit access or not.
"""
submission_precheck: SubmissionPrecheck = SubmissionPrecheck()
class SingleForm(Route):
"""
Returns, updates or deletes a single form given an ID.
Returns all fields for admins, otherwise only public fields.
"""
name = "form"
path = "/{form_id:str}"
@api.validate(
resp=Response(HTTP_200=FormWithAncillaryData, HTTP_404=ErrorMessage), tags=["forms"]
)
async def get(self, request: Request) -> JSONResponse:
"""Returns single form information by ID."""
form_id = request.path_params["form_id"].lower()
if form_id == AUTH_FORM.id:
# Empty form for login purposes
data = AUTH_FORM.dict(admin=False)
# Add in empty ancillary data
data["submission_precheck"] = SubmissionPrecheck().dict()
return JSONResponse(data)
try:
await discord.verify_edit_access(form_id, request)
admin = True
except discord.FormNotFoundError:
return JSONResponse({"error": "not_found"}, status_code=404)
except discord.UnauthorizedError:
admin = False
filters = {
"_id": form_id,
}
if not admin:
filters["features"] = {"$in": ["OPEN", "DISCOVERABLE"]}
form = await request.state.db.forms.find_one(filters)
form = FormWithAncillaryData(**form)
if not form:
return JSONResponse({"error": "not_found"}, status_code=404)
if constants.FormFeatures.OPEN.value not in form.features:
form.submission_precheck.problems.append(
SubmissionProblem(
severity=SubmissionPrecheckSeverity.DANGER,
message="This form is not open for submissions at the moment.",
)
)
form.submission_precheck.can_submit = False
elif constants.FormFeatures.UNIQUE_RESPONDER.value in form.features:
user_id = request.user.payload["id"] if request.user.is_authenticated else None
if user_id:
existing_response = await request.state.db.responses.find_one({
"form_id": form_id,
"user.id": user_id,
})
if existing_response:
form.submission_precheck.problems.append(
SubmissionProblem(
severity=SubmissionPrecheckSeverity.DANGER,
message="You have already submitted a response to this form.",
)
)
form.submission_precheck.can_submit = False
else:
form.submission_precheck.problems.append(
SubmissionProblem(
severity=SubmissionPrecheckSeverity.SECONDARY,
message="You must login at the bottom of the page before submitting this form.",
)
)
return JSONResponse(form.dict(admin=admin))
@requires(["authenticated"])
@api.validate(
resp=Response(
HTTP_200=OkayResponse,
HTTP_400=ErrorMessage,
HTTP_404=ErrorMessage,
),
tags=["forms"],
)
async def patch(self, request: Request) -> JSONResponse:
"""Updates form by ID."""
try:
data = await request.json()
except json.decoder.JSONDecodeError:
return JSONResponse({"error": "Expected a JSON body."}, 400)
form_id = request.path_params["form_id"].lower()
await discord.verify_edit_access(form_id, request)
if raw_form := await request.state.db.forms.find_one({"_id": form_id}):
if "_id" in data or "id" in data:
if (data.get("id") or data.get("_id")) != form_id:
return JSONResponse({"error": "locked_field"}, status_code=400)
# Build Data Merger
merge_strategy = [
(dict, ["merge"]),
]
merger = deepmerge.Merger(merge_strategy, ["override"], ["override"])
# Merge Form Data
updated_form = merger.merge(raw_form, data)
try:
form = Form(**updated_form)
except ValidationError as e:
return JSONResponse(e.errors(), status_code=422)
await request.state.db.forms.replace_one({"_id": form_id}, form.dict())
return JSONResponse(form.dict())
return JSONResponse({"error": "not_found"}, status_code=404)
@requires(["authenticated", "admin"])
@api.validate(
resp=Response(HTTP_200=OkayResponse, HTTP_401=ErrorMessage, HTTP_404=ErrorMessage),
tags=["forms"],
)
async def delete(self, request: Request) -> JSONResponse:
"""Deletes form by ID."""
form_id = request.path_params["form_id"].lower()
await discord.verify_edit_access(form_id, request)
await request.state.db.forms.delete_one({"_id": form_id})
await request.state.db.responses.delete_many({"form_id": form_id})
return JSONResponse({"status": "ok"})
|
