custom validate method for fastapi leveraging pydantic's validation error classes

readme.md

the idea was to have a good and user friendly error response which is like the message at end of this file. it shows the error type, location, message and the input caused the error, for each validation error that any of those validation methods like validate_identification could raise.

# validators.py
from pydantic_core import InitErrorDetails, PydanticCustomError

from . import error_messages, regex_patterns


def generate_error(err_type: str, message: str, loc: str, value: str, key: str | None = None):
    return InitErrorDetails(
        type=PydanticCustomError(err_type, message),
        loc=(loc, key) if key else (loc,),
        input=value,
        ctx={},
    )


def validate_identification(value: str, errors: list) -> str | None:
    pattern = re.compile(regex_patterns.IDENTIFICATION)
    if not pattern.match(value):
        errors.append(
            generate_error(
                err_type="value_error.invalid_value",
                message="Invalid identification",
                loc="identification",
                value=value,
            ),
        )
        return None
    return value


def validate_rule_and_type(value: str, errors: list) -> str | None:
    pattern = re.compile(regex_patterns.RULE_AND_TYPE)
    matched = pattern.match(value)
    if not matched:
        errors.append(
            generate_error(
                err_type="value_error.invalid_value",
                message="Invalid rule and type",
                loc="rule_and_type",
                value=value,
            ),
        )
        return None
    return matched.groupdict()

# api.py
from contextlib import contextmanager

from fastapi import Body, FastAPI, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import ValidationError

from backend.validators import (
    validate_identification,
    validate_rule_and_type,
)

app = FastAPI()

@contextmanager
def error_collector():
    errors = []
    yield errors
    if errors:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=jsonable_encoder(
                ValidationError.from_exception_data(
                    title="analyzer",
                    line_errors=errors,
                ).errors(),
            ),
        )

@app.post("/")
async def process_message(message: str = Body(embed=True)) -> str:
    sections = message[5:-1].split("-")  # Remove `(FPL-` and `)` before splitting

    # Validate number of sections
    if len(sections) != 9:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Expected 9 sections, found {len(sections)}.",
        )

    with error_collector() as errors:
        validate_identification(sections[0], errors)
        validate_rule_and_type(message[1], errors)

    return ""

{
  "detail": [
    {
      "type": "value_error.invalid_value",
      "loc": [
        "identification"
      ],
      "msg": "Invalid identification",
      "input": "xACF402"
    },
    {
      "type": "value_error.invalid_value",
      "loc": [
        "rule_and_type"
      ],
      "msg": "Invalid rule and type",
      "input": "INx"
    }
  ]
}