Recettes SDK

En bref

Ces recettes couvrent les tâches courantes d'une intégration : enregistrer un entrepôt et ses datasets, indexer des documents, créer des KPIs et des tableaux de bord, et tout tenir à jour depuis un pipeline de données. Chaque recette utilise upsert : elle peut tourner autant de fois que nécessaire sans créer de doublons. Elles supposent un client configuré : voir SDK Python.

Ordre de provisionnement

Les ressources dépendent les unes des autres. Créez-les dans cet ordre :

1. Système de stockage : l'entrepôt ou l'espace de fichiers.
2. Catégories de sources et sources (facultatif) : l'origine des données, pour le lineage.
3. Datasets : les tables et les vues, sur un système de stockage de type entrepôt.
4. Accès des équipes à chaque dataset, puis analyse.
5. Collections de documents sur un système de stockage de fichiers, puis ingestion.
6. KPIs, figures et tableaux de bord sur les datasets, avec l'accès de leurs équipes.

Un entrepôt et ses datasets

import os
from northstar_sdk import Client

client = Client()

client.storage_systems.upsert(
    slug="warehouse",
    name="Analytics warehouse",
    kind="POSTGRESQL",
    credentials={
        "host": "db.internal",
        "user": "integration",
        "password": os.environ["WAREHOUSE_PASSWORD"],
        "database": "analytics",
    },
)

client.data_sources.upsert(
    slug="hubspot",
    name="HubSpot",
    category=client.data_source_categories.upsert(slug="crm", name="CRM", color="#3366FF"),
    logo_url="https://example.com/hubspot.svg",
)

orders = client.datasets.upsert(
    location={"schema": "marts", "table": "orders"},
    name="Orders",
    type="TABLE",
    layer="GOLD",
    storage_system_slug="warehouse",
    data_source_slugs=["hubspot"],
    description="One row per order, deduplicated and converted to euros.",
)
orders.assign_to_team("sales")
orders.analyze()

• Les identifiants dépendent du type de stockage : voir la référence. Lisez les secrets dans l'environnement, jamais dans le code.
• storage_system_slug vaut "postgresql" par défaut : passez toujours le slug du système que vous avez créé.
• layer décide des fonctions IA qui s'appliquent : passez GOLD ou SILVER pour les datasets préparés. Sans couche, le dataset compte comme du bronze. Voir les fonctions IA selon la couche.
• assign_to_team juste après l'upsert : jusque-là, seuls les administrateurs de la plateforme voient le dataset.
• analyze() lance l'analyse complète en arrière-plan : règles de validation, profilage des colonnes, score de qualité, alertes précoces, résumé IA pour les datasets silver et gold, puis les alertes du dataset. Le résultat apparaît dans l'onglet Qualité du dataset. run_validations() et check_alerts() lancent une seule étape.
• generate_metadata() demande à l'IA de décrire les colonnes sans description. Les descriptions attendent qu'une personne les accepte.
• Une source demande une URL de logo la première fois qu'elle est créée.

Collections de documents

import json

with open("service-account.json") as fh:
    service_account = json.load(fh)

client.storage_systems.upsert(
    slug="legal-drive",
    name="Legal Drive",
    kind="GOOGLE_DRIVE",
    credentials=service_account,  # les champs du fichier, lus localement
)

contracts = client.document_collections.upsert(
    name="Contracts",
    storage_system_slug="legal-drive",
    location={"folderId": "1AbC…"},  # l'identifiant du dossier, dans l'URL Drive
)
contracts.ingest()

• Partagez le dossier avec l'e-mail du compte de service avant l'ingestion, sinon le dossier paraît vide.
• L'accès des équipes à une collection se donne dans l'application, dans l'onglet Ressources de l'équipe : le SDK ne rattache pas les collections aux équipes.
• ingest() lit le dossier et indexe ses fichiers PDF en arrière-plan. Il rend la main aussitôt ; le relancer pendant une synchronisation n'en démarre pas une seconde.
• Relancez ingest() quand des fichiers changent : les fichiers inchangés ne sont pas réindexés.
• Les clés de location dépendent du type de stockage (Drive, Amazon S3, OneDrive, SharePoint) : voir la référence.
• deindex() retire la collection du Chat sans la supprimer. Voir les collections de documents.

KPIs et tableaux de bord

revenue = client.kpis.upsert(
    name="Monthly revenue",
    dataset_id=orders.dataset_id,
    definition_type="MANUAL",
    manual_config={"valueField": "amount_eur", "aggregation": "SUM"},
)
revenue.assign_to_team("sales")

dashboard = client.dashboards.upsert(
    title="Sales review",
    icon="layout-dashboard",
    embed_url="https://bi.example.com/embed/sales-review",
    datasets=["Orders"],
)
dashboard.assign_to_team("sales")
dashboard.assign_owner("head.of.sales@example.com")

• Un KPI lit un dataset. MANUAL prend une colonne et une agrégation dans manual_config ; SQL prend une requête qui renvoie une valeur dans sql_config={"query": "…"}. Voir les KPIs.
• Un tableau de bord enregistré par le SDK est intégré depuis un autre outil : embed_url doit être une adresse http(s) absolue, unique dans l'organisation. upsert retrouve un tableau de bord existant par cette URL.
• datasets accepte des noms ou des identifiants de datasets : ils documentent le tableau de bord et le placent dans le lineage.
• client.dashboards.create_folder(name=…, color=…) crée un dossier ; passez son identifiant en folder_id.

Depuis un pipeline de données

Appelez le SDK à la fin de chaque exécution du pipeline, une fois les tables chargées :

from northstar_sdk import Client, northstar_flow

client = Client()

@northstar_flow(client)
def nightly_load():
    load_tables()                                   # votre propre code
    orders = client.datasets.upsert(
        location={"schema": "marts", "table": "orders"},
        name="Orders",
        type="TABLE",
        layer="GOLD",
        storage_system_slug="warehouse",
        prefect_deployment_id=CURRENT_DEPLOYMENT_ID,  # le traitement planifié qui le rafraîchit
    )
    orders.analyze()                                # score de qualité et alertes sur les données fraîches

• analyze() après chaque chargement tient le score de qualité à jour et vérifie les alertes sur les nouvelles données.
• prefect_deployment_id relie le dataset au traitement planifié qui le rafraîchit. Les utilisateurs qui voient le dataset suivent alors ce traitement sur la page Workflows.
• @northstar_flow(client) transforme une fonction en flow Prefect qui signale ses exécutions (en cours, terminée, échouée, plantée, annulée) à la page Workflows. Vos propres hooks d'état s'exécutent toujours après ceux de la plateforme.
• Si vos pipelines ne tournent pas sur Prefect, send_workflow_event(client, deployment_id=…, event="COMPLETED", flow_run_id=…) signale une exécution à la main.

Questions fréquentes

Puis-je lancer une recette deux fois ? Oui. upsert retrouve les ressources existantes par leur clé métier et les met à jour.

analyze() a rendu la main aussitôt. A-t-il fonctionné ? Il a lancé l'analyse, qui tourne en arrière-plan. Son résultat apparaît dans l'onglet Qualité du dataset, et les alertes dont la condition est remplie envoient leurs e-mails comme d'habitude.

Mon dataset a été créé mais mon script ne le trouve pas. Rattachez-le à l'une des équipes du compte d'intégration : jusque-là, seuls les administrateurs de la plateforme le voient.

Voir aussi

• SDK Python
• Référence du SDK
• Ce qui se met à jour seul