Welcome to Part 3 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 complete the CRUD operations for the User
model and secure them using JWT authentication. By the end of this tutorial, you’ll have a fully functional user management system with secure endpoints.
What you learn in part 3
- How to implement CRUD operations for the
User
model. - How to secure endpoints using JWT authentication.
- How to test the endpoints using
curl
.
Prerequisites
Before we begin, ensure you’ve completed Part 2 of the series. You should have:
- A Flask application with SQLAlchemy configured.
- A
User
model with password hashing. - A user registration endpoint (
/api/register
).
Step 1: Install Flask-JWT-Extended
We’ll use the Flask-JWT-Extended
library to handle JWT authentication. Install it using pip
:
pip install flask-jwt-extended
Step 2: Configure JWT in the application
Update config.py
to include JWT configuration:
import os class Config: SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db') SQLALCHEMY_TRACK_MODIFICATIONS = False JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'your-secret-key') # Change this to a secure key
This code sets JWT_SECRET_KEY
in the Config
class by retrieving its value from the JWT_SECRET_KEY
environment variable. If the variable is not set, it defaults to 'your-secret-key'
. This key secures JSON Web Tokens (JWT) by signing and verifying tokens to protect user authentication.
Update app/__init__.py
to initialize JWT:
from flask import Flask 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) with app.app_context(): from . import routes app.register_blueprint(routes.bp) db.create_all() return app
This code initializes flask_jwt_extended
by creating a JWTManager
instance and configuring it with the Flask app. The jwt.init_app(app)
method integrates JWT support, enabling secure authentication and token management for protected routes.
Step 3: Implement the login endpoint
In app/routes.py
, add the login endpoint to generate JWT tokens:
from flask import Blueprint, request, jsonify from flask_jwt_extended import create_access_token from .models import User from . import db bp = Blueprint('api', __name__, url_prefix='/api') @bp.route('/login', methods=['POST']) def login(): data = request.get_json() # Validate input if not data or 'username' not in data or 'password' not in data: return jsonify({"error": "Username and password are required"}), 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
This code defines a login endpoint at /api/login
that authenticates users and generates JWT access tokens. It extracts the username
and password
from the request, verifies the credentials against the database, and returns an access token if authentication succeeds. If the credentials are invalid, it responds with an error message.
Step 4: Complete CRUD operations
Let’s implement the remaining CRUD operations for the User
model.
4.1: Fetch all users
Add a route to fetch all users:
@bp.route('/users', methods=['GET']) @jwt_required() def get_all_users(): users = User.query.all() return jsonify([{ "id": user.id, "username": user.username, "email": user.email } for user in users])
This code defines an endpoint at /api/users
that retrieves all users from the database. It requires authentication using @jwt_required()
, ensuring only authorized users can access it. The function queries all users, formats their details into a JSON list, and returns the response.
4.2: Fetch a single user
Add a route to fetch a single user by ID:
@bp.route('/users/<int:id>', methods=['GET']) @jwt_required() def get_user(id): user = User.query.get_or_404(id) return jsonify({ "id": user.id, "username": user.username, "email": user.email })
This code defines an endpoint at /api/users/<id>
that retrieves a specific user by their ID. It requires authentication using @jwt_required()
, ensuring only authorized users can access it. The function searches for the user in the database and returns their details in JSON format. If the user does not exist, it automatically returns a 404 Not Found error.
4.3: Update a user
Add a route to update a user’s details:
@bp.route('/users/<int:id>', methods=['PUT']) @jwt_required() def update_user(id): user = User.query.get_or_404(id) data = request.get_json() if 'username' in data: user.username = data['username'] if 'email' in data: user.email = data['email'] if 'password' in data: user.set_password(data['password']) db.session.commit() return jsonify({ "id": user.id, "username": user.username, "email": user.email })
This code defines an endpoint at /api/users/<id>
that updates a specific user’s details. It requires authentication using @jwt_required()
, ensuring only authorized users can modify user data. The function retrieves the user by ID, updates the username
, email
, or password
if provided, saves the changes to the database, and returns the updated user details in JSON format. If the user does not exist, it returns a 404 Not Found error.
4.4: Delete a user
Add a route to delete a user:
@bp.route('/users/<int:id>', methods=['DELETE']) @jwt_required() def delete_user(id): user = User.query.get_or_404(id) db.session.delete(user) db.session.commit() return jsonify({"message": "User deleted successfully"})
This code defines an endpoint at /api/users/<id>
that deletes a specific user. It requires authentication using @jwt_required()
, ensuring only authorized users can perform this action. The function retrieves the user by ID, removes them from the database, commits the changes, and returns a confirmation message in JSON format. If the user does not exist, it returns a 404 Not Found error.
Step 5: Test the endpoints
Let’s test the CRUD endpoints using curl
or Postman.
5.1: Log in to get a JWT token
curl -X POST -H "Content-Type: application/json" -d '{"username": "testuser", "password": "testpass"}' http://127.0.0.1:5000/api/login
This command sends a POST request to the /api/login
endpoint on http://127.0.0.1:5000
, allowing a user to log in. It includes a JSON payload with the username
and password
, which the server verifies for authentication. The -H "Content-Type: application/json"
header ensures the request is properly formatted as JSON. If the credentials are correct, the server responds with a JWT access token.
Replace "testuser"
with your actual username and "testpass"
with your real password before executing the command. 🚀
5.2: Fetch all users
curl -H "Authorization: Bearer <your-access-token>" http://127.0.0.1:5000/api/users
5.3: Fetch a single user
curl -H "Authorization: Bearer <your-access-token>" http://127.0.0.1:5000/api/users/1
5.4: Update a user
curl -X PUT -H "Authorization: Bearer <your-access-token>" -H "Content-Type: application/json" -d '{"username": "updateduser"}' http://127.0.0.1:5000/api/users/1
5.5: Delete a user
curl -X DELETE -H "Authorization: Bearer <your-access-token>" http://127.0.0.1:5000/api/users/1
Full code for part 3
You can find the complete code for this tutorial in the GitHub repository.
What’s Next?
In Part 4, we’ll add error handling and input validation to make the application more robust. Stay tuned!
Facebook Comments