add validator page and api route

Author: rshewitt

app/api_schemas.py

@@ -1,7 +1,21 @@
 from enum import Enum as PyEnum
+import os
+import json
 
 from apiflask import Schema, validators
-from apiflask.fields import UUID, Boolean, DateTime, Dict, Enum, Integer, List, String
+from apiflask.fields import (
+    UUID,
+    Boolean,
+    DateTime,
+    Dict,
+    Enum,
+    Integer,
+    List,
+    String,
+    URL,
+)
+import marshmallow
+from marshmallow import ValidationError, validate
 
 from shared.constants import (
     ACTION_VALUES,
@@ -14,6 +28,8 @@
     SOURCE_TYPE_VALUES,
 )
 
+IS_PROD = os.getenv("FLASK_ENV") == "production"
+
 
 def _to_enum(name, values):
     """Make an iterable of values into a Python enum.
@@ -117,3 +133,57 @@ class SourceInfo(Schema):
     source_type = Enum(SOURCE_TYPE_ENUM, required=True)
     notification_frequency = Enum(NOTIFICATION_FREQUENCY_ENUM, required=True)
     collection_parent_url = String()
+
+
+class ValidatorInfo(Schema):
+    schema = String(
+        required=True,
+        validate=validators.OneOf(
+            [
+                "dcatus1.1: federal dataset",
+                "dcatus1.1: non-federal dataset",
+            ]
+        ),
+    )
+    fetch_method = String(
+        required=True,
+        validate=validators.OneOf(
+            [
+                "url",
+                "paste",
+            ]
+        ),
+    )
+    url = URL(require_tld=IS_PROD)
+
+    @marshmallow.validates_schema
+    def validate_url(self, data, **kwargs):
+        if data.get("fetch_method") == "url" and not data.get("url"):
+            raise ValidationError("'url' field is required when fetch_method is 'url'")
+
+    json_text = String()
+
+    @marshmallow.validates_schema
+    def validate_json_text(self, data, **kwargs):
+        if data.get("fetch_method") == "paste" and not data.get("json_text"):
+            raise ValidationError(
+                "'json_text' field is required when fetch_method is 'paste'"
+            )
+        try:
+            json.loads(data.get("json_text"))
+        except json.JSONDecodeError:
+            raise ValidationError("Invalid JSON")
+
+
+class ValidationResultSchema(Schema):
+    validation_errors = List(
+        List(
+            String(),
+            validate=validate.Length(equal=2),
+        ),
+        required=True,
+    )
+
+
+class ValidationErrorResponseSchema(Schema):
+    error = String(required=True)

app/forms.py

@@ -1,6 +1,7 @@
 import ast
 import os
 import re
+import json
 
 from flask_wtf import FlaskForm
 from wtforms import BooleanField, SelectField, StringField, SubmitField, TextAreaField
@@ -18,6 +19,18 @@
 is_prod = os.getenv("FLASK_ENV") == "production"
 
 
+def validate_json_format(form, field):
+    """
+    Custom validator to check if the field data is valid JSON.
+    """
+    try:
+        json.loads(field.data)
+    except json.JSONDecodeError:
+        raise ValidationError(
+            "Invalid JSON format. Please ensure the data is a valid JSON string."
+        )
+
+
 def validate_email_list(form, field):
     emails = field.data
     for email in emails:
@@ -165,3 +178,53 @@ class HarvestTriggerForm(FlaskForm):
 class OrganizationTriggerForm(FlaskForm):
     edit = SubmitField("Edit")
     delete = SubmitField("Delete")
+
+
+def url_paste_validate(form, field):
+    if form.fetch_method.data == "url":
+        if not form.url.data:
+            raise ValidationError("URL is required.")
+    elif form.fetch_method.data == "paste":
+        if not form.json_text.data:
+            raise ValidationError("JSON input is required.")
+        else:
+            validate_json_format(form, form.json_text)
+
+    return True
+
+
+class ValidatorForm(FlaskForm):
+    schema = SelectField(
+        "Schema",
+        choices=[
+            "dcatus1.1: federal dataset",
+            "dcatus1.1: non-federal dataset",
+        ],
+        validators=[DataRequired()],
+    )
+    fetch_method = SelectField(
+        "Fetch Method",
+        choices=[
+            ("url", "Fetch from URL"),
+            ("paste", "Paste JSON"),
+        ],
+        validators=[DataRequired()],
+    )
+    url = StringField(
+        "URL",
+        validators=[url_paste_validate],
+        filters=[strip_filter],
+    )
+
+    def validate_url(self, field):
+        if self.fetch_method.data == "url" and field.data:
+            try:
+                URL(require_tld=is_prod)(self, field)
+            except ValidationError:
+                raise ValidationError("Invalid URL")
+
+    json_text = TextAreaField(
+        "DCATUS Catalog JSON Input",
+        validators=[url_paste_validate],
+    )
+    submit = SubmitField("Validate")