Building REST APIs is a fundamental skill for any backend developer. In this comprehensive guide, we'll walk through creating a production-ready API using Flask, one of Python's most popular microframeworks. We'll cover everything from project setup to deployment, with a focus on security, scalability, and maintainability.
Prerequisites
Before we dive in, make sure you have the following installed:
- Python 3.9 or higher
- pip (Python package manager)
- A code editor (VS Code recommended)
- Basic understanding of HTTP and REST principles
Project Setup
Let's start by creating a new project directory and setting up a virtual environment. This keeps our dependencies isolated and makes the project more portable.
# Create project directory
mkdir flask-api && cd flask-api
# Create virtual environment
python -m venv venv
# Activate virtual environment
# On macOS/Linux:
source venv/bin/activate
# On Windows:
venv\Scripts\activate
# Install dependencies
pip install flask flask-restful flask-jwt-extended flask-sqlalchemy marshmallow
Application Structure
A well-organized project structure is crucial for maintainability. Here's the structure we'll be using throughout this tutorial:
flask-api/
├── app/
│ ├── __init__.py # Application factory
│ ├── config.py # Configuration settings
│ ├── models/ # Database models
│ ├── routes/ # API endpoints
│ ├── schemas/ # Request/Response schemas
│ ├── services/ # Business logic
│ └── utils/ # Helper functions
├── tests/
├── .env
├── requirements.txt
└── run.py
Creating the App Factory
The application factory pattern is a best practice in Flask development. It allows us to create multiple instances of the app with different configurations, which is especially useful for testing.
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from flask_restful import Api
db = SQLAlchemy()
jwt = JWTManager()
def create_app(config_class='app.config.Config'):
"""Application factory pattern."""
app = Flask(__name__)
app.config.from_object(config_class)
# Initialize extensions
db.init_app(app)
jwt.init_app(app)
# Create API instance
api = Api(app)
# Register blueprints
from app.routes import auth, users, posts
app.register_blueprint(auth.bp)
app.register_blueprint(users.bp)
app.register_blueprint(posts.bp)
# Create database tables
with app.app_context():
db.create_all()
return app
Configuration
Keep your configuration in a separate file and use environment variables for sensitive data. Never commit secrets to version control!
# app/config.py
import os
from datetime import timedelta
class Config:
"""Base configuration."""
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
# JWT Settings
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
TESTING = False
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
Creating Models
Let's create a User model for authentication and a Post model for our blog API. We'll use SQLAlchemy's ORM to define our database schema.
# app/models/user.py
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from app import db
class User(db.Model):
"""User model for authentication."""
__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(256), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
is_active = db.Column(db.Boolean, default=True)
# Relationships
posts = db.relationship('Post', backref='author', lazy='dynamic')
def set_password(self, password):
"""Hash and set the user's password."""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Verify the password against the stored hash."""
return check_password_hash(self.password_hash, password)
def to_dict(self):
"""Serialize user to dictionary."""
return {
'id': self.id,
'username': self.username,
'email': self.email,
'created_at': self.created_at.isoformat(),
'is_active': self.is_active
}
Authentication with JWT
JWT (JSON Web Tokens) is a standard way to handle authentication in modern APIs. Here's how to implement login and registration endpoints with proper security measures.
# app/routes/auth.py
from flask import Blueprint, request, jsonify
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity
)
from app import db
from app.models.user import User
from app.schemas.user import UserSchema, LoginSchema
bp = Blueprint('auth', __name__, url_prefix='/api/auth')
@bp.route('/register', methods=['POST'])
def register():
"""Register a new user."""
data = request.get_json()
# Validate input
schema = UserSchema()
errors = schema.validate(data)
if errors:
return jsonify({'errors': errors}), 400
# Check if user exists
if User.query.filter_by(email=data['email']).first():
return jsonify({'error': 'Email already registered'}), 409
if User.query.filter_by(username=data['username']).first():
return jsonify({'error': 'Username already taken'}), 409
# Create user
user = User(
username=data['username'],
email=data['email']
)
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
# Generate tokens
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
'message': 'User created successfully',
'user': user.to_dict(),
'access_token': access_token,
'refresh_token': refresh_token
}), 201
@bp.route('/login', methods=['POST'])
def login():
"""Authenticate user and return tokens."""
data = request.get_json()
user = User.query.filter_by(email=data.get('email')).first()
if not user or not user.check_password(data.get('password', '')):
return jsonify({'error': 'Invalid email or password'}), 401
if not user.is_active:
return jsonify({'error': 'Account is deactivated'}), 403
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
'user': user.to_dict(),
'access_token': access_token,
'refresh_token': refresh_token
}), 200
@bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
"""Refresh access token."""
current_user_id = get_jwt_identity()
access_token = create_access_token(identity=current_user_id)
return jsonify({'access_token': access_token}), 200
Error Handling
Proper error handling is crucial for a production API. Users should receive clear, consistent error messages without exposing internal details.
# app/utils/errors.py
from flask import jsonify
class APIError(Exception):
"""Base API exception."""
status_code = 500
def __init__(self, message, status_code=None, payload=None):
super().__init__()
self.message = message
if status_code is not None:
self.status_code = status_code
self.payload = payload
def to_dict(self):
rv = dict(self.payload or ())
rv['error'] = self.message
rv['status_code'] = self.status_code
return rv
class NotFoundError(APIError):
status_code = 404
class ValidationError(APIError):
status_code = 400
class UnauthorizedError(APIError):
status_code = 401
def register_error_handlers(app):
"""Register error handlers with the Flask app."""
@app.errorhandler(APIError)
def handle_api_error(error):
response = jsonify(error.to_dict())
response.status_code = error.status_code
return response
@app.errorhandler(404)
def handle_not_found(error):
return jsonify({'error': 'Resource not found'}), 404
@app.errorhandler(500)
def handle_internal_error(error):
return jsonify({'error': 'Internal server error'}), 500
Next Steps
We've covered the fundamentals of building a production-ready Flask API. In future articles, we'll explore:
- Rate limiting and request throttling
- API versioning strategies
- Caching with Redis
- Background tasks with Celery
- Deployment to cloud platforms
The complete source code for this tutorial is available on GitHub. Feel free to star the repo and submit issues or pull requests!
If you found this tutorial helpful, consider sharing it with other developers who might benefit from it. Happy coding!