Tutorial Lengkap Streamlit Advanced: Build Production-Ready ML Apps

Streamlit adalah library Python yang powerful untuk membangun aplikasi web interaktif untuk machine learning dan data science. Tutorial advanced ini mencakup production patterns, optimasi performa, dan fitur enterprise.

Mengapa Streamlit untuk Production?

Keunggulan Streamlit:

• Rapid development: Build apps dalam hitungan jam
• Pure Python: Tidak perlu frontend knowledge
• Interactive widgets: Rich UI components
• Easy deployment: Streamlit Cloud, Docker, Kubernetes
• Active ecosystem: Community components dan integrasi

Use Cases:

• ML model demos dan dashboards
• Data exploration tools
• Internal analytics apps
• Customer-facing applications
• Prototyping dan MVPs

Instalasi

pip install streamlit

Dengan fitur tambahan
pip install streamlit-extras
pip install streamlit-aggrid
pip install plotly

Verify installation
streamlit --version

App Architecture

1. Multi-Page Apps

# pages/1Home.py
import streamlit as st

st.setpageconfig(

    pagetitle="ML Dashboard",
    pageicon="🤖",

    layout="wide",
)

st.title("Welcome to ML Dashboard")
st.write("Navigate menggunakan sidebar")

# pages/2DataExplorer.py
import streamlit as st
import pandas as pd

st.title("Data Explorer")

uploadedfile = st.fileuploader("Upload CSV", type="csv")

if uploadedfile:
    df = pd.readcsv(uploadedfile)
    st.dataframe(df)

# pages/3ModelInference.py
import streamlit as st

st.title("Model Inference")
Model inference code disini

2. Session State Management

import streamlit as st

Initialize session state
if 'counter' not in st.sessionstate:

    st.sessionstate.counter = 0

if 'userdata' not in st.sessionstate:
    st.sessionstate.userdata = {}

Update session state
def increment():
    st.sessionstate.counter += 1


st.button("Increment", onclick=increment)
st.write(f"Counter: {st.sessionstate.counter}")


Store user data
name = st.textinput("Name", key="nameinput")

if name:
    st.sessionstate.userdata['name'] = name


Access across pages
st.write(st.sessionstate.userdata)

3. Callbacks dan Events

import streamlit as st

Callback function
def onsubmit():
    st.sessionstate.submitted = True

    st.sessionstate.result = f"Hello, {st.sessionstate.namefield}!"

Form dengan callback
with st.form("myform"):

    st.textinput("Name", key="namefield")

    submitted = st.formsubmitbutton("Submit", onclick=onsubmit)


if st.sessionstate.get('submitted'):
    st.success(st.sessionstate.result)


Multiple callbacks
def clearform():
    st.sessionstate.namefield = ""
    st.sessionstate.submitted = False


st.button("Clear", onclick=clearform)

Caching dan Performance

1. Cache Data

import streamlit as st
import pandas as pd

@st.cachedata(ttl=3600)  # Cache selama 1 jam
def loaddata(url):

    """Load dan cache data"""
    return pd.readcsv(url)

@st.cachedata(showspinner="Loading data...")
def expensivecomputation(df):

    """Expensive computation dengan spinner"""
    # Simulate long computation
    import time
    time.sleep(5)
    return df.describe()

Gunakan cached functions
df = loaddata("https://example.com/data.csv")
stats = expensivecomputation(df)

2. Cache Resources

import streamlit as st
from transformers import pipeline
import joblib

@st.cacheresource
def loadmodel():

    """Cache ML model (singleton)"""
    return joblib.load("model.joblib")

@st.cacheresource
def loadllm():

    """Cache LLM pipeline"""
    return pipeline("text-generation", model="gpt2")

Models loaded sekali dan reused
model = loadmodel()
llm = loadllm()

3. Cache Configuration

import streamlit as st

Cache dengan hash functions
@st.cachedata(hashfuncs={pd.DataFrame: lambda x: x.tojson()})
def processdataframe(df):

    return df.groupby('category').sum()

Cache dengan max entries
@st.cachedata(maxentries=100)

def getuserdata(userid):
    return fetchfromdatabase(userid)


Clear cache
if st.button("Clear Cache"):
    st.cachedata.clear()
    st.cacheresource.clear()

Advanced Components

1. Interactive Tables dengan AgGrid

import streamlit as st
from staggrid import AgGrid, GridOptionsBuilder
import pandas as pd

df = pd.readcsv("data.csv")


Configure grid options
gb = GridOptionsBuilder.fromdataframe(df)
gb.configurepagination(paginationAutoPageSize=True)

gb.configureselection('multiple', usecheckbox=True)

gb.configuresidebar()

gridoptions = gb.build()

Display grid
gridresponse = AgGrid(

    df,
    gridOptions=gridoptions,
    enableenterprisemodules=True,
    theme='streamlit'
)

Get selected rows
selectedrows = gridresponse['selectedrows']

if selectedrows:
    st.write("Selected:", selectedrows)

2. Interactive Charts dengan Plotly

import streamlit as st
import plotly.express as px
import plotly.graphobjects as go

Interactive scatter plot
fig = px.scatter(
    df,
    x="feature1",
    y="feature2",
    color="category",
    size="value",
    hoverdata=["name"]

)
st.plotlychart(fig, usecontainerwidth=True)

Interactive line chart dengan range slider
fig = go.Figure()
fig.addtrace(go.Scatter(x=dates, y=values, mode='lines'))

fig.updatelayout(
    xaxis=dict(rangeslider=dict(visible=True)),
    title="Time Series dengan Range Slider"
)
st.plotlychart(fig)


Callback on chart selection
selectedpoints = st.plotlychart(fig, onselect="rerun")
if selectedpoints:

    st.write("Selected points:", selectedpoints)

3. Custom Components

import streamlit as st
import streamlit.components.v1 as components

Embed HTML/JS
components.html(
    """
    
    
    
    """,
    height=400
)

Embed iframe
components.iframe("https://example.com/dashboard", height=600)

Real-time Updates

1. Auto-refresh

import streamlit as st
import time

Auto-refresh setiap 5 detik
stautorefresh = st.empty()

with stautorefresh:
    st.write(f"Last updated: {time.strftime('%H:%M:%S')}")
    # Real-time data Anda disini

Manual refresh
if st.button("Refresh"):
    st.rerun()

2. Streaming Updates

import streamlit as st
import time

Streaming text
def streamresponse():

    response = "Ini adalah streaming response..."
    for word in response.split():
        yield word + " "
        time.sleep(0.1)

st.writestream(streamresponse)


Streaming dengan placeholder
placeholder = st.empty()
for i in range(100):
    placeholder.metric("Progress", f"{i}%")
    time.sleep(0.1)

3. WebSocket Updates

import streamlit as st
import asyncio
import websockets

@st.cacheresource
def getwebsocketdata():
    """Connect ke WebSocket dan return data"""
    async def connect():
        async with websockets.connect("wss://example.com/ws") as ws:
            return await ws.recv()

    return asyncio.run(connect())

Display real-time data
data = getwebsocketdata()
st.write(data)

Authentication dan Security

1. Basic Authentication

import streamlit as st
import hashlib

def checkpassword():

    """Returns True jika user punya password benar"""

    def passwordentered():
        if st.sessionstate["username"] in st.secrets["passwords"] and \

           hashlib.sha256(st.sessionstate["password"].encode()).hexdigest() == \
           st.secrets["passwords"][st.sessionstate["username"]]:

            st.sessionstate["passwordcorrect"] = True

            del st.sessionstate["password"]
        else:
            st.sessionstate["passwordcorrect"] = False

    if "passwordcorrect" not in st.sessionstate:
        st.textinput("Username", key="username")

        st.textinput("Password", type="password", key="password")
        st.button("Login", onclick=passwordentered)
        return False

    elif not st.sessionstate["passwordcorrect"]:
        st.textinput("Username", key="username")

        st.textinput("Password", type="password", key="password")
        st.button("Login", onclick=passwordentered)
        st.error("Username atau password salah")
        return False

    return True

if checkpassword():

    st.write("Welcome to the app!")

2. Secrets Management

# .streamlit/secrets.toml
[passwords]
admin = "5e884898da28047d91..."  # hashed password

[database]
host = "localhost"
port = 5432
user = "appuser"
password = "secretpassword"


[apikeys]
openai = "sk-..."

import streamlit as st

Akses secrets
dbhost = st.secrets["database"]["host"]

apikey = st.secrets["apikeys"]["openai"]

3. Role-based Access

import streamlit as st

Define user roles
USERS = {
    "admin": {"password": "admin123", "role": "admin"},
    "analyst": {"password": "analyst123", "role": "analyst"},
    "viewer": {"password": "viewer123", "role": "viewer"},
}

def getuserrole():

    if "userrole" not in st.sessionstate:

        return None
    return st.sessionstate.userrole


def requirerole(allowedroles):

    """Decorator untuk check user role"""
    role = getuserrole()

    if role not in allowedroles:
        st.error("Anda tidak punya permission untuk akses fitur ini")
        st.stop()

Usage
role = getuserrole()
if role == "admin":
    st.write("Admin panel")
    # Show admin features
elif role == "analyst":
    st.write("Analyst dashboard")
    # Show analyst features
else:
    st.write("View-only dashboard")

Database Integration

1. SQL Connection

import streamlit as st
import pandas as pd

@st.cacheresource

def getconnection():
    return st.connection("postgresql", type="sql")

conn = getconnection()


Query data
df = conn.query("SELECT  FROM users WHERE active = true", ttl=600)

st.dataframe(df)

Parameterized query
userid = st.numberinput("User ID", minvalue=1)
user = conn.query(
    "SELECT  FROM users WHERE id = :id",
    params={"id": userid}

)

2. MongoDB Integration

import streamlit as st
from pymongo import MongoClient

@st.cacheresource
def getmongoclient():
    return MongoClient(st.secrets["mongodb"]["uri"])

client = getmongoclient()
db = client["mydb"]
collection = db["users"]

Query data
users = list(collection.find({"active": True}))
st.write(users)

Insert data
if st.button("Add User"):
    collection.insertone({"name": "New User", "active": True})

    st.success("User added!")

3. Redis untuk Caching

import streamlit as st
import redis
import json

@st.cacheresource
def getredis():

    return redis.Redis(
        host=st.secrets["redis"]["host"],
        port=st.secrets["redis"]["port"],
        password=st.secrets["redis"]["password"]
    )

r = getredis()

Cache expensive computation
def getcacheddata(key):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    # Compute jika tidak cached
    data = expensivecomputation()

    r.setex(key, 3600, json.dumps(data))  # Cache selama 1 jam
    return data

ML Model Integration

1. Model Inference Dashboard

import streamlit as st
import pandas as pd
import joblib

@st.cacheresource
def loadmodel():

    return joblib.load("model.joblib")

model = loadmodel()

st.title("ML Model Inference")

Input features
col1, col2 = st.columns(2)
with col1:
    feature1 = st.numberinput("Feature 1", value=0.0)

    feature2 = st.numberinput("Feature 2", value=0.0)
with col2:
    feature3 = st.selectbox("Category", ["A", "B", "C"])
    feature4 = st.slider("Score", 0, 100, 50)

Predict
if st.button("Predict"):
    features = pd.DataFrame([[feature1, feature2, feature3, feature4]],
                           columns=["f1", "f2", "category", "score"])

    prediction = model.predict(features)[0]
    probability = model.predictproba(features)[0]


    st.success(f"Prediction: {prediction}")
    st.write("Probabilities:", dict(zip(model.classes, probability)))

2. Batch Prediction

import streamlit as st
import pandas as pd

st.title("Batch Prediction")

uploadedfile = st.fileuploader("Upload CSV untuk prediction", type="csv")

if uploadedfile:

    df = pd.readcsv(uploadedfile)

    st.write("Preview:", df.head())

    if st.button("Run Batch Prediction"):
        with st.spinner("Running predictions..."):
            predictions = model.predict(df)
            df["prediction"] = predictions

        st.success(f"Completed {len(df)} predictions")
        st.dataframe(df)

        # Download results
        csv = df.tocsv(index=False)
        st.downloadbutton(

            "Download Results",
            csv,
            "predictions.csv",
            "text/csv"
        )

3. LLM Chat Interface

import streamlit as st
from openai import OpenAI

st.title("AI Chat Assistant")

client = OpenAI(apikey=st.secrets["openai"]["apikey"])


Initialize chat history
if "messages" not in st.sessionstate:
    st.sessionstate.messages = []


Display chat history
for message in st.sessionstate.messages:
    with st.chatmessage(message["role"]):

        st.markdown(message["content"])

Chat input
if prompt := st.chatinput("Ada yang bisa dibantu?"):
    st.sessionstate.messages.append({"role": "user", "content": prompt})


    with st.chatmessage("user"):
        st.markdown(prompt)

    with st.chatmessage("assistant"):

        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=st.sessionstate.messages,
            stream=True,
        )
        response = st.writestream(stream)


    st.sessionstate.messages.append({"role": "assistant", "content": response})

Deployment

1. Docker Deployment

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

# docker-compose.yml
version: '3.8'
services:
  streamlit:
    build: .
    ports:
"8501:8501"
    volumes:

./data:/app/data
    environment:

STREAMLITSERVERHEADLESS=true

2. Kubernetes Deployment

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: streamlit-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: streamlit
  template:
    metadata:
      labels:
        app: streamlit
    spec:
      containers:
name: streamlit
        image: myregistry/streamlit-app:latest

        ports:
containerPort: 8501
        resources:

          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "500m"

apiVersion: v1
kind: Service
metadata:
  name: streamlit-service
spec:
  selector:
    app: streamlit
  ports:
port: 80
    targetPort: 8501

  type: LoadBalancer

3. Configuration

# .streamlit/config.toml
[server]
headless = true
port = 8501
enableCORS = false
maxUploadSize = 200

[theme]
primaryColor = "#FF4B4B"
backgroundColor = "#FFFFFF"
secondaryBackgroundColor = "#F0F2F6"
textColor = "#262730"
font = "sans serif"

[browser]
gatherUsageStats = false

Best Practices

1. Error Handling

import streamlit as st

def safeoperation():

    try:
        result = riskyoperation()
        return result
    except ValueError as e:
        st.error(f"Invalid value: {e}")
    except ConnectionError:
        st.error("Connection failed. Silakan coba lagi.")
    except Exception as e:
        st.exception(e)  # Show full traceback
    return None

Global error handler
try:
    main()
except Exception as e:
    st.error("Terjadi unexpected error")
    st.exception(e)

2. Logging

import streamlit as st
import logging

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

def processdata(df):

    logger.info(f"Processing {len(df)} rows")
    # Process data
    logger.info("Processing complete")
    return df

Log user actions
if st.button("Run Analysis"):
    logger.info(f"User started analysis at {datetime.now()}")

3. Testing

# testapp.py
from streamlit.testing.v1 import AppTest

def testapploads():
    at = AppTest.fromfile("app.py")

    at.run()
    assert not at.exception

def testbuttonclick():

    at = AppTest.fromfile("app.py")
    at.run()
    at.button[0].click().run()
    assert at.success[0].value == "Success!"

Kesimpulan

Streamlit production-ready untuk ML applications dengan:

Multi-page apps: Organized navigation

Caching: Performance optimization

Authentication: Security features

Database integration: SQL, NoSQL support

Deployment options: Cloud, Docker, Kubernetes

Key takeaways:

• Gunakan caching untuk expensive operations
• Implement proper authentication
• Structure apps dengan multi-page architecture
• Handle errors dengan graceful
• Monitor dan log user actions