annif.openapi.validation

Complexity

Total Complexity

8

Size/Duplication

Total Lines	60
Duplicated Lines	0 %

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	CustomRequestBodyValidator._validate()	0	16	4
A	CustomRequestBodyValidator.__init__()	0	2	1
A	CustomFormDataValidator._validate_params_strictly()	0	11	2
A	CustomFormDataValidator.__init__()	0	2	1

"""Custom validator for the Annif API."""

from __future__ import annotations

import logging
from typing import Any

from connexion.exceptions import BadRequestProblem, ExtraParameterProblem
from connexion.json_schema import format_error_with_path
from connexion.validators import FormDataValidator, JSONRequestBodyValidator
from jsonschema.exceptions import ValidationError

logger = logging.getLogger("openapi.validation")


class CustomRequestBodyValidator(JSONRequestBodyValidator):
    """Custom request body validator that overrides the default error message for the
    'maxItems' validator for the 'documents' property to prevent logging request body
    with the contents of all documents."""

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

    def _validate(self, body: Any) -> dict | None:
        try:
            return self._validator.validate(body)
        except ValidationError as exception:
            if exception.validator == "maxItems" and list(exception.schema_path) == [
                "properties",
                "documents",
                "maxItems",
            ]:
                exception.message = "too many items"
            error_path_msg = format_error_with_path(exception=exception)
            logger.error(
                f"Validation error: {exception.message}{error_path_msg}",
                extra={"validator": "body"},
            )
            raise BadRequestProblem(detail=f"{exception.message}{error_path_msg}")


class CustomFormDataValidator(FormDataValidator):
    """Custom request body validator that allows additional metadata fields starting
    with 'metadata_' in the request body while rejecting other fields."""

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

    def _validate_params_strictly(self, data: dict) -> None:
        form_params = data.keys()
        spec_params = self._schema.get("properties", {}).keys()

        reduced_form_params = [
            fp for fp in form_params if not fp.startswith("metadata_")
        ]

        errors = set(reduced_form_params).difference(set(spec_params))
        if errors:
            raise ExtraParameterProblem(param_type="formData", extra_params=errors)