Skip to content
northstar
Docs menu

SDK recipes

Step-by-step code for the usual jobs: warehouses, datasets, documents, KPIs, dashboards and pipelines.

In short

These recipes cover the usual jobs of an integration: register a warehouse and its datasets, index documents, create KPIs and dashboards, and keep everything current from a data pipeline. Each recipe uses upsert, so it can run again and again without creating duplicates. They assume a configured client: see Python SDK.

Provisioning order

Resources depend on each other. Create them in this order:

  1. Storage system: the warehouse or file store.
  2. Data source categories and data sources (optional): where the data comes from, for lineage.
  3. Datasets: the tables and views, on a warehouse storage system.
  4. Team access on each dataset, then analysis.
  5. Document collections on a file storage system, then ingestion.
  6. KPIs, figures and dashboards on top of the datasets, with their team access.

A warehouse and its datasets

python
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()
  • Credentials depend on the storage kind: see the reference. Read secrets from the environment, never from the code.
  • storage_system_slug defaults to "postgresql": always pass the slug of the system you created.
  • layer decides which AI features apply: pass GOLD or SILVER for curated datasets. Without a layer, the dataset counts as bronze. See AI features by layer.
  • assign_to_team right after the upsert: until then, only platform administrators see the dataset.
  • analyze() starts the full analysis in the background: validation rules, column profiling, quality score, early warnings, AI summary for silver and gold datasets, then the dataset's alerts. The result appears on the dataset's Quality tab. run_validations() and check_alerts() run a single step.
  • generate_metadata() asks the AI to describe the columns that have no description. The descriptions wait for a person to accept them.
  • A data source needs a logo URL the first time it is created.

Document collections

python
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,  # the fields of the file, parsed locally
)

contracts = client.document_collections.upsert(
    name="Contracts",
    storage_system_slug="legal-drive",
    location={"folderId": "1AbC…"},  # the folder id from the Drive URL
)
contracts.ingest()
  • Share the folder with the service account's email before ingesting, or the folder looks empty.
  • Team access to a collection is given in the app, on the team's Resources tab: the SDK does not assign collections to teams.
  • ingest() reads the folder and indexes its PDF files in the background. It returns at once; running it again while a sync is in progress does not start a second one.
  • Run ingest() again after files change: unchanged files are not indexed twice.
  • The location keys depend on the storage kind (Drive, Amazon S3, OneDrive, SharePoint): see the reference.
  • deindex() removes the collection from Chat without deleting it. See document collections.

KPIs and dashboards

python
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")
  • A KPI reads one dataset. MANUAL takes a column and an aggregation in manual_config; SQL takes a query that returns one value in sql_config={"query": "…"}. See KPIs.
  • A dashboard registered by the SDK is embedded from another tool: embed_url must be an absolute http(s) address, unique in the organization. upsert finds an existing dashboard by that URL.
  • datasets accepts dataset names or ids: they document the dashboard and place it in the lineage.
  • client.dashboards.create_folder(name=…, color=…) creates a folder; pass its id as folder_id.

From a data pipeline

Call the SDK at the end of each pipeline run, after the tables are loaded:

python
from northstar_sdk import Client, northstar_flow

client = Client()

@northstar_flow(client)
def nightly_load():
    load_tables()                                   # your own 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,  # the scheduled job that refreshes it
    )
    orders.analyze()                                # quality score and alerts on fresh data
  • analyze() after each load keeps the quality score current and checks the alerts on the new data.
  • prefect_deployment_id links the dataset to the scheduled job that refreshes it. Users who can see the dataset then follow that job on the Workflows page.
  • @northstar_flow(client) turns a function into a Prefect flow that reports its runs (running, completed, failed, crashed, cancelled) to the Workflows page. Your own state hooks still run after the platform's.
  • If your pipelines do not run on Prefect, send_workflow_event(client, deployment_id=…, event="COMPLETED", flow_run_id=…) reports a run by hand.

FAQ

Can I run a recipe twice? Yes. upsert finds the existing resources by their business key and updates them.

analyze() returned at once. Did it work? It started the analysis, which runs in the background. Its result appears on the dataset's Quality tab, and alerts whose condition is met send their emails as usual.

My dataset was created but my script cannot find it. Assign it to one of the integration account's teams: until then, only platform administrators see it.

See also