You are currently viewing Part 4: Error handling and input validation – Making your microservice robust and user-friendly

Part 4: Error handling and input validation – Making your microservice robust and user-friendly

  • Post author:
  • Post category:Python

Welcome to Part 4 of the tutorial series: “Build a Production-Ready User Management Microservice with Flask and SQLAlchemy: A Step-by-Step Guide”. In this part, we’ll enhance the application by adding error handling and input validation. By the end of this tutorial, your microservice will be more robust, user-friendly, and production-ready.

What you learn in part 4

  • How to add global error handling for common issues (e.g., 404, 500).
  • How to validate user input using Flask validators.
  • How to return meaningful error messages to the client.
  • How to test error handling and validation.

Prerequisites

Before we begin, ensure you’ve completed Part 1, Part 2, and Part 3 of the series. You should have:

  • A Flask application with SQLAlchemy configured.
  • A User model with password hashing.
  • CRUD operations for the User model.
  • JWT authentication for securing endpoints.

Step 1: Add global error handling

Let’s add global error handlers to handle common HTTP errors like 404 (Not Found) and 500 (Internal Server Error).

Update app/__init__.py

Add the following error handlers to app/__init__.py:

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from config import Config

db = SQLAlchemy()
jwt = JWTManager()

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)
    jwt.init_app(app)

    @app.errorhandler(404)
    def not_found(error):
        return  jsonify({"error": "Not Found", "message": "The requested resource was not found"}), 404
    
    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({"error": "Internal Server Error", "message": "Something went wrong on the server"}), 500

    with app.app_context():
        from . import routes
        app.register_blueprint(routes.bp)

        db.create_all()

    return app

The code defines two error handlers to improve API responses when errors occur. The @app.errorhandler(404) method catches “Not Found” errors and returns a JSON response with a 404 status code, informing the client that the requested resource does not exist. The @app.errorhandler(500) method handles “Internal Server Error” cases by returning a structured JSON response with a 500 status code, indicating a server-side issue. These handlers ensure that errors return meaningful messages instead of generic HTML error pages. 🚀

Step 2: Add input validation

Let’s validate user input for the /api/register and /api/login endpoints to ensure the data is correct before processing it.

Install Flask-Validators

We’ll use the marshmallow library for input validation. Install it using pip:

pip install marshmallow

Marshmallow is a powerful tool for handling data validation and serialization in Python applications. It’s widely used in web development, APIs, and microservices to ensure data integrity and consistency.

Create a validation schema

Create a new file app/schemas.py to define validation schemas:

from marshmallow import Schema, fields, ValidationError

class UserSchema(Schema):
    username = fields.Str(required=True)
    email = fields.Email(required=True)
    password = fields.Str(required=True)

user_schema = UserSchema()

The code defines a schema for user data validation using Marshmallow. The UserSchema class inherits from Schema and specifies three required fields: username (a string), email (a valid email address), and password (a string). The user_schema instance allows validation and serialization of user data, ensuring inputs meet the required format before processing them in the application. 🚀

Update the registration endpoint

Update the /api/register endpoint in app/routes.py to use the validation schema:

from .schemas import user_schema, ValidationError

@bp.route('/register', methods=['POST'])
def register():
    try:
        # Validate and deserialize the input data
        data = user_schema.load(request.get_json())
    except ValidationError as err:
        return jsonify({"error": "Validation Error", "message": err.messages}), 400
    
    # Check if username or email already exists
    if User.query.filter_by(username=data['username']).first():
        return jsonify({"error": "Username already exists"}), 400
    
    if User.query.filter_by(email=data['email']).first():
        return jsonify({"error": "Email already exists"}), 400
    
    # Create a new user
    new_user = User(username=data['username'], email=data['email'])
    new_user.set_password(data['password'])
    db.session.add(new_user)
    db.session.commit()

    return jsonify({
        "id": new_user.id,
        "username": new_user.username,
        "email": new_user.email
    }), 201

The code imports user_schema and ValidationError from .schemas to handle user input validation. It validates and deserializes incoming JSON data using user_schema.load(request.get_json()). If the data is invalid, the code raises a ValidationError and returns a JSON response with an error message. This ensures that only properly formatted user data proceeds to the registration process. 🚀

Update the login endpoint

Update the /api/login endpoint in app/routes.py to use the validation schema:

@bp.route('/login', methods=['POST'])
def login():
    try:
        # Validate and deserialize the input data
        data = user_schema.load(request.get_json)
    except ValidationError as err:
        return jsonify({"error": "Validation Error", "message": err.messages}), 400

    username = data['username']
    password = data['password']

    # Find the user
    user = User.query.filter_by(username=username).first()
    if user and user.check_password(password):
        access_token = create_access_token(identity=str(user.id))
        return jsonify(access_token=access_token), 200
    return jsonify({"error": "Invalid credentials"}), 401

The code validates and deserializes incoming JSON data using user_schema.load(request.get_json()). If the data is invalid, it raises a ValidationError and returns a JSON response with an error message. This ensures the login process only accepts properly formatted user input. 🚀

Step 3: Test error handling and validation

Let’s test the error handling and validation using curl.

Test 404 Error

Try accessing a non-existent endpoint:

curl http://127.0.0.1:5000/api/nonexistent

Expected response

{
  "error": "Not Found",
  "message": "The requested resource was not found"
}

Test validation errors

Send invalid data to the /api/register endpoint:

curl -X POST -H "Content-Type: application/json" -d '{"username": "testuser", "email": "invalid-email", "password": "testpass"}' http://127.0.0.1:5000/api/register

Expected response

{
  "error": "Validation Error",
  "messages": {
    "email": ["Not a valid email address."]
  }
}

Full code for part 4

You can find the complete code for this tutorial in the GitHub repository.

What’s Next?

In Part 5, we’ll add logging and monitoring to the application to track requests, errors, and performance. Stay tuned!

Facebook Comments

Tags: , ,