Error handler returns wrong body

[voider1]

I have defined an error handler like so:

@api.errorhandler(ValidationError)
def handle_validation_error(error):
    return {"errors": error.messages}, 422

In this case I'm catching a validation error, however whenever this error occurs the invalid request body is used as the response instead of the response I return in the errorhandler. The HTTP response code however is correct.

[miquelvir]

I have tried this and I get the expected body.

from flask import Flask
from flask_restx import Api, Resource, fields, ValidationError

app = Flask(__name__)
api = Api(app, version='1.0', title='TodoMVC API',
    description='A simple TodoMVC API',
)

@api.errorhandler(ValidationError)
def handle_validation_error(error):
    return {"errors": "this is an error"}, 422

ns = api.namespace('todos', description='TODO operations')

todo = api.model('Todo', {
    'id': fields.Integer(readonly=True, description='The task unique identifier'),
    'task': fields.String(required=True, description='The task details')
})

@ns.route('/<int:id>')
@ns.response(404, 'Todo not found')
@ns.param('id', 'The task identifier')
class Todo(Resource):
    '''Show a single todo item and lets you delete them'''
    @ns.doc('get_todo')
    @ns.marshal_with(todo)
    def get(self, id):
        raise ValidationError("ddd")

if __name__ == '__main__':
    app.run(debug=False)

Response has code 422 and body:

{
  "errors": "this is an error",
  "message": "ddd"
}

[voider1]

The only difference between your and mine code is that you use a namespace to register your resource to, while I register all my resources to the api directly. I also use the api.add_resource() function to assign a resource to a certain endpoint. But that shouldn't make any difference.

[miquelvir]

please provide your full code for me to repeat the bug

[voider1]

from datetime import datetime
import json
from typing import Any, Dict, Optional
import os

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_restx import Api, Resource
from flask_jwt_extended import get_jwt_identity
from marshmallow import ValidationError, fields, pre_load, validate
from marshmallow.decorators import pre_dump

app = Flask(__name__, instance_relative_config=True)
app.config.from_mapping(
    SECRET_KEY="dev",
    JWT_SECRET_KEY="dev",
    SQLALCHEMY_DATABASE_URI=
    f"sqlite:///{os.path.join(app.instance_path, 'api.db')}",
    PROPAGATE_EXCEPTIONS=True,
    Debug=True,
    CASBIN_MODEL_CONF="model.conf",
    SQLALCHEMY_TRACK_MODIFICATIONS=False,
)
db = SQLAlchemy(app)
api = Api(app)
ma = Marshmallow(app)

try:
    os.makedirs(app.instance_path)
except OSError:
    pass

class Base:
    def save(self) -> None:
        db.session.add(self)
        db.session.commit()

    def delete(self):
        db.session.delete(self)
        db.session.commit()

class UserModel(db.Model, Base):
    """The user table in the database which stores all basic info.

    The attributes field contains authorization related information.
    This information has been encoded as a JSON string and stored as a
    piece of text.
    """

    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.Text, nullable=False, unique=True)
    password_hash = db.Column(db.Text, nullable=False)
    full_name = db.Column(db.Text, nullable=False)
    registration_date = db.Column(db.DateTime)
    attributes = db.Column(db.Text)

    def __init__(self, email: str, password: str, full_name: str):
        self.email = email
        self.set_password(password)
        self.full_name = full_name
        now = datetime.now()
        self.registration_date = now
        self.attributes = json.dumps(self.default_attributes())

    def __repr__(self) -> str:
        return f"<User {self.full_name}, email: {self.email} registered on {self.registration_date} attributes: {self.attributes}"

    def set_password(self, password: str) -> None:
        """Hashes the given password using the Argon2 hashing
        algorithm.
        """
        ph = PasswordHasher()
        self.password_hash = ph.hash(password)

    def check_password(self, password: str) -> bool:
        """Checks if the given password is valid.

        The given password will be hashed using the Argon2 hashing
        algorithm and will be compared to the hash stored in the database.
        """
        ph = PasswordHasher()
        expected_hash = self.password_hash

        try:
            return ph.verify(expected_hash, password)
        except VerifyMismatchError:
            return False

    def get_attributes(self) -> Dict[str, bool]:
        """Retrieves the user's attributes as a dictionary."""
        return json.loads(self.attributes)

    def set_attribute(self, attribute: str) -> None:
        """Sets a certain attribute to True."""
        attributes = self.get_attributes()
        attributes[attribute] = True
        self.attributes = json.dumps(attributes)

    def unset_attribute(self, attribute: str) -> None:
        """Sets a certain attribute to False."""
        attributes = self.get_attributes()
        attributes[attribute] = False
        self.attributes = json.dumps(attributes)

    @staticmethod
    def default_attributes() -> Dict[str, bool]:
        """Returns the default attributes for a new user."""
        return {"user": True, "admin": False}

class UserSchema(ma.Schema):
    """
    The user schema is used to verify the signup process and to output
    a user's profile.
    """

    id = fields.Int()
    email = fields.Email(required=True)
    password = fields.Str(
        required=True,
        validate=validate.Regexp(
            "^(?=.*[A-Za-z])(?=.*\\d)(?=.*[@$!%*#?&])[A-Za-z\\d@$!%*#?&]{10,64}$"
        ),
    )
    full_name = fields.Str(required=True)
    registrated_date = fields.DateTime()

    # Clean up data.
    @pre_load
    def process_input(self, data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        if data is None:
            return data
        email = data.get("email")
        if email is not None:
            data["email"] = email.lower().strip()
        return data

user_schema = UserSchema()

@api.errorhandler(ValidationError)
def handle_validation_error(error):
    return {"error": ["foo"]}, 422

class User(Resource):
    # This endpoint is to register a user.
    def post(self):
        json_input = request.get_json()

        # try:
        data = user_schema.load(json_input)
        # except ValidationError as err:
        #     return {"errors": err.messages}, 422

        if UserModel.query.filter_by(email=data["email"]).first():
            return {"message": "email exists"}

        user = UserModel(**data)
        user.save()

        data = user_schema.dump(user)
        return data, 201

api.add_resource(User, "/user")

For the argon2 encryption I am using the argon2-cffi library, not the normal argon2 one.

[miquelvir]

Please provide a proper MRE (https://stackoverflow.com/help/minimal-reproducible-example).

[voider1]

You are right, my bad. I have updated the previous comment with a MRE which showcases the bug. Thanks for your patience and help up until now.