138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
"""Validation for task-input manifest values."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from typing import BinaryIO, Protocol
|
|
|
|
from .discovery import InputField
|
|
|
|
|
|
class Upload(Protocol):
|
|
filename: str | None
|
|
content_length: int | None
|
|
stream: BinaryIO
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PendingUpload:
|
|
field_name: str
|
|
extension: str
|
|
maximum_bytes: int | None
|
|
stream: BinaryIO
|
|
|
|
|
|
def validate_inputs(
|
|
fields: Sequence[InputField],
|
|
values: Mapping[str, list[str]],
|
|
) -> tuple[dict[str, str | int | list[str]], dict[str, str]]:
|
|
result: dict[str, str | int | list[str]] = {}
|
|
errors: dict[str, str] = {}
|
|
for field in fields:
|
|
if field.type == "file":
|
|
continue
|
|
raw_values = values.get(field.name, [])
|
|
value, error = validate_field(field, raw_values)
|
|
if error:
|
|
errors[field.name] = error
|
|
elif value is not None:
|
|
result[field.name] = value
|
|
return result, errors
|
|
|
|
|
|
def validate_uploads(
|
|
fields: Sequence[InputField],
|
|
uploads: Mapping[str, Upload],
|
|
) -> tuple[dict[str, PendingUpload], dict[str, str]]:
|
|
result: dict[str, PendingUpload] = {}
|
|
errors: dict[str, str] = {}
|
|
for field in fields:
|
|
if field.type != "file":
|
|
continue
|
|
upload = uploads.get(field.name)
|
|
filename = upload.filename if upload is not None else None
|
|
if upload is None or not filename:
|
|
if field.required:
|
|
errors[field.name] = "Choose a file."
|
|
continue
|
|
extension = Path(filename).suffix.lower()
|
|
if field.accept and extension not in field.accept:
|
|
errors[field.name] = f"Choose a file with one of: {', '.join(field.accept)}."
|
|
continue
|
|
if upload.content_length is not None and field.maximum_bytes is not None and upload.content_length > field.maximum_bytes:
|
|
errors[field.name] = f"Choose a file no larger than {field.maximum_bytes} bytes."
|
|
continue
|
|
result[field.name] = PendingUpload(
|
|
field_name=field.name,
|
|
extension=extension,
|
|
maximum_bytes=field.maximum_bytes,
|
|
stream=upload.stream,
|
|
)
|
|
return result, errors
|
|
|
|
|
|
def validate_field(
|
|
field: InputField,
|
|
raw_values: list[str],
|
|
) -> tuple[str | int | list[str] | None, str | None]:
|
|
if field.type == "multi_choice":
|
|
return validate_multi_choice(field, raw_values)
|
|
if len(raw_values) > 1:
|
|
return None, "Only one value is allowed."
|
|
raw = raw_values[0] if raw_values else ""
|
|
if not raw:
|
|
if field.default is not None:
|
|
return field.default, None
|
|
if field.required:
|
|
return None, "This field is required."
|
|
return None, None
|
|
if field.type == "integer":
|
|
try:
|
|
value = int(raw)
|
|
except ValueError:
|
|
return None, "Enter a whole number."
|
|
if field.minimum is not None and value < field.minimum:
|
|
return None, f"Enter a value of at least {field.minimum}."
|
|
if field.maximum is not None and value > field.maximum:
|
|
return None, f"Enter a value no greater than {field.maximum}."
|
|
if field.step is not None and (value - (field.minimum or 0)) % field.step:
|
|
return None, f"Enter a value in increments of {field.step}."
|
|
return value, None
|
|
if field.type == "date":
|
|
try:
|
|
date.fromisoformat(raw)
|
|
except ValueError:
|
|
return None, "Enter a valid date."
|
|
elif field.type == "datetime":
|
|
try:
|
|
datetime.fromisoformat(raw)
|
|
except ValueError:
|
|
return None, "Enter a valid date and time."
|
|
elif field.type == "choice" and raw not in {option.value for option in field.options}:
|
|
return None, "Choose one of the available options."
|
|
elif field.type == "text" and field.pattern and not re.fullmatch(field.pattern, raw):
|
|
return None, "Enter a value in the required format."
|
|
return raw, None
|
|
|
|
|
|
def validate_multi_choice(
|
|
field: InputField,
|
|
raw_values: list[str],
|
|
) -> tuple[list[str] | None, str | None]:
|
|
if not raw_values:
|
|
if field.default is not None:
|
|
return list(field.default), None
|
|
if field.required:
|
|
return None, "Choose at least one option."
|
|
return None, None
|
|
allowed = {option.value for option in field.options}
|
|
if any(value not in allowed for value in raw_values):
|
|
return None, "Choose only from the available options."
|
|
if len(raw_values) != len(set(raw_values)):
|
|
return None, "Each option can be selected only once."
|
|
return raw_values, None
|