writing-streamlit-apps

Par posthog · skills

Écrire le code source d'une application Streamlit qui fonctionne correctement dans un sandbox PostHog — le bridge `posthog_apps.query()` pour lire les données PostHog, les packages intégrés dans l'image sandbox, la gestion du cache et du session state entre les reruns Streamlit, les patterns de mise en page et de graphiques, et la structure en fichier unique `app.py`. À utiliser lors de la création ou du débogage du code Python d'une application Streamlit PostHog, lorsqu'une requête dans une application échoue, ou quand on demande de « créer une application Streamlit affichant des données PostHog ».

npx skills add https://github.com/posthog/skills --skill writing-streamlit-apps

Écrire des apps Streamlit pour le sandbox PostHog

Le code source que tu écris devient app.py à la racine d'un runtime Streamlit 1.31 en sandbox. Les mécaniques de déploiement (create/start/share) relèvent de la skill managing-streamlit-apps ; celle-ci concerne le code.

Lire les données PostHog : posthog_apps.query()

La seule et unique porte d'accès aux données est le pont in-sandbox :

import posthog_apps

df = posthog_apps.query("SELECT event, count() FROM events GROUP BY event LIMIT 10")
  • Accepte une chaîne HogQL, retourne un pandas DataFrame.
  • Lève RuntimeError en cas d'échec. Le message est volontairement générique (« Query execution failed ») — le pont ne retourne pas les détails internes au sandbox, tu ne peux donc pas diagnostiquer une mauvaise requête depuis l'app. Capte l'erreur et affiche-la avec st.error(str(e)) pour que les spectateurs voient un message au lieu d'une pile d'appels, et teste les requêtes dans l'éditeur SQL où les vraies erreurs sont visibles.
  • import posthog N'EXISTE PAS dans le sandbox — le module est posthog_apps, volontairement distinct du nom du SDK posthog-python.
  • Le pont est pré-authentifié au projet de l'app ; le code utilisateur ne voit jamais de token, et il n'y a rien à configurer.
  • Les requêtes s'exécutent avec des plafonds côté serveur (30 s d'exécution, 256 MB de mémoire) qui ne varient pas selon la taille du sandbox — borne donc les plages temporelles, LIMIT les résultats, et agrège en HogQL plutôt que de tirer des événements bruts dans pandas.

Concevoir pour le modèle de rerun de Streamlit

Streamlit réexécute le script entier de haut en bas à chaque interaction avec un widget. Deux conséquences :

  1. Cache chaque appel au pont — sans cache, un glissement de curseur re-tire chaque requête :

    @st.cache_data(ttl=300, show_spinner="Running query...")
    def run_query(hogql: str) -> pd.DataFrame:
        return posthog_apps.query(hogql)
  2. Utilise st.session_state pour tout ce qui doit survivre aux reruns — sélections accumulées, curseurs de pagination, timestamps « dernière actualisation ». Les variables au niveau du module se réinitialisent à chaque interaction.

Les widgets pilotent naturellement les paramètres, mais n'interole jamais une valeur de widget en texte libre dans HogQL. Le pont exécute ta requête avec l'accès aux données de l'auteur de la version, et quiconque peut voir l'app pilote ces widgets — un st.text_input brut épissé dans une requête donne aux spectateurs l'accès de l'auteur pour écrire les leurs. Contrains l'entrée à la place : choisis parmi une liste fixe que tu contrôles (st.selectbox sur des valeurs connues), ou force à un type qui ne peut pas porter du SQL (int(days), une date de st.date_input), et valide avant qu'elle n'atteigne la requête.

Disposition et graphiques

  • st.set_page_config(page_title=..., layout="wide") en premier — la disposition narrow par défaut gaspille l'écran pour les apps de données.
  • Structure avec st.columns pour des métriques côte à côte, st.tabs pour des vues alternatives, st.expander pour des sections de détail ; st.metric pour les nombres phares.
  • Graphiques : st.plotly_chart(fig, use_container_width=True) avec plotly express est la référence fiable ; st.dataframe(df, use_container_width=True) pour les tableaux. matplotlib/seaborn fonctionnent aussi via st.pyplot.

Ce qui est installé

L'image embarque Python 3.11 avec : streamlit 1.31, pandas, numpy, polars, plotly, matplotlib, seaborn, scipy, scikit-learn, pyarrow, duckdb, requests, beautifulsoup4, lxml, sqlalchemy, aiohttp. Il n'y a aucun moyen d'ajouter des dépendances : le sandbox ne lance jamais pip (une posture de sécurité délibérée — pas de code package arbitraire au boot), et un requirements.txt dans un zip uploadé est toléré mais supprimé. Importe seulement ce qui est listé ci-dessus.

Structure et contraintes runtime

  • Un seul fichier. Via le flux set-source du MCP ta source EST app.py ; il n'y a pas d'autres modules, garde donc tout dedans.
  • Le sandbox est éphémère : tout ce qui est écrit sur le disque disparaît à l'arrêt/redémarrage. Ne construis pas d'état sur des fichiers ; recalcule à partir de requêtes (avec cache) ou tiens-le dans st.session_state.
  • Ton code s'exécute comme utilisateur non privilégié ; pas de SDK posthog, aucun moyen de passer tes propres variables d'environnement ou secrets à l'app, et pas d'attente d'egress réseau général. Ne lis pas depuis os.environ — tout ce qui s'y trouve appartient au runtime du sandbox, pas à ton app. Conçois autour de posthog_apps.query() comme source de données.

Une app minimale bien formée

import pandas as pd
import plotly.express as px
import posthog_apps
import streamlit as st

st.set_page_config(page_title="Events overview", layout="wide")
st.title("Events overview")


@st.cache_data(ttl=300, show_spinner="Running query...")
def run_query(hogql: str) -> pd.DataFrame:
    return posthog_apps.query(hogql)


# Un curseur est borné et forcé à int, donc il est sûr d'interpoler.
days = int(st.slider("Days to show", 1, 30, 7))
try:
    daily = run_query(
        f"""
        SELECT toDate(timestamp) AS day, count() AS events
        FROM events
        WHERE timestamp >= now() - INTERVAL {days} DAY
        GROUP BY day ORDER BY day
        """
    )
    st.plotly_chart(px.bar(daily, x="day", y="events"), use_container_width=True)
except RuntimeError as e:
    st.error(str(e))

Skills similaires