🛵 QuickBite ETA#

This demo runs a single FastAPI container holding a pre-trained scikit-learn model. The browser sends a POST request to the flask proxy, which forwards it to the quickbite container. The container extracts features, runs model.predict(), and returns the ETA in minutes. There are no external API calls and no API keys required.

The QuickBite ETA demo demonstrates how to reliably deploy a classical machine learning model into production using Containerization. When a data scientist trains a model locally, it depends on specific library versions (like scikit-learn or pandas) and Python configurations. Deploying this model directly to a server often results in the "it works on my machine" problem due to environment mismatches.

This demo solves that by wrapping the entire prediction pipeline into a single, immutable Docker container. When you submit an order via the browser, here is how the architecture handles it:

• Containerized Environment: The Docker image bundles the exact versions of Python, FastAPI, and scikit-learn needed. This guarantees that whether the QuickBite container runs on a developer's laptop or a cloud server, the environment is perfectly consistent.
• FastAPI Serving Layer: The container runs a high-performance web server that exposes a /predict endpoint. It translates the incoming JSON order details (like distance, prep time, and weather) into a Pandas DataFrame that the model can understand.
• The Inference Engine: At startup, the container loads a pre-trained scikit-learn model (eta_model.pkl) into memory. When the FastAPI route receives your data, it runs model.predict() to instantly calculate the estimated delivery time in minutes and returns it to your browser.

QuickBite ETA - Model Training Trains a RandomForest on synthetic food-delivery data and saves it as eta_model.pkl

"""
QuickBite ETA - Model Training
Trains a RandomForest on synthetic food-delivery data
and saves it as eta_model.pkl
"""
import pandas as pd
import numpy as np
import joblib
from sklearn.ensemble import RandomForestRegressor

# ① make the synthetic dataset reproducible and choose its size
np.random.seed(42)
n = 5000

# ② generate fake delivery features the model can learn from
df = pd.DataFrame({
    "distance_km": np.random.uniform(0.5, 12, n),
    "prep_time_min": np.random.uniform(5, 30, n),
    "rider_available": np.random.randint(0, 2, n),
    "is_raining": np.random.randint(0, 2, n),
})

# ③ calculate ETA from distance, prep, rain, rider supply, and noise
# ETA = base + distance*3 + prep + rain penalty + rider penalty + noise
df["eta_min"] = (
    8
    + df.distance_km * 3
    + df.prep_time_min * 0.7
    + df.is_raining * 9
    + (1 - df.rider_available) * 6
    + np.random.normal(0, 2, n)
)

# ④ split features from target and train the RandomForest model
X, y = df.drop(columns=["eta_min"]), df["eta_min"]
model = RandomForestRegressor(n_estimators=60, random_state=42).fit(X, y)

# ⑤ save the trained model so the FastAPI app can load it
joblib.dump(model, "eta_model.pkl")
print("Model saved: eta_model.pkl ✅")

QuickBite ETA - FastAPI Serving POST /predict with order details -> returns ETA in minutes

"""
QuickBite ETA - FastAPI Serving
POST /predict with order details -> returns ETA in minutes
"""
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import pandas as pd

app = FastAPI(title="QuickBite ETA")

# ① load the trained model once at startup so predictions are fast
model = joblib.load("eta_model.pkl")  # loaded once at startup, reused for every request


class Order(BaseModel):
    distance_km: float
    prep_time_min: float
    rider_available: int
    is_raining: int


@app.get("/")
def health():
    return {"status": "QuickBite ETA is live 🛵"}


@app.post("/predict")
def predict(order: Order):
    # ① sklearn expects tabular input; wrap the single order in a one-row
    # DataFrame whose column names must match the Order fields exactly.
    X = pd.DataFrame([order.model_dump()])

    # ② run the model and round the ETA for a friendly response
    eta = round(float(model.predict(X)[0]), 1)

    # ③ return the numeric ETA plus the message the UI shows
    return {"eta_minutes": eta, "message": f"Your food arrives in {eta} min 🍔"}

Flask server proxying browser requests to internal Docker demo services.

"""Flask server proxying browser requests to internal Docker demo services.

Architecture notes
------------------
- All routes are attached to a Blueprint (``bp``) instead of directly to
  ``app``. This lets us register the entire Blueprint under a runtime URL
  prefix (``PATH_PREFIX``) without touching individual route strings.
- In local development PATH_PREFIX is empty, so routes are at "/",
  "/quickbite/predict", etc. In production Nginx forwards ``/docker/...``
  traffic to the container and PATH_PREFIX is set to "/docker".
- flask-cors adds ``Access-Control-Allow-Origin: *`` headers so the HTML
  page can call the API even if it is served from a different origin during
  development.
- Proxy routes forward browser requests to internal Docker services
  (quickbite, scalergpt, deskbuddy-agent) using service-name networking.
"""

import os
from pathlib import Path

import requests as http_client
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

# PATH_PREFIX is set by the deployment environment (e.g. "/docker") so the app
# works correctly behind an Nginx location block. Locally it is an empty
# string, which mounts all routes at the root.
PATH_PREFIX = os.environ.get("PATH_PREFIX", "")

# app.py lives in src/python, while index.html, css/, and js/ live in src/.
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))

# Allow cross-origin requests from any origin. In production you would
# restrict this to the specific front-end domain.
CORS(app)

# A Blueprint groups related routes. We register it once at the bottom with
# the runtime PATH_PREFIX, avoiding any hardcoded path strings in the routes.
bp = Blueprint("main", __name__)


@bp.before_request
def enforce_rate_limit():
    """Enforce strict 10 requests per hour limit on all POST endpoints."""
    # ① only rate-limit write requests so page assets stay fast
    if request.method == "POST":
        # ② check the caller's hourly quota before proxying work
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=10, window_seconds=3600
        )
        # ③ return a 429 with Retry-After when the quota is used up
        if blocked:
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


# Internal service URLs — these use Docker Compose service names, never IPs.
QUICKBITE_URL = "http://quickbite:8000"
SCALERGPT_URL = "http://scalergpt:8000"
DESKBUDDY_URL = "http://deskbuddy-agent:9000"

# Timeout for proxy requests to example services (seconds).
PROXY_TIMEOUT = 30


# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------

def proxy_request(method, url, json_body=None):
    """Forward a request to an internal service and return its JSON response.

    Returns a tuple of (response_dict, http_status_code). On connection
    errors, returns a helpful error message instead of crashing.
    """
    # ① forward the request to the selected internal service
    try:
        if method == "GET":
            resp = http_client.get(url, timeout=PROXY_TIMEOUT)
        else:
            resp = http_client.post(url, json=json_body, timeout=PROXY_TIMEOUT)
        # ② pass through the service JSON and HTTP status code
        return resp.json(), resp.status_code
    except http_client.ConnectionError:
        # ③ turn connection failures into a helpful service-start message
        service = url.split("//")[1].split(":")[0]
        return {
            "error": f"Service '{service}' is not running. "
                     f"Start it with: docker compose up {service}"
        }, 503
    except Exception as e:
        # ④ return unexpected proxy failures as JSON instead of crashing
        return {"error": str(e)}, 500


# ---------------------------------------------------------------------------
# Routes — Static files
# ---------------------------------------------------------------------------


@bp.route("/")
def index():
    """Serve index.html, injecting the correct API base URL for the environment."""
    # ① read the static homepage template from the shared src folder
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    # ② inject the runtime API prefix so browser calls hit this gateway
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ③ return the rendered HTML with the correct MIME type
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    """Serve stylesheets from the src/css directory."""
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    """Serve scripts from the src/js directory."""
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/info/<path:filename>")
def info(filename):
    """Serve the "how this demo works" explainer pages from src/info."""
    return app.send_static_file(os.path.join("info", filename))


# ---------------------------------------------------------------------------
# Routes — QuickBite ETA (Level 1, keyless)
# ---------------------------------------------------------------------------


@bp.route("/quickbite/predict", methods=["POST"])
def quickbite_predict():
    """Proxy ETA prediction to the QuickBite FastAPI service."""
    # ① parse the browser's order JSON
    body = request.get_json(force=True)
    # ② proxy the order to the QuickBite prediction service
    data, status = proxy_request("POST", f"{QUICKBITE_URL}/predict", body)
    # ③ return the service JSON and status code unchanged
    return jsonify(data), status


@bp.route("/quickbite/status")
def quickbite_status():
    """Check if QuickBite service is running."""
    # ① ask QuickBite for its health payload
    data, status = proxy_request("GET", f"{QUICKBITE_URL}/")
    # ② return the health JSON and status code unchanged
    return jsonify(data), status


# ---------------------------------------------------------------------------
# Routes — ScalerGPT (Level 2, needs OPENAI_API_KEY)
# ---------------------------------------------------------------------------


@bp.route("/scalergpt/ask", methods=["POST"])
def scalergpt_ask():
    """Proxy RAG question to the ScalerGPT FastAPI service."""
    # ① parse the browser's question JSON
    body = request.get_json(force=True)
    # ② proxy the question to the ScalerGPT RAG service
    data, status = proxy_request("POST", f"{SCALERGPT_URL}/ask", body)
    # ③ return the service JSON and status code unchanged
    return jsonify(data), status


@bp.route("/scalergpt/status")
def scalergpt_status():
    """Check if ScalerGPT service is running and how many docs are indexed."""
    # ① ask ScalerGPT for its health and index summary
    data, status = proxy_request("GET", f"{SCALERGPT_URL}/")
    # ② return the health JSON and status code unchanged
    return jsonify(data), status


# ---------------------------------------------------------------------------
# Routes — DeskBuddy (Level 3, needs OPENAI_API_KEY)
# ---------------------------------------------------------------------------


@bp.route("/deskbuddy/chat", methods=["POST"])
def deskbuddy_chat():
    """Proxy chat message to the DeskBuddy agent service."""
    # ① parse the browser's chat JSON
    body = request.get_json(force=True)
    # ② proxy the message to the DeskBuddy agent loop
    data, status = proxy_request("POST", f"{DESKBUDDY_URL}/chat", body)
    # ③ return the agent JSON and status code unchanged
    return jsonify(data), status


@bp.route("/deskbuddy/status")
def deskbuddy_status():
    """Check if DeskBuddy agent service is running."""
    # ① ask DeskBuddy for its health payload
    data, status = proxy_request("GET", f"{DESKBUDDY_URL}/")
    # ② return the health JSON and status code unchanged
    return jsonify(data), status


# ---------------------------------------------------------------------------
# Blueprint registration + server entry point
# ---------------------------------------------------------------------------

app.register_blueprint(bp, url_prefix=PATH_PREFIX)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)