RESTful APIs with Flask

• What to learn: Creating a REST API that returns JSON instead of HTML. Using modules like flask-restful or building it manually with jsonify. Handling API endpoints for your models.

• Django Connection: This knowledge is directly transferable to Django REST Framework (DRF), the most popular toolkit for building APIs in Django. Understanding the principles of REST in a simpler context is invaluable.

1. Setup and Installation

bash

pip install flask flask-sqlalchemy flask-restful flask-marshmallow marshmallow-sqlalchemy

2. Manual API Implementation (Understanding the Basics)

First, let's build an API manually to understand the fundamentals:

python

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///api.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# Model
class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    quantity = db.Column(db.Integer, nullable=False, default=0)

# Create tables
with app.app_context():
    db.create_all()
    # Add sample data
    if not Product.query.first():
        sample_products = [
            Product(name="Laptop", price=999.99, quantity=10),
            Product(name="Mouse", price=29.99, quantity=50),
            Product(name="Keyboard", price=79.99, quantity=25)
        ]
        db.session.add_all(sample_products)
        db.session.commit()

# Manual API Routes
@app.route('/api/products', methods=['GET'])
def get_products():
    products = Product.query.all()
    return jsonify([{
        'id': product.id,
        'name': product.name,
        'price': product.price,
        'quantity': product.quantity
    } for product in products])

@app.route('/api/products/<int:product_id>', methods=['GET'])
def get_product(product_id):
    product = Product.query.get_or_404(product_id)
    return jsonify({
        'id': product.id,
        'name': product.name,
        'price': product.price,
        'quantity': product.quantity
    })

@app.route('/api/products', methods=['POST'])
def create_product():
    data = request.get_json()
    
    # Basic validation
    if not data or not all(key in data for key in ['name', 'price']):
        return jsonify({'error': 'Missing required fields'}), 400
    
    product = Product(
        name=data['name'],
        price=data['price'],
        quantity=data.get('quantity', 0)
    )
    
    db.session.add(product)
    db.session.commit()
    
    return jsonify({
        'id': product.id,
        'name': product.name,
        'price': product.price,
        'quantity': product.quantity
    }), 201

@app.route('/api/products/<int:product_id>', methods=['PUT'])
def update_product(product_id):
    product = Product.query.get_or_404(product_id)
    data = request.get_json()
    
    if 'name' in data:
        product.name = data['name']
    if 'price' in data:
        product.price = data['price']
    if 'quantity' in data:
        product.quantity = data['quantity']
    
    db.session.commit()
    
    return jsonify({
        'id': product.id,
        'name': product.name,
        'price': product.price,
        'quantity': product.quantity
    })

@app.route('/api/products/<int:product_id>', methods=['DELETE'])
def delete_product(product_id):
    product = Product.query.get_or_404(product_id)
    db.session.delete(product)
    db.session.commit()
    
    return jsonify({'message': 'Product deleted successfully'}), 200

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

3. Using Flask-RESTful for Structured APIs

Now let's use Flask-RESTful for a more organized approach:

python

from flask import Flask
from flask_restful import Api, Resource, reqparse, abort, fields, marshal_with
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///restful.db'
db = SQLAlchemy(app)

# Model
class Book(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    author = db.Column(db.String(100), nullable=False)
    year = db.Column(db.Integer, nullable=False)

# Request parser for input validation
book_parser = reqparse.RequestParser()
book_parser.add_argument('title', type=str, required=True, help='Title is required')
book_parser.add_argument('author', type=str, required=True, help='Author is required')
book_parser.add_argument('year', type=int, required=True, help='Year is required')

# Response formatting
book_fields = {
    'id': fields.Integer,
    'title': fields.String,
    'author': fields.String,
    'year': fields.Integer
}

# Resource classes
class BookListResource(Resource):
    @marshal_with(book_fields)
    def get(self):
        books = Book.query.all()
        return books
    
    @marshal_with(book_fields)
    def post(self):
        args = book_parser.parse_args()
        
        # Check if book already exists
        existing_book = Book.query.filter_by(title=args['title'], author=args['author']).first()
        if existing_book:
            abort(409, message="Book already exists")
        
        book = Book(
            title=args['title'],
            author=args['author'],
            year=args['year']
        )
        
        db.session.add(book)
        db.session.commit()
        return book, 201

class BookResource(Resource):
    @marshal_with(book_fields)
    def get(self, book_id):
        book = Book.query.get_or_404(book_id)
        return book
    
    @marshal_with(book_fields)
    def put(self, book_id):
        book = Book.query.get_or_404(book_id)
        args = book_parser.parse_args()
        
        book.title = args['title']
        book.author = args['author']
        book.year = args['year']
        
        db.session.commit()
        return book
    
    def delete(self, book_id):
        book = Book.query.get_or_404(book_id)
        db.session.delete(book)
        db.session.commit()
        return {'message': 'Book deleted successfully'}, 200

# Register routes
api.add_resource(BookListResource, '/api/books')
api.add_resource(BookResource, '/api/books/<int:book_id>')

# Create tables
with app.app_context():
    db.create_all()

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

4. Using Marshmallow for Serialization/Deserialization

For more advanced serialization:

python

from flask import Flask, request
from flask_restful import Api, Resource
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow import ValidationError

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///marshmallow.db'
db = SQLAlchemy(app)
ma = Marshmallow(app)

# Model
class User(db.Model):
    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)
    created_at = db.Column(db.DateTime, server_default=db.func.now())

# Schema
class UserSchema(ma.SQLAlchemySchema):
    class Meta:
        model = User
    
    id = ma.auto_field()
    username = ma.auto_field()
    email = ma.auto_field()
    created_at = ma.auto_field()

user_schema = UserSchema()
users_schema = UserSchema(many=True)

# Validation Schema
class UserValidationSchema(ma.Schema):
    username = ma.Str(required=True, validate=ma.validate.Length(min=3, max=80))
    email = ma.Email(required=True)

user_validation_schema = UserValidationSchema()

# Resources
class UserListResource(Resource):
    def get(self):
        users = User.query.all()
        return users_schema.dump(users)
    
    def post(self):
        try:
            # Validate and deserialize input
            data = user_validation_schema.load(request.json)
            
            # Check for existing user
            if User.query.filter_by(username=data['username']).first():
                return {'error': 'Username already exists'}, 400
            
            if User.query.filter_by(email=data['email']).first():
                return {'error': 'Email already exists'}, 400
            
            user = User(**data)
            db.session.add(user)
            db.session.commit()
            
            return user_schema.dump(user), 201
        
        except ValidationError as err:
            return {'errors': err.messages}, 400

class UserResource(Resource):
    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return user_schema.dump(user)
    
    def put(self, user_id):
        user = User.query.get_or_404(user_id)
        
        try:
            data = user_validation_schema.load(request.json)
            
            # Check for duplicates (excluding current user)
            if User.query.filter(User.username == data['username'], User.id != user_id).first():
                return {'error': 'Username already exists'}, 400
            
            if User.query.filter(User.email == data['email'], User.id != user_id).first():
                return {'error': 'Email already exists'}, 400
            
            user.username = data['username']
            user.email = data['email']
            db.session.commit()
            
            return user_schema.dump(user)
        
        except ValidationError as err:
            return {'errors': err.messages}, 400
    
    def delete(self, user_id):
        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()
        return {'message': 'User deleted successfully'}, 200

api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')

with app.app_context():
    db.create_all()

5. Complete Blog API Example

python

from flask import Flask
from flask_restful import Api, Resource, reqparse
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from datetime import datetime

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog_api.db'
db = SQLAlchemy(app)
ma = Marshmallow(app)

# Models
class User(db.Model):
    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)
    posts = db.relationship('Post', backref='author', lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

# Schemas
class UserSchema(ma.SQLAlchemySchema):
    class Meta:
        model = User
    
    id = ma.auto_field()
    username = ma.auto_field()
    email = ma.auto_field()
    posts = ma.auto_field()

class PostSchema(ma.SQLAlchemySchema):
    class Meta:
        model = Post
    
    id = ma.auto_field()
    title = ma.auto_field()
    content = ma.auto_field()
    created_at = ma.auto_field()
    user_id = ma.auto_field()
    author = ma.Nested(UserSchema(only=('id', 'username')))

user_schema = UserSchema()
users_schema = UserSchema(many=True)
post_schema = PostSchema()
posts_schema = PostSchema(many=True)

# Request parsers
post_parser = reqparse.RequestParser()
post_parser.add_argument('title', type=str, required=True)
post_parser.add_argument('content', type=str, required=True)
post_parser.add_argument('user_id', type=int, required=True)

# Resources
class PostListResource(Resource):
    def get(self):
        posts = Post.query.all()
        return posts_schema.dump(posts)
    
    def post(self):
        args = post_parser.parse_args()
        
        # Verify user exists
        user = User.query.get(args['user_id'])
        if not user:
            return {'error': 'User not found'}, 404
        
        post = Post(
            title=args['title'],
            content=args['content'],
            user_id=args['user_id']
        )
        
        db.session.add(post)
        db.session.commit()
        return post_schema.dump(post), 201

class PostResource(Resource):
    def get(self, post_id):
        post = Post.query.get_or_404(post_id)
        return post_schema.dump(post)
    
    def put(self, post_id):
        post = Post.query.get_or_404(post_id)
        args = post_parser.parse_args()
        
        post.title = args['title']
        post.content = args['content']
        post.user_id = args['user_id']
        
        db.session.commit()
        return post_schema.dump(post)
    
    def delete(self, post_id):
        post = Post.query.get_or_404(post_id)
        db.session.delete(post)
        db.session.commit()
        return {'message': 'Post deleted'}, 200

# Register routes
api.add_resource(PostListResource, '/api/posts')
api.add_resource(PostResource, '/api/posts/<int:post_id>')

# Initialize database
with app.app_context():
    db.create_all()
    # Add sample data
    if not User.query.first():
        user1 = User(username='john_doe', email='john@example.com')
        user2 = User(username='jane_smith', email='jane@example.com')
        db.session.add_all([user1, user2])
        db.session.commit()

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

6. Testing Your API

Use tools like curl, Postman, or httpie to test your API:

bash

# Get all posts
curl http://localhost:5000/api/posts

# Create a new post
curl -X POST http://localhost:5000/api/posts \
  -H "Content-Type: application/json" \
  -d '{"title": "New Post", "content": "Post content", "user_id": 1}'

# Update a post
curl -X PUT http://localhost:5000/api/posts/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "Updated Title", "content": "Updated content", "user_id": 1}'

# Delete a post
curl -X DELETE http://localhost:5000/api/posts/1

7. Django Connection - What You're Learning for Django REST Framework

Flask Concept → Django REST Framework Equivalent:

• Manual JSON routes → DRF APIViews and ViewSets

• Flask-RESTful Resources → DRF ViewSets

• Marshmallow Schemas → DRF Serializers

• reqparse → DRF Serializer validation

• Manual CRUD operations → DRF ModelViewSet automatic CRUD

• Manual error handling → DRF built-in exception handling

Key REST Concepts You're Mastering:

1. HTTP Methods (GET, POST, PUT, DELETE) for different operations

2. Resource-based URL design (/api/books, /api/books/1)

3. JSON serialization/deserialization

4. Input validation and error handling

5. Status codes (200, 201, 400, 404, etc.)

6. API structure and organization

This foundation in Flask APIs will make Django REST Framework feel very intuitive. You'll recognize the same REST principles but with Django's more structured and feature-rich approach!