Web Application Frameworks

9.3. Web Application Frameworks#

For maximum control over your application, traditional web frameworks provide the flexibility to build custom REST APIs and full-stack applications. This section covers Flask and FastAPI, the two most popular Python web frameworks for ML deployment.

9.3.1. Why Web Frameworks?#

Full control over application structure
API-first design for client-server architecture
Production-ready with proper authentication, logging, etc.
Language-agnostic clients (JavaScript, mobile apps, etc.)
Microservices architectures

9.3.2. Flask#

9.3.2.1. Overview#

Flask is a lightweight, flexible web framework that’s been the Python standard for over a decade.

Key Features:

Minimalist core with extensions
Easy to learn
Mature ecosystem
Synchronous by default

9.3.2.2. Installation#

pip install flask flask-cors

9.3.2.3. Basic Flask App#

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/')
def home():
    return "ML Model API"

@app.route('/predict', methods=['POST'])
def predict():
    data = request.get_json()
    # Process data
    result = {"prediction": 42}
    return jsonify(result)

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

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
Cell In[1], line 1
----> 1 from flask import Flask, jsonify, request
      3 app = Flask(__name__)
      5 @app.route('/')
      6 def home():

ModuleNotFoundError: No module named 'flask'

9.3.2.4. ML Model API with Flask#

app.py:

from flask import Flask, request, jsonify
from flask_cors import CORS
import joblib
import numpy as np
import pandas as pd

app = Flask(__name__)
CORS(app)  # Enable CORS

# Load model at startup
model = joblib.load('models/churn_model.joblib')
scaler = joblib.load('models/scaler.joblib')

@app.route('/health', methods=['GET'])
def health():
    """Health check endpoint"""
    return jsonify({
        'status': 'healthy',
        'model': 'churn_prediction_v1'
    })

@app.route('/predict', methods=['POST'])
def predict():
    """
    Predict customer churn
    
    Expected JSON:
    {
        "age": 35,
        "income": 50000,
        "credit_score": 650
    }
    """
    try:
        # Parse input
        data = request.get_json()
        
        # Validate
        required_fields = ['age', 'income', 'credit_score']
        if not all(field in data for field in required_fields):
            return jsonify({
                'error': 'Missing required fields',
                'required': required_fields
            }), 400
        
        # Create features
        features = pd.DataFrame([data])
        features_scaled = scaler.transform(features)
        
        # Predict
        prediction = model.predict(features_scaled)[0]
        probability = model.predict_proba(features_scaled)[0]
        
        # Return response
        return jsonify({
            'prediction': int(prediction),
            'probability': {
                'retain': float(probability[0]),
                'churn': float(probability[1])
            },
            'input': data
        })
        
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/batch_predict', methods=['POST'])
def batch_predict():
    """
    Batch prediction endpoint
    
    Expected JSON:
    {
        "customers": [
            {"age": 35, "income": 50000, "credit_score": 650},
            {"age": 42, "income": 75000, "credit_score": 720}
        ]
    }
    """
    try:
        data = request.get_json()
        customers = data['customers']
        
        # Create DataFrame
        df = pd.DataFrame(customers)
        df_scaled = scaler.transform(df)
        
        # Predict
        predictions = model.predict(df_scaled)
        probabilities = model.predict_proba(df_scaled)
        
        # Format results
        results = []
        for i, customer in enumerate(customers):
            results.append({
                'input': customer,
                'prediction': int(predictions[i]),
                'probability': {
                    'retain': float(probabilities[i][0]),
                    'churn': float(probabilities[i][1])
                }
            })
        
        return jsonify({'results': results})
        
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

9.3.2.5. Testing Flask API#

Using curl:

curl -X POST http://localhost:5000/predict \\
  -H "Content-Type: application/json" \\
  -d '{"age": 35, "income": 50000, "credit_score": 650}'

Using Python requests:

import requests

url = 'http://localhost:5000/predict'
data = {
    'age': 35,
    'income': 50000,
    'credit_score': 650
}

response = requests.post(url, json=data)
print(response.json())

9.3.3. FastAPI#

9.3.3.1. Overview#

FastAPI is a modern, high-performance framework built on Python 3.6+ type hints.

Key Features:

Fast: Based on Starlette and Pydantic
Async support: Native async/await
Auto documentation: Interactive API docs (Swagger UI)
Type validation: Automatic data validation
Modern Python: Leverages type hints

9.3.3.2. Installation#

pip install fastapi uvicorn[standard]

9.3.3.3. Basic FastAPI App#

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "ML Model API"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

Run:

uvicorn main:app --reload

Automatic docs at: http://localhost:8000/docs

9.3.3.4. ML Model API with FastAPI#

main.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import joblib
import numpy as np
import pandas as pd
from typing import List

app = FastAPI(
    title="Customer Churn Prediction API",
    description="Predict customer churn using ML",
    version="1.0.0"
)

# Load models at startup
model = joblib.load('models/churn_model.joblib')
scaler = joblib.load('models/scaler.joblib')

# Define request/response schemas
class Customer(BaseModel):
    age: int = Field(..., ge=18, le=100, description="Customer age")
    income: float = Field(..., gt=0, description="Annual income")
    credit_score: int = Field(..., ge=300, le=850, description="Credit score")
    
    class Config:
        schema_extra = {
            "example": {
                "age": 35,
                "income": 50000,
                "credit_score": 650
            }
        }

class PredictionResponse(BaseModel):
    prediction: int
    probability: dict
    input: dict

class BatchRequest(BaseModel):
    customers: List[Customer]

class BatchResponse(BaseModel):
    results: List[PredictionResponse]

# Endpoints
@app.get("/")
def root():
    return {
        "message": "Customer Churn Prediction API",
        "docs": "/docs"
    }

@app.get("/health")
def health():
    return {
        "status": "healthy",
        "model": "churn_prediction_v1"
    }

@app.post("/predict", response_model=PredictionResponse)
def predict(customer: Customer):
    """
    Predict churn for a single customer
    """
    try:
        # Create features
        features = pd.DataFrame([customer.dict()])
        features_scaled = scaler.transform(features)
        
        # Predict
        prediction = model.predict(features_scaled)[0]
        probability = model.predict_proba(features_scaled)[0]
        
        return PredictionResponse(
            prediction=int(prediction),
            probability={
                "retain": float(probability[0]),
                "churn": float(probability[1])
            },
            input=customer.dict()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/batch_predict", response_model=BatchResponse)
def batch_predict(batch: BatchRequest):
    """
    Batch prediction for multiple customers
    """
    try:
        # Create DataFrame
        df = pd.DataFrame([c.dict() for c in batch.customers])
        df_scaled = scaler.transform(df)
        
        # Predict
        predictions = model.predict(df_scaled)
        probabilities = model.predict_proba(df_scaled)
        
        # Format results
        results = []
        for i, customer in enumerate(batch.customers):
            results.append(PredictionResponse(
                prediction=int(predictions[i]),
                probability={
                    "retain": float(probabilities[i][0]),
                    "churn": float(probabilities[i][1])
                },
                input=customer.dict()
            ))
        
        return BatchResponse(results=results)
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Run with: uvicorn main:app --host 0.0.0.0 --port 8000

9.3.3.5. Testing FastAPI#

Interactive Docs: Visit http://localhost:8000/docs - test directly in browser!

Using curl:

curl -X POST http://localhost:8000/predict \\
  -H "Content-Type: application/json" \\
  -d '{"age": 35, "income": 50000, "credit_score": 650}'

9.3.4. Client-Server Architecture#

9.3.4.1. Frontend with JavaScript#

index.html:

<!DOCTYPE html>
<html>
<head>
    <title>Churn Prediction</title>
</head>
<body>
    <h1>Customer Churn Prediction</h1>
    
    <form id="predictionForm">
        <label>Age: <input type="number" id="age" required></label><br>
        <label>Income: <input type="number" id="income" required></label><br>
        <label>Credit Score: <input type="number" id="credit_score" required></label><br>
        <button type="submit">Predict</button>
    </form>
    
    <div id="result"></div>
    
    <script>
        document.getElementById('predictionForm').addEventListener('submit', async (e) => {
            e.preventDefault();
            
            const data = {
                age: parseInt(document.getElementById('age').value),
                income: parseFloat(document.getElementById('income').value),
                credit_score: parseInt(document.getElementById('credit_score').value)
            };
            
            const response = await fetch('http://localhost:8000/predict', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify(data)
            });
            
            const result = await response.json();
            
            document.getElementById('result').innerHTML = `
                <h2>Result:</h2>
                <p>Prediction: ${result.prediction === 1 ? 'CHURN' : 'RETAIN'}</p>
                <p>Churn Probability: ${(result.probability.churn * 100).toFixed(1)}%</p>
            `;
        });
    </script>
</body>
</html>

9.3.5. Deployment#

9.3.5.1. Docker#

Dockerfile (FastAPI):

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

9.3.5.2. Production Considerations#

Use production WSGI/ASGI server:
- Flask: Gunicorn
- FastAPI: Uvicorn with Gunicorn workers
Environment variables for config:

import os

MODEL_PATH = os.getenv('MODEL_PATH', 'models/model.joblib')
PORT = int(os.getenv('PORT', 8000))

Logging:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

logger.info(f"Model loaded from {MODEL_PATH}")

Error handling and validation
Authentication (API keys, OAuth)
Rate limiting
Monitoring (Prometheus, Grafana)

9.3.6. Flask vs FastAPI#

Feature	Flask	FastAPI
Performance	Good	Excellent (async)
Learning curve	Gentle	Moderate
Type validation	Manual	Automatic
Documentation	Manual	Auto-generated
Async support	Extensions needed	Native
Community	Large, mature	Growing fast
Best for	Traditional apps	Modern APIs

9.3.7. Summary#

Web frameworks provide:

Full control over application structure
API-first design for any client
Production-ready features
Scalability for microservices

Choose Flask for:

Traditional web applications
Extensive extension ecosystem
Mature, stable projects

Choose FastAPI for:

Modern API development
High performance requirements
Automatic documentation
Type-safe development