Deploying AI-Powered Data Applications

FastAPI, Docker, and Streamlit Dashboards

The final step is putting your models and dashboards into the hands of users. For **Walmart**, this means deploying a **Streamlit Dashboard** that supply chain managers use every morning. For **Visa**, it's a **FastAPI** microservice that scores transactions in real-time. And for **Netflix**, it's a containerized recommendation engine running in **Docker**.

In this chapter, we cover **API Serving** with FastAPI, **Containerization** with Docker, **Interactive Dashboards** with Streamlit, and the architecture of a complete end-to-end AI data application.

FastAPI: Serving Models as REST APIs

**FastAPI** turns your Python model into a production REST API in minutes. For **Visa**, this means their XGBoost fraud model sits behind an endpoint that accepts a transaction JSON and returns a fraud score in under 50ms. Every card swipe worldwide hits this API.

PythonRuns entirely in your browser — nothing is sent to a server.

# Visa Fraud Scoring: FastAPI Endpoint Simulation
class FastAPIApp:
    def __init__(self):
        self.routes = {}

    def post(self, path):
        def decorator(func):
            self.routes[path] = func
            return func
        return decorator

    def simulate_request(self, path, body):
        if path in self.routes:
            print("POST {0}".format(path))
            print("Request Body: {0}".format(body))
            result = self.routes[path](body)
            print("Response (200 OK): {0}".format(result))
            return result
        print("404 Not Found: {0}".format(path))

# Create the app
app = FastAPIApp()

@app.post("/api/v1/fraud-score")
def score_transaction(tx):
    amount = tx.get("amount", 0)
    is_foreign = tx.get("is_foreign", False)
    risk = round(min(amount / 5000 + (0.3 if is_foreign else 0), 1.0), 3)
    return {"risk_score": risk, "decision": "BLOCK" if risk > 0.7 else "ALLOW"}

# Simulate a transaction scoring request
app.simulate_request("/api/v1/fraud-score", {
    "amount": 2500, "is_foreign": True, "merchant": "UnknownShop"
})

# Visa Fraud Scoring: FastAPI Endpoint Simulation
class FastAPIApp:
    def __init__(self):
        self.routes = {}

    def post(self, path):
        def decorator(func):
            self.routes[path] = func
            return func
        return decorator

    def simulate_request(self, path, body):
        if path in self.routes:
            print("POST {0}".format(path))
            print("Request Body: {0}".format(body))
            result = self.routes[path](body)
            print("Response (200 OK): {0}".format(result))
            return result
        print("404 Not Found: {0}".format(path))

# Create the app
app = FastAPIApp()

@app.post("/api/v1/fraud-score")
def score_transaction(tx):
    amount = tx.get("amount", 0)
    is_foreign = tx.get("is_foreign", False)
    risk = round(min(amount / 5000 + (0.3 if is_foreign else 0), 1.0), 3)
    return {"risk_score": risk, "decision": "BLOCK" if risk > 0.7 else "ALLOW"}

# Simulate a transaction scoring request
app.simulate_request("/api/v1/fraud-score", {
    "amount": 2500, "is_foreign": True, "merchant": "UnknownShop"
})

Docker: Containerizing Your Application

**Docker** packages your application, dependencies, and configuration into a **container** that runs identically everywhere. For **Netflix**, this means the recommendation engine runs the same way on a developer's laptop, in staging, and across 10,000 production servers. "It works on my machine" becomes "It works everywhere."

PythonRuns entirely in your browser — nothing is sent to a server.

# Netflix Deployment: Docker Container Simulation
class DockerContainer:
    def __init__(self, image_name, tag):
        self.image = image_name
        self.tag = tag
        self.layers = []

    def add_layer(self, instruction, command):
        self.layers.append({"instruction": instruction, "command": command})

    def build(self):
        print("DOCKER BUILD: {0}:{1}".format(self.image, self.tag))
        print("=" * 50)
        for i, layer in enumerate(self.layers):
            print("Layer {0}: {1} {2}".format(i + 1, layer["instruction"], layer["command"]))
        print("\nImage Size: ~850MB (Python + ML deps)")
        print("Build: SUCCESS")

    def run(self):
        print("\nDOCKER RUN: {0}:{1}".format(self.image, self.tag))
        print(" -> Container started on port 8000")
        print(" -> Health check: OK")
        print(" -> Ready to serve predictions!")

# Scenario: Netflix recommendation engine container
container = DockerContainer("netflix-recommender", "v2.1")
container.add_layer("FROM", "python:3.11-slim")
container.add_layer("COPY", "requirements.txt .")
container.add_layer("RUN", "pip install -r requirements.txt")
container.add_layer("COPY", "app/ /app")
container.add_layer("CMD", "uvicorn app.main:app --host 0.0.0.0 --port 8000")
container.build()
container.run()

# Netflix Deployment: Docker Container Simulation
class DockerContainer:
    def __init__(self, image_name, tag):
        self.image = image_name
        self.tag = tag
        self.layers = []

    def add_layer(self, instruction, command):
        self.layers.append({"instruction": instruction, "command": command})

    def build(self):
        print("DOCKER BUILD: {0}:{1}".format(self.image, self.tag))
        print("=" * 50)
        for i, layer in enumerate(self.layers):
            print("Layer {0}: {1} {2}".format(i + 1, layer["instruction"], layer["command"]))
        print("\nImage Size: ~850MB (Python + ML deps)")
        print("Build: SUCCESS")

    def run(self):
        print("\nDOCKER RUN: {0}:{1}".format(self.image, self.tag))
        print(" -> Container started on port 8000")
        print(" -> Health check: OK")
        print(" -> Ready to serve predictions!")

# Scenario: Netflix recommendation engine container
container = DockerContainer("netflix-recommender", "v2.1")
container.add_layer("FROM", "python:3.11-slim")
container.add_layer("COPY", "requirements.txt .")
container.add_layer("RUN", "pip install -r requirements.txt")
container.add_layer("COPY", "app/ /app")
container.add_layer("CMD", "uvicorn app.main:app --host 0.0.0.0 --port 8000")
container.build()
container.run()

Streamlit: Interactive Data Dashboards

**Streamlit** lets data scientists create beautiful, interactive dashboards in pure Python—no frontend skills needed. For **Walmart**, a Streamlit dashboard shows real-time inventory levels, demand forecasts, and supply chain bottleneck alerts. Managers interact with filters, sliders, and charts to explore the data themselves.

PythonRuns entirely in your browser — nothing is sent to a server.

# Walmart Dashboard: Streamlit App Simulation
class StreamlitApp:
    def __init__(self, title):
        self.title = title
        self.components = []

    def add_metric(self, label, value, delta=None):
        self.components.append(("METRIC", label, value, delta))

    def add_chart(self, chart_type, data):
        self.components.append(("CHART", chart_type, data, None))

    def render(self):
        print("STREAMLIT DASHBOARD: {0}".format(self.title))
        print("=" * 55)
        
        for comp_type, label, value, delta in self.components:
            if comp_type == "METRIC":
                arrow = "+" if delta and delta > 0 else ""
                delta_str = " ({0}{1}%)".format(arrow, delta) if delta else ""
                print(" [{0}] {1}{2}".format(label, value, delta_str))
            elif comp_type == "CHART":
                print(" [CHART: {0}] Data points: {1}".format(label, value))
        
        print("\nDashboard is LIVE at http://localhost:8501")

# Scenario: Walmart's Morning Ops Dashboard
app = StreamlitApp("Walmart Supply Chain Command Center")
app.add_metric("Global Inventory", "2.4M units", delta=3.2)
app.add_metric("Active Shipments", "18,492", delta=-1.5)
app.add_metric("Cold Chain Compliance", "97.3%", delta=0.8)
app.add_chart("bar_chart", "Revenue by Region (5 regions)")
app.add_chart("line_chart", "Demand Forecast (Next 4 weeks)")
app.render()

# Walmart Dashboard: Streamlit App Simulation
class StreamlitApp:
    def __init__(self, title):
        self.title = title
        self.components = []

    def add_metric(self, label, value, delta=None):
        self.components.append(("METRIC", label, value, delta))

    def add_chart(self, chart_type, data):
        self.components.append(("CHART", chart_type, data, None))

    def render(self):
        print("STREAMLIT DASHBOARD: {0}".format(self.title))
        print("=" * 55)
        
        for comp_type, label, value, delta in self.components:
            if comp_type == "METRIC":
                arrow = "+" if delta and delta > 0 else ""
                delta_str = " ({0}{1}%)".format(arrow, delta) if delta else ""
                print(" [{0}] {1}{2}".format(label, value, delta_str))
            elif comp_type == "CHART":
                print(" [CHART: {0}] Data points: {1}".format(label, value))
        
        print("\nDashboard is LIVE at http://localhost:8501")

# Scenario: Walmart's Morning Ops Dashboard
app = StreamlitApp("Walmart Supply Chain Command Center")
app.add_metric("Global Inventory", "2.4M units", delta=3.2)
app.add_metric("Active Shipments", "18,492", delta=-1.5)
app.add_metric("Cold Chain Compliance", "97.3%", delta=0.8)
app.add_chart("bar_chart", "Revenue by Region (5 regions)")
app.add_chart("line_chart", "Demand Forecast (Next 4 weeks)")
app.render()

Practice Questions

Question 1

Why is Docker essential for deploying ML models at scale?

It makes Python code faster
It packages the entire application environment into a portable container that runs identically everywhere, eliminating environment-specific bugs
It replaces the need for a database
It only works on Linux

Question 2

What makes Streamlit uniquely suited for data science teams at Walmart?

It uses JavaScript instead of Python
It allows data scientists to create interactive, beautiful dashboards in pure Python without needing frontend development skills
It only works with Tableau data
It requires a team of frontend developers