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,59 @@ 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":
+            if not data.get("json_text"):
+                raise ValidationError(
+                    "'json_text' field is required when fetch_method is 'paste'"
+                )
+            else:
+                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")

app/routes.py

@@ -51,19 +51,25 @@
     QueryInfo,
     RecordInfo,
     SourceInfo,
+    ValidatorInfo,
+    ValidationResultSchema,
+    ValidationErrorResponseSchema,
 )
 from .auth import LoginRequiredAuth
 from .forms import (
     HarvestSourceForm,
     HarvestTriggerForm,
     OrganizationForm,
     OrganizationTriggerForm,
+    ValidatorForm,
 )
 from .paginate import Pagination
 from .util import (
     make_new_org_contract,
     make_new_record_error_contract,
     make_new_source_contract,
+    fetch_json_from_url,
+    validate_records,
 )
 
 logger = logging.getLogger("harvest_admin")
@@ -1331,6 +1337,85 @@ def json_builder_query(**kwargs):
         return "Error with query", 400
 
 
+@api.route("/api/validate", methods=["POST"])
+@api.input(ValidatorInfo)
+@api.output(ValidationResultSchema, status_code=200)
+@api.doc(
+    summary="Validate a DCAT catalog against a v1.1 schema",
+    description="Downloads or parses a DCATUS catalog and validates each dataset.",
+    responses={
+        500: {
+            "description": "Failed to download or process the catalog",
+            "content": {"application/json": {"schema": ValidationErrorResponseSchema}},
+        },
+    },
+)
+def validator(json_data):
+    """
+    api route for validating v1.1 dcatus catalogs
+    """
+    errors = []
+    data = []
+
+    try:
+        if json_data["fetch_method"] == "url":
+            data = fetch_json_from_url(json_data["url"])
+
+        if json_data["fetch_method"] == "paste":
+            data = json.loads(json_data["json_text"])
+
+        errors = validate_records(data, json_data["schema"])
+    except Exception as e:
+        logger.error(f"API Validator error :: {repr(e)}")
+        return make_response(
+            jsonify(
+                {"error": "API Validator error: failed to validate dcatus catalog"}
+            ),
+            500,
+        )
+
+    return make_response(
+        jsonify({"validation_errors": errors}),
+        200,
+    )
+
+
+@main.route("/validate/", methods=["GET", "POST"])
+def view_validators():
+    """
+    view for validating v1.1 dcatus catalogs using form
+    """
+    data = []
+    errors = []
+    submitted = False
+
+    form = ValidatorForm()
+    if form.validate_on_submit():
+        if form.fetch_method.data == "url":
+            try:
+                data = fetch_json_from_url(form.url.data)
+            except Exception as e:
+                form.url.errors.append(str(e))
+        if form.fetch_method.data == "paste":
+            data = json.loads(form.json_text.data)
+
+        errors = validate_records(data, form.schema.data)
+
+        if not form.errors:
+            submitted = True
+
+    data = {
+        "record_errors": errors,
+    }
+
+    return render_template(
+        "view_validators.html",
+        form=form,
+        data=data,
+        submitted=submitted,
+    )
+
+
 def register_routes(app):
     app.register_blueprint(main)
     app.register_blueprint(api)

app/templates/base.html

@@ -37,6 +37,7 @@
             <li><a href="{{ url_for('main.organization_list') }}">Organizations</a></li>
             <li><a href="{{ url_for('main.harvest_source_list') }}">Harvest Sources</a></li>
             <li><a href="{{ url_for('main.view_metrics') }}">Metrics</a></li>
+            <li><a href="{{ url_for('main.view_validators') }}">Validators</a></li>
             {% if session['user'] %}
             <li>{{ session['user'] }}</li>
             <li><a href="{{ url_for('main.logout') }}">Logout</a></li>