Performance Streamlit

La performance est le plus grand gain. Sans caching et fragments, votre app réexécute tout à chaque interaction.

Caching

@st.cache_data pour les données

Utilisez cette décoration pour toute fonction qui charge ou calcule des données.

# BAD: Recalcule à chaque réexécution
def load_data(path):
    return pd.read_csv(path)

# GOOD: En cache
@st.cache_data
def load_data(path):
    return pd.read_csv(path)

@st.cache_resource pour les connexions

Utilisez cette décoration pour les connexions, clients API, modèles ML—les objets qui ne peuvent pas être sérialisés.

@st.cache_resource
def get_connection():
    return st.connection("snowflake")

@st.cache_resource
def load_model():
    return torch.load("model.pt")

Avertissement critique : Ne modifiez jamais les retours de @st.cache_resource—les modifications affectent tous les utilisateurs :

# BAD: Mutation de ressource partagée
@st.cache_resource
def get_config():
    return {"setting": "default"}

config = get_config()
config["setting"] = "custom"  # Affecte TOUS les utilisateurs !

# GOOD: Copier avant de modifier
config = get_config().copy()
config["setting"] = "custom"

Nettoyage avec on_release : Nettoyez les ressources quand elles sont évincées du cache :

def cleanup_connection(conn):
    conn.close()

@st.cache_resource(on_release=cleanup_connection)
def get_database():
    return create_connection()

TTL pour les données fraîches

@st.cache_data(ttl="5m")  # 5 minutes
def get_metrics():
    return api.fetch()

@st.cache_data(ttl="1h")  # 1 heure
def load_reference_data():
    return pd.read_csv("large_reference.csv")

Directives :

Tableaux de bord temps réel → ttl="1m" ou moins
Métriques/rapports → ttl="5m" à ttl="15m"
Données de référence → ttl="1h" ou plus
Données statiques → Pas de TTL

Empêcher la croissance illimitée du cache

Important : Les caches sans ttl ou max_entries peuvent croître indéfiniment et causer des problèmes de mémoire. Pour toute fonction en cache qui stocke des objets changeants (données spécifiques à l'utilisateur, requêtes paramétrées), définissez des limites :

# BAD: Cache non limité - la mémoire croîtra indéfiniment
@st.cache_data
def get_user_data(user_id):
    return fetch_user(user_id)

# GOOD: Cache limité avec TTL
@st.cache_data(ttl="1h")
def get_user_data(user_id):
    return fetch_user(user_id)

# GOOD: Cache limité avec max entries
@st.cache_data(max_entries=100)
def get_user_data(user_id):
    return fetch_user(user_id)

Utilisez ttl pour l'expiration temporelle OU max_entries pour les limites de taille. Vous n'avez généralement besoin que de l'un ou de l'autre.

Antipatterns de caching

Ne mettez pas en cache les fonctions qui lisent les widgets :

# BAD: Widget à l'intérieur d'une fonction en cache
@st.cache_data
def filtered_data():
    query = st.text_input("Query")  # Widget à l'intérieur d'une fonction en cache !
    return df[df["name"].str.contains(query)]

# GOOD: Passer les valeurs des widgets comme paramètres
@st.cache_data
def filtered_data(query: str):
    return df[df["name"].str.contains(query)]

query = st.text_input("Query")
result = filtered_data(query)

Mettez en cache avec la bonne granularité :

# BAD: Mise en cache excessive - nouvelle entrée de cache par valeur de filtre
@st.cache_data
def get_and_filter_data(filter_value):
    data = load_all_data()  # Coûteux !
    return data[data["col"] == filter_value]

# GOOD: Mettre en cache la partie coûteuse, filtrer séparément
@st.cache_data(ttl="1h")
def load_all_data():
    return fetch_from_database()

data = load_all_data()
filtered = data[data["col"] == filter_value]

Fragments

Utilisez @st.fragment pour isoler les réexécutions dans des éléments UI autonomes.

# BAD: Réexécution complète de l'app
st.metric("Users", get_count())
if st.button("Refresh"):
    st.rerun()

# GOOD: Seul le fragment réexécute
@st.fragment
def live_metrics():
    st.metric("Users", get_count())
    st.button("Refresh")

live_metrics()

Pour les métriques qui s'auto-actualisent, utilisez run_every :

@st.fragment(run_every="30s")
def auto_refresh_metrics():
    st.metric("Users", get_count())

auto_refresh_metrics()

Utilisez pour : les métriques en direct, les boutons d'actualisation, les graphiques interactifs qui n'affectent pas l'état global.

Formulaires pour regrouper les interactions

Par défaut, chaque interaction avec un widget déclenche une réexécution complète. Utilisez st.form pour regrouper plusieurs entrées et ne réexécuter que lors de la soumission.

# BAD: Réexécute à chaque frappe et sélection
name = st.text_input("Name")
email = st.text_input("Email")
role = st.selectbox("Role", ["Admin", "User"])

# GOOD: Réexécution unique quand l'utilisateur clique sur Submit
with st.form("user_form"):
    name = st.text_input("Name")
    email = st.text_input("Email")
    role = st.selectbox("Role", ["Admin", "User"])
    submitted = st.form_submit_button("Submit")

if submitted:
    save_user(name, email, role)

Utilisez border=False pour les formulaires inline transparents qui ne ressemblent pas à des formulaires :

with st.form("search", border=False):
    with st.container(horizontal=True):
        query = st.text_input("Search", label_visibility="collapsed")
        st.form_submit_button(":material/search:")

Quand utiliser les formulaires :

Plusieurs entrées liées (inscription, filtres, paramètres)
Entrées de texte où la saisie déclenche des opérations coûteuses
Toute interface où la sémantique « submit » a du sens

Quand NE PAS utiliser les formulaires : Si les entrées dépendent les unes des autres (p. ex., sélectionner un pays devrait mettre à jour les villes disponibles), les formulaires ne fonctionnent pas puisqu'il n'y a pas de réexécution avant la soumission.

Rendu conditionnel

C'est critique et souvent oublié.

Les conteneurs de mise en page comme st.tabs, st.expander et st.popover rendent toujours tout leur contenu, même quand masqué ou fermé.

Pour rendre le contenu uniquement quand nécessaire, utilisez des éléments comme st.segmented_control, st.toggle ou @st.dialog avec une logique conditionnelle :

# BAD: Le contenu lourd se charge même quand l'onglet n'est pas visible
tab1, tab2 = st.tabs(["Light", "Heavy"])
with tab2:
    expensive_chart()  # Toujours calculé !

# GOOD: Le contenu se charge uniquement quand sélectionné
view = st.segmented_control("View", ["Light", "Heavy"])
if view == "Heavy":
    expensive_chart()  # Calculé uniquement quand sélectionné

# BAD: Le contenu de l'expander se charge toujours
with st.expander("Advanced options"):
    heavy_computation()  # S'exécute même quand fermé !

# GOOD: Le toggle contrôle le chargement
if st.toggle("Show advanced options"):
    heavy_computation()  # S'exécute uniquement quand activé

Pré-calcul

Déplacez le travail coûteux hors du flux principal :

Calculez les agrégations en SQL/dbt, pas en Python
Pré-calculez les métriques dans des tâches planifiées
Utilisez les vues matérialisées pour les requêtes complexes

Gestion des grandes données

Pour les datasets de moins de ~100M lignes

@st.cache_data
def load_data():
    return pd.read_parquet("large_file.parquet")

Pour les très grands datasets (plus de ~100M lignes)

Note : Ceci n'est une échappatoire que quand la sérialisation devient trop lente. Dans la plupart des cas, les données aussi volumineuses ne doivent pas être chargées entièrement en mémoire—préférez une base de données qui requête et charge les données à la demande.

@st.cache_data utilise pickle qui ralentit avec les énormes données. Utilisez plutôt @st.cache_resource :

@st.cache_resource  # Pas de surcharge de sérialisation
def load_huge_data():
    return pd.read_parquet("huge_file.parquet")

# AVERTISSEMENT : Ne modifiez pas le DataFrame retourné !

Échantillonnage pour l'exploration

Quand vous explorez de grands datasets, chargez un échantillon aléatoire au lieu des données complètes :

@st.cache_data(ttl="1h")
def load_sample(n=10000):
    df = pd.read_parquet("huge.parquet")
    return df.sample(n=n)

Multithreading

Les threads personnalisés ne peuvent pas appeler les commandes Streamlit (pas de contexte de session).

import threading

def fetch_in_background(url, results, index):
    results[index] = requests.get(url).json()  # Pas d'appels st.* !

# Collecter les résultats, puis afficher dans le thread principal
results = [None] * len(urls)
threads = [
    threading.Thread(target=fetch_in_background, args=(url, results, i))
    for i, url in enumerate(urls)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

# Maintenant afficher dans le thread principal
for result in results:
    st.write(result)

Préférez les alternatives si possible :

@st.cache_data pour les calculs coûteux
@st.fragment(run_every="5s") pour les mises à jour périodiques

optimizing-streamlit-performance