Design System — Core layer / Couche Core

Update date : 2026-05-16 02:50

Session management, page initialization, data-loading strategies, and CRUD operations. Gestion de session, initialisation de page, stratégies de chargement des données et opérations CRUD.

---

Session — session.py

from pinky_streamlit import get_session, LocalSessionConfig

get_session() detects the execution environment automatically. / get_session() détecte l'environnement d'exécution automatiquement.

• In SiS → uses get_active_session() and st.user.login_name. / En SiS → utilise get_active_session() et st.user.login_name.
• Locally → uses ~/.snowflake/connections.toml via LocalSessionConfig. / En local → utilise ~/.snowflake/connections.toml via LocalSessionConfig.

# Local dev / Dev local
config  = LocalSessionConfig(connection="myconn")
session = get_session(config)

# SiS — no config needed / SiS — aucune config nécessaire
session = get_session()

# ~/.snowflake/connections.toml
[myconn]
account       = "ORG-ACCOUNT"
user          = "me@example.com"
warehouse     = "DEV_WH"
database      = "MY_DB"
schema        = "PUBLIC"
authenticator = "externalbrowser"

Resolve environment / Résoudre l'environnement

The active database suffix gives the current environment. / Le suffixe de la base active donne l'environnement courant.

env = str(session.get_current_database()).replace('"', '').split("_")[-1]
# e.g. "MY_DB_QA" → "QA", "MY_DB_PROD" → "PROD"

---

Page setup — page.py

from pinky_streamlit import setup_page, Logo, AppFont, DEFAULT_FONT

Call setup_page() once at the top of every page, before any other st.* call. / Appeler setup_page() une seule fois en haut de chaque page, avant tout autre appel st.*.

Signature

setup_page(
    session: snowpark.Session,
    loaders: list[Callable] | None = None,
    show_progress: bool = True,
    logo: Logo | None = None,
    palette: Palette | None = None,
    font: AppFont | None = DEFAULT_FONT,
) -> tuple[Any, ...]

Returns (messages_container, main_container, *loader_results). / Retourne (messages_container, main_container, *loader_results).

Loaders

Each loader has signature (session) → Any. Use functools.partial to pre-fill additional arguments. / Chaque loader a la signature (session) → Any. Utiliser functools.partial pour pré-remplir les arguments supplémentaires.

from functools import partial

messages, main, df_kpis, df_detail = setup_page(
    session=session,
    loaders=[
        partial(load_analytic, _query_fn=query_kpis, key="kpis"),
        partial(load_crud,     query_fn=query_detail, key="detail"),
    ],
    palette=PALETTES.PYTHIA,
)

Logo

Logo(
    images_filepath="/app/images",  # directory containing the image files
    image="logo_img.png",           # main logo (sidebar)
    icon_image="logo_icon.png",     # small icon (collapsed sidebar) — optional
    size="large",                   # "small" | "medium" | "large"
)

Font

DEFAULT_FONT embeds DM Sans (100–900) as a base64 data URI — identical rendering locally and in SiS, no external URL required. / DEFAULT_FONT embarque DM Sans (100–900) en base64 — rendu identique en local et en SiS, aucune URL externe nécessaire.

# App-level font override / Override police au niveau app
setup_page(session, font=AppFont(family="Syne", woff2_path="/app/fonts/Syne.woff2"))

# Disable bundled font — use Streamlit default / Désactiver la police — utiliser le défaut Streamlit
setup_page(session, font=None)

---

Cache strategies / Stratégies de cache

All data loading goes through db.py. Four patterns, each matching a real data update context. / Tout chargement de données passe par db.py. Quatre patterns, chacun adapté à un contexte réel de mise à jour.

from pinky_streamlit import (
    load_static, load_analytic, load_crud, load_nocache
)

Decision tree / Arbre de décision

Pattern	When to use / Quand utiliser	Invalidation
load_static	Assets deployed with the app (CSS, JSON config, lookup files on stage) — never change without a redeploy / Assets déployés avec l'app — ne changent qu'au redéploiement	Process restart only / Uniquement au redémarrage du process
load_analytic	Pipeline data refreshed ~1×/day — staleness acceptable within the pipeline cadence / Données pipeline rafraîchies ~1×/jour — péremption acceptable dans la cadence pipeline	TTL (default 1h) — set TTL = pipeline frequency / TTL = fréquence pipeline
load_crud	App-managed reference data — users can INSERT or UPDATE during the session / Données de référentiel gérées dans l'app — les utilisateurs peuvent INSERT ou UPDATE pendant la session	Manual: call invalidate_crud(key) after every write / Manuel : appeler invalidate_crud(key) après chaque écriture
load_nocache	Dev/debug only — reloads on every rerun / Dev/debug uniquement — rechargement à chaque rerun	Never used in production / Jamais utilisé en production

load_static — @st.cache_resource (process lifetime / durée de vie du process)

from functools import partial

messages, main, cfg = setup_page(
    session,
    loaders=[partial(load_static, _query_fn=lambda s: s.file.get("@STAGE/config.json"), key="cfg")]
)

• Shared across all sessions on the same process. / Partagé entre toutes les sessions du même process.
• Never serialized / pickled. / Jamais sérialisé / picklé.
• Use for: JSON/YAML config on stage, reference lookup tables that never change. / Utiliser pour : config JSON/YAML sur stage, tables de référence immuables.

load_analytic — @st.cache_data(ttl=3600)

from functools import partial

messages, main, df = setup_page(
    session,
    loaders=[partial(load_analytic, _query_fn=query_kpis, key="kpis")]
)

• TTL = 1 hour by default — matches typical daily pipeline cadence. / TTL = 1 heure par défaut — correspond à la cadence pipeline journalière typique.
• Data is serialized and cached per Streamlit session. / Les données sont sérialisées et mises en cache par session Streamlit.
• Use for: KPIs, aggregates, read-only dashboards fed by a scheduled pipeline. / Utiliser pour : KPIs, agrégats, dashboards en lecture seule alimentés par un pipeline planifié.

load_crud — st.session_state + flag

from functools import partial

messages, main, df = setup_page(
    session,
    loaders=[partial(load_crud, query_fn=query_buildings, key="buildings")]
)

• Data stored in st.session_state["df_buildings"]. / Données stockées dans st.session_state["df_buildings"].
• Reload triggered by setting st.session_state["refresh_buildings"] = True. / Rechargement déclenché en définissant st.session_state["refresh_buildings"] = True.
• Always call invalidate_crud(key) after INSERT / UPDATE / DELETE. / Toujours appeler invalidate_crud(key) après INSERT / UPDATE / DELETE.

update_data(
    session=session, data=changes, key="ID", table="MY_TABLE",
    on_success=lambda: invalidate_crud("buildings"),
)

load_nocache — no cache / sans cache

from functools import partial

messages, main, df = setup_page(
    session,
    loaders=[partial(load_nocache, query_fn=query_data, key="data")]
)

• Fetches fresh data on every rerun. / Récupère des données fraîches à chaque rerun.
• Only for development and debugging. / Uniquement pour le développement et le débogage.

---

NaN → None convention

Pandas converts SQL NULL to NaN on to_pandas(). This breaks is not None checks in Python. / Pandas convertit les NULL SQL en NaN lors du to_pandas(). Cela casse les tests is not None en Python.

Rule: always chain .where(lambda d: d.notna(), None) after every to_pandas(). / Règle : toujours chaîner .where(lambda d: d.notna(), None) après chaque to_pandas().

df = session.table("MY_TABLE").to_pandas().where(lambda d: d.notna(), None)

This is already applied internally by load_static, load_analytic, load_crud, and load_nocache. No action needed when using these functions. / Déjà appliqué en interne par load_static, load_analytic, load_crud et load_nocache. Aucune action nécessaire lors de l'utilisation de ces fonctions.

---

Write operations / Opérations d'écriture

from pinky_streamlit import update_data, insert_data, invalidate_crud
from snowflake.snowpark.functions import current_timestamp, lit

update_data

Updates existing rows via DataFrame.update() conditioned on a key column. / Met à jour des lignes existantes via DataFrame.update() conditionné par une colonne clé.

update_data(
    session=session,
    data=changes,                  # list[dict] — must contain key column
    key="ID",                      # unique identifier column
    table="MY_TABLE",
    metadata={
        "UPDATED_AT": current_timestamp(),
        "UPDATED_BY": lit(login),
    },
    on_success=lambda: invalidate_crud("my_key"),
)

insert_data

Inserts rows via DataFrame.merge() (MERGE / INSERT IF NOT MATCHED). Deduplicates on the key column. / Insère des lignes via DataFrame.merge() (MERGE / INSERT IF NOT MATCHED). Déduplique sur la colonne clé.

insert_data(
    session=session,
    data=[row_dict],
    key="CODE",
    table="MY_TABLE",
    metadata={
        "CREATED_AT": current_timestamp(),
        "CREATED_BY": lit(login),
    },
    on_success=lambda: invalidate_crud("my_key"),
)

Audit metadata columns / Colonnes de métadonnées d'audit

Pass metadata at call site — never hardcode in the function. / Passer les métadonnées à l'appel — ne jamais les coder en dur dans la fonction.

Column	Value	Operation
CREATED_AT	current_timestamp()	insert
CREATED_BY	lit(login)	insert
UPDATED_AT	current_timestamp()	update
UPDATED_BY	lit(login)	update

---

add_status_icons

Appends a {COLUMN}_ICON column mapping status values to icons or emojis. / Ajoute une colonne {COLUMN}_ICON qui mappe les valeurs de statut vers des icônes ou emojis.

from pinky_streamlit import add_status_icons

df = add_status_icons(
    df=df,
    column_name="STATUS",
    mapping_status_icons={
        "Active":   ICONS.SUCCESS.emoji,   # ✅
        "Inactive": ICONS.CANCEL.emoji,    # ❌
        "Pending":  ICONS.WARNING.emoji,   # ⚠️
    },
    default_icon=ICONS.UNDEFINED.emoji,    # —
)
# df now has a STATUS_ICON column

Material icons (:material/xxx:) are not supported in st.dataframe / st.data_editor. Use .emoji for table columns. / Les Material icons (:material/xxx:) ne sont pas supportées dans st.dataframe / st.data_editor. Utiliser .emoji pour les colonnes de tableau.

---

Skip specific loaders in the progress bar / Sauter des loaders spécifiques dans la barre de progression

# Use show_progress=False when all data is already in session_state
# Utiliser show_progress=False quand toutes les données sont déjà dans session_state
messages, main = setup_page(session, show_progress=False)

---

Telemetry

setup_page() emits a page_loaded Snowflake telemetry event automatically (if the snowflake.telemetry package is available). Attributes include loader count, per-loader duration, and total duration. / setup_page() émet automatiquement un événement de télémétrie Snowflake page_loaded (si le package snowflake.telemetry est disponible). Les attributs incluent le nombre de loaders, la durée par loader et la durée totale.

No configuration needed — telemetry failures are silently swallowed. / Aucune configuration nécessaire — les échecs de télémétrie sont silencieusement ignorés.