60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Request validation shared by every slice. Errors carry a status, a fix-it message, and the field."""
|
|
|
|
from flask import request
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(self, status: int, message: str, field: str | None = None):
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.message = message
|
|
self.field = field
|
|
|
|
|
|
def json_body(allowed: set[str]) -> dict:
|
|
body = request.get_json(silent=True)
|
|
if not isinstance(body, dict):
|
|
raise ApiError(400, "Request body must be a JSON object")
|
|
unknown = sorted(set(body) - allowed)
|
|
if unknown:
|
|
raise ApiError(
|
|
400,
|
|
f"Unknown field(s): {', '.join(unknown)}. Allowed: {', '.join(sorted(allowed))}",
|
|
unknown[0],
|
|
)
|
|
return body
|
|
|
|
|
|
def require_utf8(value: str, field: str) -> None:
|
|
try:
|
|
value.encode("utf-8")
|
|
except UnicodeEncodeError:
|
|
raise ApiError(
|
|
400,
|
|
f"{field} contains unpaired surrogate characters; send valid UTF-8 text",
|
|
field,
|
|
) from None
|
|
|
|
|
|
def text(body: dict, field: str, max_len: int) -> str:
|
|
value = body.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ApiError(400, f"{field} is required and must be non-blank text", field)
|
|
value = value.strip()
|
|
if len(value) > max_len:
|
|
raise ApiError(400, f"{field} must be at most {max_len} characters", field)
|
|
if "\x00" in value:
|
|
raise ApiError(400, f"{field} must not contain NUL characters", field)
|
|
require_utf8(value, field)
|
|
return value
|
|
|
|
|
|
def integer(body: dict, field: str, low: int, high: int) -> int:
|
|
value = body.get(field)
|
|
# bool is a subclass of int in Python; JSON true must not count as 1.
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise ApiError(400, f"{field} must be a whole number", field)
|
|
if not low <= value <= high:
|
|
raise ApiError(400, f"{field} must be between {low} and {high}", field)
|
|
return value
|