How to Build a REST API in Python with Flask: Complete Step-by-Step Guide
Building robust REST APIs is a fundamental skill for modern web development. Flask, Python's lightweight and flexible web framework, provides an excellent foundation for creating scalable APIs. This comprehensive tutorial will guide you through building a complete REST API from scratch, covering everything from basic setup to advanced authentication and CRUD operations.
What is Flask and Why Choose It for REST APIs?
Flask is a micro web framework written in Python that's designed to make getting started with web development quick and easy, with the ability to scale up to complex applications. Unlike heavyweight frameworks, Flask gives you the flexibility to choose your components and structure your application as needed.
Key advantages of Flask for API development: - Lightweight and minimalist - Highly customizable and extensible - Excellent documentation and community support - Built-in development server and debugger - RESTful request dispatching - Secure cookie support for client-side sessions
Prerequisites and Environment Setup
Before diving into development, ensure you have the following prerequisites:
System Requirements: - Python 3.7 or higher - pip (Python package installer) - Text editor or IDE (VS Code, PyCharm recommended) - Basic understanding of HTTP methods and REST principles
Setting up your development environment:
`bash
Create a new directory for your project
mkdir flask-rest-api cd flask-rest-apiCreate a virtual environment
python -m venv venvActivate the virtual environment
On Windows:
venv\Scripts\activateOn macOS/Linux:
source venv/bin/activateInstall required packages
pip install Flask Flask-SQLAlchemy Flask-JWT-Extended python-dotenvCreate a requirements.txt file to track dependencies:
`txt
Flask==2.3.3
Flask-SQLAlchemy==3.0.5
Flask-JWT-Extended==4.5.2
python-dotenv==1.0.0
Werkzeug==2.3.7
`
Project Structure and Organization
Organizing your Flask application properly from the start ensures maintainability and scalability. Here's the recommended project structure:
`
flask-rest-api/
├── app/
│ ├── __init__.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── routes/
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── users.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── config.py
├── run.py
├── .env
└── requirements.txt
`
Creating Your First Flask Application
Let's start by creating the basic Flask application structure:
config.py:
`python
import os
from datetime import timedelta
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'your-secret-key-here'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-string'
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
`
app/__init__.py:
`python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from config import Config
Initialize extensions
db = SQLAlchemy() jwt = JWTManager()def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
# Initialize extensions with app
db.init_app(app)
jwt.init_app(app)
# Register blueprints
from app.routes.auth import auth_bp
from app.routes.users import users_bp
app.register_blueprint(auth_bp, url_prefix='/api/auth')
app.register_blueprint(users_bp, url_prefix='/api/users')
return app
`
run.py:
`python
from app import create_app, db
from dotenv import load_dotenv
load_dotenv()
app = create_app()
@app.before_first_request def create_tables(): db.create_all()
if __name__ == '__main__':
app.run(debug=True)
`
Understanding REST Principles and HTTP Methods
REST (Representational State Transfer) is an architectural style that defines a set of constraints for creating web services. A RESTful API uses standard HTTP methods to perform operations on resources.
Core HTTP Methods: - GET: Retrieve data from the server - POST: Create new resources - PUT: Update existing resources (complete replacement) - PATCH: Partial update of resources - DELETE: Remove resources
REST Best Practices: - Use nouns for resource names (not verbs) - Use HTTP status codes appropriately - Implement consistent URL patterns - Version your API - Use JSON for data exchange
Building Your First Route
Let's create a simple "Hello World" route to test our setup:
app/routes/__init__.py:
`python
from flask import Blueprint
This file can remain empty or contain shared route utilities
app/routes/users.py:
`python
from flask import Blueprint, jsonify, request
from app import db
from app.models.user import User
from flask_jwt_extended import jwt_required, get_jwt_identity
users_bp = Blueprint('users', __name__)
@users_bp.route('/hello', methods=['GET']) def hello_world(): return jsonify({ 'message': 'Hello, World!', 'status': 'success' }), 200
@users_bp.route('/', methods=['GET'])
def get_all_users():
users = User.query.all()
return jsonify({
'users': [user.to_dict() for user in users],
'count': len(users)
}), 200
`
Database Models with SQLAlchemy
SQLAlchemy is Flask's recommended ORM (Object-Relational Mapping) tool. Let's create our User model:
app/models/user.py:
`python
from app import db
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
first_name = db.Column(db.String(50), nullable=True)
last_name = db.Column(db.String(50), nullable=True)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __init__(self, username, email, password, first_name=None, last_name=None):
self.username = username
self.email = email
self.set_password(password)
self.first_name = first_name
self.last_name = last_name
def set_password(self, password):
"""Hash and set the user's password"""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Check if the provided password matches the hash"""
return check_password_hash(self.password_hash, password)
def to_dict(self):
"""Convert user object to dictionary"""
return {
'id': self.id,
'username': self.username,
'email': self.email,
'first_name': self.first_name,
'last_name': self.last_name,
'is_active': self.is_active,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
def __repr__(self):
return f'`
Implementing CRUD Operations
CRUD operations form the backbone of most APIs. Let's implement comprehensive CRUD functionality for our User model:
Enhanced app/routes/users.py:
`python
from flask import Blueprint, jsonify, request
from app import db
from app.models.user import User
from flask_jwt_extended import jwt_required, get_jwt_identity
from sqlalchemy.exc import IntegrityError
users_bp = Blueprint('users', __name__)
CREATE - Add a new user
@users_bp.route('/', methods=['POST']) def create_user(): try: data = request.get_json() # Validate required fields required_fields = ['username', 'email', 'password'] for field in required_fields: if field not in data or not data[field]: return jsonify({ 'error': f'{field} is required' }), 400 # Create new user user = User( username=data['username'], email=data['email'], password=data['password'], first_name=data.get('first_name'), last_name=data.get('last_name') ) db.session.add(user) db.session.commit() return jsonify({ 'message': 'User created successfully', 'user': user.to_dict() }), 201 except IntegrityError: db.session.rollback() return jsonify({ 'error': 'Username or email already exists' }), 409 except Exception as e: db.session.rollback() return jsonify({ 'error': 'An error occurred while creating the user' }), 500READ - Get all users
@users_bp.route('/', methods=['GET']) @jwt_required() def get_all_users(): try: page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) users = User.query.paginate( page=page, per_page=per_page, error_out=False ) return jsonify({ 'users': [user.to_dict() for user in users.items], 'pagination': { 'page': users.page, 'pages': users.pages, 'per_page': users.per_page, 'total': users.total } }), 200 except Exception as e: return jsonify({ 'error': 'An error occurred while fetching users' }), 500READ - Get a specific user
@users_bp.route('/UPDATE - Update a user
@users_bp.route('/DELETE - Delete a user
@users_bp.route('/Implementing JWT Authentication
JSON Web Tokens (JWT) provide a secure way to authenticate API requests. Let's implement comprehensive authentication:
app/routes/auth.py:
`python
from flask import Blueprint, jsonify, request
from app import db
from app.models.user import User
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity,
get_jwt
)
from datetime import datetime, timedelta
auth_bp = Blueprint('auth', __name__)
Token blacklist (in production, use Redis or database)
blacklisted_tokens = set()@auth_bp.route('/register', methods=['POST']) def register(): try: data = request.get_json() # Validate required fields required_fields = ['username', 'email', 'password'] for field in required_fields: if field not in data or not data[field]: return jsonify({ 'error': f'{field} is required' }), 400 # Check if user already exists if User.query.filter_by(username=data['username']).first(): return jsonify({ 'error': 'Username already exists' }), 409 if User.query.filter_by(email=data['email']).first(): return jsonify({ 'error': 'Email already exists' }), 409 # Create new user user = User( username=data['username'], email=data['email'], password=data['password'], first_name=data.get('first_name'), last_name=data.get('last_name') ) db.session.add(user) db.session.commit() # Create tokens access_token = create_access_token(identity=user.id) refresh_token = create_refresh_token(identity=user.id) return jsonify({ 'message': 'User registered successfully', 'access_token': access_token, 'refresh_token': refresh_token, 'user': user.to_dict() }), 201 except Exception as e: db.session.rollback() return jsonify({ 'error': 'An error occurred during registration' }), 500
@auth_bp.route('/login', methods=['POST']) def login(): try: data = request.get_json() # Validate required fields if not data.get('username') or not data.get('password'): return jsonify({ 'error': 'Username and password are required' }), 400 # Find user by username or email user = User.query.filter( (User.username == data['username']) | (User.email == data['username']) ).first() if not user or not user.check_password(data['password']): return jsonify({ 'error': 'Invalid credentials' }), 401 if not user.is_active: return jsonify({ 'error': 'Account is deactivated' }), 401 # Create tokens access_token = create_access_token(identity=user.id) refresh_token = create_refresh_token(identity=user.id) return jsonify({ 'message': 'Login successful', 'access_token': access_token, 'refresh_token': refresh_token, 'user': user.to_dict() }), 200 except Exception as e: return jsonify({ 'error': 'An error occurred during login' }), 500
@auth_bp.route('/refresh', methods=['POST']) @jwt_required(refresh=True) def refresh(): try: current_user_id = get_jwt_identity() user = User.query.get(current_user_id) if not user or not user.is_active: return jsonify({ 'error': 'User not found or inactive' }), 404 new_access_token = create_access_token(identity=current_user_id) return jsonify({ 'access_token': new_access_token }), 200 except Exception as e: return jsonify({ 'error': 'An error occurred while refreshing token' }), 500
@auth_bp.route('/logout', methods=['POST']) @jwt_required() def logout(): try: jti = get_jwt()['jti'] blacklisted_tokens.add(jti) return jsonify({ 'message': 'Successfully logged out' }), 200 except Exception as e: return jsonify({ 'error': 'An error occurred during logout' }), 500
@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
try:
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
if not user:
return jsonify({
'error': 'User not found'
}), 404
return jsonify({
'user': user.to_dict()
}), 200
except Exception as e:
return jsonify({
'error': 'An error occurred while fetching user data'
}), 500
`
Advanced Authentication Features
Let's enhance our authentication system with additional security features:
Enhanced app/__init__.py with JWT callbacks:
`python
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from config import Config
Initialize extensions
db = SQLAlchemy() jwt = JWTManager()def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
# Initialize extensions with app
db.init_app(app)
jwt.init_app(app)
# JWT error handlers
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
return jsonify({
'error': 'Token has expired',
'message': 'Please refresh your token'
}), 401
@jwt.invalid_token_loader
def invalid_token_callback(error):
return jsonify({
'error': 'Invalid token',
'message': 'Please provide a valid token'
}), 401
@jwt.unauthorized_loader
def missing_token_callback(error):
return jsonify({
'error': 'Authorization token is required',
'message': 'Please provide an access token'
}), 401
@jwt.token_in_blocklist_loader
def check_if_token_revoked(jwt_header, jwt_payload):
from app.routes.auth import blacklisted_tokens
jti = jwt_payload['jti']
return jti in blacklisted_tokens
@jwt.revoked_token_loader
def revoked_token_callback(jwt_header, jwt_payload):
return jsonify({
'error': 'Token has been revoked',
'message': 'Please login again'
}), 401
# Register blueprints
from app.routes.auth import auth_bp
from app.routes.users import users_bp
app.register_blueprint(auth_bp, url_prefix='/api/auth')
app.register_blueprint(users_bp, url_prefix='/api/users')
# Global error handlers
@app.errorhandler(404)
def not_found(error):
return jsonify({
'error': 'Resource not found'
}), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return jsonify({
'error': 'Internal server error'
}), 500
return app
`
Input Validation and Error Handling
Proper validation and error handling are crucial for API security and user experience:
app/utils/validators.py:
`python
import re
from functools import wraps
from flask import request, jsonify
def validate_email(email): """Validate email format""" pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}