Expert REST API Development Guide

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flasgger import Swagger
import yaml
import datetime

app = Flask(__name__)

# Configuration
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'supersecretkey'  # In production, use env variables
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = datetime.timedelta(hours=1)

# Initialize extensions
db = SQLAlchemy(app)
jwt = JWTManager(app)
limiter = Limiter(get_remote_address, app=app, default_limits=["5 per minute"])

# Load Swagger documentation
with open("swagger.yaml", "r") as f:
    swagger_template = yaml.safe_load(f)
Swagger(app, template=swagger_template)

# User model
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password = db.Column(db.String(100), nullable=False)  # In production, use password hashing
    role = db.Column(db.String(20), nullable=False, default='user')
    created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)

# Create database tables before first request
@app.before_first_request
def create_tables():
    db.create_all()
    # Add default users if they don't exist
    if User.query.filter_by(username='admin').first() is None:
        admin = User(username='admin', password='password123', role='admin')
        user = User(username='user', password='password123', role='user')
        db.session.add(admin)
        db.session.add(user)
        db.session.commit()

# Login endpoint
@app.route('/login', methods=['POST'])
@limiter.limit("10 per minute")
def login():
    """
    User login endpoint
    ---
    parameters:
      - name: body
        in: body
        required: true
        schema:
          type: object
          properties:
            username:
              type: string
            password:
              type: string
    responses:
      200:
        description: Login successful, returns JWT token
      401:
        description: Invalid credentials
    """
    data = request.get_json()
    
    if not data or 'username' not in data or 'password' not in data:
        return jsonify({"error": "Missing username or password"}), 400
        
    user = User.query.filter_by(username=data['username'], password=data['password']).first()
    
    if user:
        # Create identity with user information including role
        access_token = create_access_token(identity={
            'username': user.username, 
            'role': user.role,
            'id': user.id
        })
        return jsonify(access_token=access_token)
    
    return jsonify({"error": "Invalid credentials"}), 401

# Protected route that any authenticated user can access
@app.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    """
    Protected endpoint requiring authentication
    ---
    security:
      - JWT: []
    responses:
      200:
        description: Returns user information
      401:
        description: Missing or invalid JWT token
    """
    current_user = get_jwt_identity()
    return jsonify(
        message=f"Welcome {current_user['username']}!",
        role=current_user['role'],
        authenticated=True
    )

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

# Admin-only route
@app.route('/admin', methods=['GET'])
@jwt_required()
def admin():
    """
    Admin-only endpoint
    ---
    security:
      - JWT: []
    responses:
      200:
        description: Returns admin information
      403:
        description: Access denied for non-admin users
      401:
        description: Missing or invalid JWT token
    """
    current_user = get_jwt_identity()
    
    # Check if user has admin role
    if current_user['role'] != 'admin':
        return jsonify({"error": "Access denied. Admin privileges required."}), 403
        
    return jsonify(
        message="Welcome to the admin panel!",
        admin_features=["User Management", "System Configuration", "Analytics"],
        admin_level=current_user['role']
    )

# User management endpoint (admin only)
@app.route('/users', methods=['GET'])
@jwt_required()
def get_users():
    """
    Get all users (admin only)
    ---
    security:
      - JWT: []
    responses:
      200:
        description: Returns list of users
      403:
        description: Access denied for non-admin users
      401:
        description: Missing or invalid JWT token
    """
    current_user = get_jwt_identity()
    
    # Check if user has admin role
    if current_user['role'] != 'admin':
        return jsonify({"error": "Access denied. Admin privileges required."}), 403
    
    users = User.query.all()
    users_list = [
        {
            'id': user.id,
            'username': user.username,
            'role': user.role,
            'created_at': user.created_at.strftime('%Y-%m-%d %H:%M:%S')
        } for user in users
    ]
    
    return jsonify(users=users_list)

# Rate-limited endpoint example
@app.route('/rate-limited', methods=['GET'])
@limiter.limit("3 per minute")
def rate_limited_route():
    """
    Rate limited endpoint demonstration
    ---
    responses:
      200:
        description: Successful request
      429:
        description: Too many requests
    """
    return jsonify(
        message="This endpoint is rate-limited to 3 requests per minute.",
        tip="Try refreshing multiple times to see the rate limiting in action."
    )

# Apply different rate limits based on user role
@app.route('/api/data', methods=['GET'])
@jwt_required()
def get_data():
    """
    Data endpoint with role-based rate limiting
    ---
    security:
      - JWT: []
    responses:
      200:
        description: Returns sample data
      429:
        description: Too many requests
      401:
        description: Missing or invalid JWT token
    """
    current_user = get_jwt_identity()
    
    # Define custom rate limits based on user role
    if current_user['role'] == 'admin':
        limiter.limit("100 per minute")(lambda: None)()
    else:
        limiter.limit("10 per minute")(lambda: None)()
    
    return jsonify(
        data={"sample": "data", "items": [1, 2, 3, 4, 5]},
        rate_limit_info=f"Your role ({current_user['role']}) determines your rate limit."
    )

# Add indexing to frequently queried fields
class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False, index=True)  # Added index
    # ... other fields

# Add refresh token support
@app.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    identity = get_jwt_identity()
    access_token = create_access_token(identity=identity)
    return jsonify(access_token=access_token)

from flask_caching import Cache

cache = Cache(app, config={'CACHE_TYPE': 'simple'})

@app.route('/cached-data')
@cache.cached(timeout=60)  # Cache for 60 seconds
def get_cached_data():
    # Expensive operation here
    return jsonify(data="This response is cached for 60 seconds")